sakshamandClaude Sonnet 5
18305b545c
Reports: the LLM reliably used company_enrichment for prose fields but inconsistently populated the parallel Finding-list/string-list fields from the same evidence, even with progressively more explicit prompting. Add a code-level backfill (products, recent developments, financial signals, strategic initiatives, regulatory signals, risks/opportunities mirrored from SWOT, unknowns, monitoring recommendations) that only ever fills in what the model left empty, never overwrites what it produced. Enrichment tab: reorder sections (Products/Recent updates before Customers/Competitors) and add a per-section "Refresh" button that re-fetches just one of NinjaPear's six independent per-company endpoints when it came back empty - confirmed live that a data-coverage gap (e.g. Amazon returning no products) is real provider behavior, not a bug. Auth: the first account registered on a deployment with zero existing admins is now auto-promoted to admin, closing the chicken-and-egg gap where the only path to admin access was direct DB access. Self-heals if the last admin ever deletes their account. Also bumps nginx's proxy_read_timeout for api.ciagent.org to cover the enrichment refresh's synchronous funding-endpoint call (up to 5 minutes per NinjaPear's docs). Co-Authored-By: Claude Sonnet 5 <[email protected]>
328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""Company CRUD, monitor configuration, and ownership isolation tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
def _register_and_login(client) -> dict[str, str]:
|
|
email = f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
def _create_company(client, headers, **overrides):
|
|
payload = {
|
|
"name": "Acme Mobility Systems",
|
|
"official_website": "acme-mobility.example.com",
|
|
"monitoring_focus": "EV manufacturing expansion",
|
|
"competitor_names": ["Rival Motors"],
|
|
"alias_names": ["Acme"],
|
|
"frequency_type": "weekly",
|
|
**overrides,
|
|
}
|
|
return client.post("/api/v1/companies", json=payload, headers=headers)
|
|
|
|
|
|
def test_create_company_normalizes_website_and_sets_defaults(client):
|
|
headers = _register_and_login(client)
|
|
resp = _create_company(client, headers)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["official_website"] == "https://acme-mobility.example.com"
|
|
assert body["slug"] == "acme-mobility-systems"
|
|
assert body["status"] == "active"
|
|
assert body["aliases"] == ["Acme"]
|
|
assert body["competitors"] == ["Rival Motors"]
|
|
assert body["monitor_configuration"]["frequency_type"] == "weekly"
|
|
assert body["monitor_configuration"]["enabled"] is True
|
|
assert body["monitor_configuration"]["next_run"] is not None
|
|
|
|
|
|
def test_duplicate_company_names_get_distinct_slugs(client):
|
|
headers = _register_and_login(client)
|
|
first = _create_company(client, headers).json()
|
|
second = _create_company(client, headers).json()
|
|
assert first["slug"] != second["slug"]
|
|
|
|
|
|
def test_duplicate_company_names_get_a_unique_display_name(client):
|
|
headers = _register_and_login(client)
|
|
first = _create_company(client, headers).json()
|
|
second = _create_company(client, headers).json()
|
|
third = _create_company(client, headers).json()
|
|
|
|
assert first["name"] == "Acme Mobility Systems"
|
|
assert second["name"] == "Acme Mobility Systems (2)"
|
|
assert third["name"] == "Acme Mobility Systems (3)"
|
|
|
|
|
|
def test_company_name_uniqueness_is_case_insensitive(client):
|
|
headers = _register_and_login(client)
|
|
first = _create_company(client, headers, name="Stripe").json()
|
|
second = _create_company(client, headers, name="STRIPE").json()
|
|
|
|
assert first["name"] == "Stripe"
|
|
assert second["name"] == "STRIPE (2)"
|
|
|
|
|
|
def test_company_name_uniqueness_is_scoped_per_user(client):
|
|
owner_headers = _register_and_login(client)
|
|
other_headers = _register_and_login(client)
|
|
owner_company = _create_company(client, owner_headers).json()
|
|
other_company = _create_company(client, other_headers).json()
|
|
|
|
# Two different users can each have a company with the exact same name -
|
|
# uniqueness is per-user, not global.
|
|
assert owner_company["name"] == other_company["name"] == "Acme Mobility Systems"
|
|
|
|
|
|
def test_company_list_and_detail_are_scoped_to_owner(client):
|
|
owner_headers = _register_and_login(client)
|
|
other_headers = _register_and_login(client)
|
|
|
|
created = _create_company(client, owner_headers).json()
|
|
|
|
owner_list = client.get("/api/v1/companies", headers=owner_headers).json()
|
|
assert any(c["id"] == created["id"] for c in owner_list)
|
|
|
|
other_list = client.get("/api/v1/companies", headers=other_headers).json()
|
|
assert all(c["id"] != created["id"] for c in other_list)
|
|
|
|
other_detail = client.get(f"/api/v1/companies/{created['id']}", headers=other_headers)
|
|
assert other_detail.status_code == 404
|
|
|
|
owner_detail = client.get(f"/api/v1/companies/{created['id']}", headers=owner_headers)
|
|
assert owner_detail.status_code == 200
|
|
|
|
|
|
def test_pause_and_resume_company_toggles_monitor_enabled(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json()
|
|
|
|
paused = client.post(f"/api/v1/companies/{company['id']}/pause", headers=headers).json()
|
|
assert paused["status"] == "paused"
|
|
assert paused["monitor_configuration"]["enabled"] is False
|
|
|
|
resumed = client.post(f"/api/v1/companies/{company['id']}/resume", headers=headers).json()
|
|
assert resumed["status"] == "active"
|
|
assert resumed["monitor_configuration"]["enabled"] is True
|
|
|
|
|
|
def test_update_company_replaces_aliases_and_competitors(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json()
|
|
|
|
resp = client.patch(
|
|
f"/api/v1/companies/{company['id']}",
|
|
json={"alias_names": ["New Alias"], "competitor_names": []},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["aliases"] == ["New Alias"]
|
|
assert body["competitors"] == []
|
|
|
|
|
|
def test_delete_company_removes_it(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json()
|
|
|
|
resp = client.delete(f"/api/v1/companies/{company['id']}", headers=headers)
|
|
assert resp.status_code == 204
|
|
|
|
resp = client.get(f"/api/v1/companies/{company['id']}", headers=headers)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_monitor_configuration_get_and_patch(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json()
|
|
|
|
resp = client.get(f"/api/v1/companies/{company['id']}/monitor", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["frequency_type"] == "weekly"
|
|
|
|
patch_resp = client.patch(
|
|
f"/api/v1/companies/{company['id']}/monitor",
|
|
json={"frequency_type": "daily"},
|
|
headers=headers,
|
|
)
|
|
assert patch_resp.status_code == 200
|
|
assert patch_resp.json()["frequency_type"] == "daily"
|
|
|
|
|
|
def test_custom_schedule_below_minimum_interval_is_rejected(client, settings):
|
|
headers = _register_and_login(client)
|
|
resp = _create_company(
|
|
client,
|
|
headers,
|
|
name="Too Frequent Co",
|
|
frequency_type="custom",
|
|
interval_minutes=settings.minimum_monitoring_interval_minutes - 1,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_custom_schedule_with_valid_interval_is_accepted(client, settings):
|
|
headers = _register_and_login(client)
|
|
resp = _create_company(
|
|
client,
|
|
headers,
|
|
name="Custom Interval Co",
|
|
frequency_type="custom",
|
|
interval_minutes=settings.minimum_monitoring_interval_minutes + 30,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["monitor_configuration"]["interval_minutes"] == (
|
|
settings.minimum_monitoring_interval_minutes + 30
|
|
)
|
|
|
|
|
|
def test_custom_schedule_requires_interval_or_cron(client):
|
|
headers = _register_and_login(client)
|
|
resp = _create_company(client, headers, name="No Schedule Co", frequency_type="custom")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_company_creation_requires_authentication(client):
|
|
resp = client.post("/api/v1/companies", json={"name": "No Auth Co"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_max_companies_per_user_enforced(client, settings, monkeypatch):
|
|
headers = _register_and_login(client)
|
|
from app.core import config as config_module
|
|
|
|
limited_settings = settings.model_copy(update={"max_companies_per_user": 1})
|
|
from app.main import app
|
|
|
|
app.dependency_overrides[config_module.get_settings] = lambda: limited_settings
|
|
try:
|
|
first = _create_company(client, headers, name="Company One")
|
|
assert first.status_code == 201
|
|
second = _create_company(client, headers, name="Company Two")
|
|
assert second.status_code == 409
|
|
finally:
|
|
app.dependency_overrides.pop(config_module.get_settings, None)
|
|
|
|
|
|
def test_create_company_enqueues_enrichment_when_a_key_is_configured(client, settings):
|
|
"""The real regression guard is every OTHER test in this suite: none of
|
|
them configure NINJAPEAR_API_KEY, so the whole rest of the suite proves
|
|
the enqueue is skipped by default (see the "without a key" test below
|
|
for the direct assertion)."""
|
|
from unittest.mock import patch
|
|
|
|
from app.core import config as config_module
|
|
from app.main import app
|
|
|
|
key_settings = settings.model_copy(update={"ninjapear_api_key": "test-key"})
|
|
app.dependency_overrides[config_module.get_settings] = lambda: key_settings
|
|
try:
|
|
headers = _register_and_login(client)
|
|
with patch("app.tasks.enrichment.enrich_company.delay") as mock_delay:
|
|
resp = _create_company(client, headers, name="Enriched Co")
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
mock_delay.assert_called_once_with(body["id"])
|
|
assert body["enrichment"] == {
|
|
"status": "pending",
|
|
"data": {},
|
|
"errors": {},
|
|
"credits_spent": None,
|
|
"fetched_at": None,
|
|
}
|
|
finally:
|
|
app.dependency_overrides.pop(config_module.get_settings, None)
|
|
|
|
|
|
def test_create_company_does_not_enqueue_enrichment_without_a_key(client):
|
|
from unittest.mock import patch
|
|
|
|
headers = _register_and_login(client)
|
|
with patch("app.tasks.enrichment.enrich_company.delay") as mock_delay:
|
|
resp = _create_company(client, headers, name="Plain Co")
|
|
assert resp.status_code == 201
|
|
mock_delay.assert_not_called()
|
|
assert resp.json()["enrichment"] is None
|
|
|
|
|
|
def test_refresh_enrichment_section_requires_authentication(client):
|
|
resp = client.post(f"/api/v1/companies/{uuid.uuid4()}/enrichment/sections/products/refresh")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_refresh_enrichment_section_rejects_an_unknown_section(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json()
|
|
|
|
resp = client.post(
|
|
f"/api/v1/companies/{company['id']}/enrichment/sections/not_a_section/refresh",
|
|
headers=headers,
|
|
)
|
|
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_refresh_enrichment_section_404s_without_an_enrichment_record(client):
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers).json() # no NINJAPEAR_API_KEY -> no record
|
|
|
|
resp = client.post(
|
|
f"/api/v1/companies/{company['id']}/enrichment/sections/products/refresh",
|
|
headers=headers,
|
|
)
|
|
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_refresh_enrichment_section_is_scoped_to_owner(client):
|
|
owner_headers = _register_and_login(client)
|
|
other_headers = _register_and_login(client)
|
|
company = _create_company(client, owner_headers).json()
|
|
|
|
resp = client.post(
|
|
f"/api/v1/companies/{company['id']}/enrichment/sections/products/refresh",
|
|
headers=other_headers,
|
|
)
|
|
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_enrichment_section_updates_just_that_section(client, db_session, settings):
|
|
from app.enrichment.base import Product
|
|
from app.enrichment.mock import MockEnrichmentProvider
|
|
from app.repositories.company_repository import CompanyRepository
|
|
from app.services.enrichment_service import enrich_company
|
|
|
|
headers = _register_and_login(client)
|
|
company_id = _create_company(client, headers).json()["id"]
|
|
|
|
company = await CompanyRepository(db_session).get_by_id(uuid.UUID(company_id))
|
|
await enrich_company(db_session, settings, MockEnrichmentProvider(), company)
|
|
|
|
class _ProductsProvider(MockEnrichmentProvider):
|
|
async def get_products(self, name, website):
|
|
return [Product(name="Acme Pay", description="Payments product")]
|
|
|
|
with patch("app.api.v1.companies.get_enrichment_provider", return_value=_ProductsProvider()):
|
|
resp = client.post(
|
|
f"/api/v1/companies/{company_id}/enrichment/sections/products/refresh",
|
|
headers=headers,
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
products = resp.json()["enrichment"]["data"]["products"]
|
|
assert products == [{"name": "Acme Pay", "description": "Payments product", "category": None}]
|