saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
255 lines
9.3 KiB
Python
255 lines
9.3 KiB
Python
"""Company CRUD, monitor configuration, and ownership isolation tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
|
|
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
|