Celes Renata
898f89926d
Extends integration test coverage from 108 to 193 tests for the beta gate. New test modules: - test_query_api_extended.py (33 tests): documents, evidence, macro/competitive, ops/admin, agents, analytics - test_registry_write_paths.py (16 tests): write paths, validation, duplicates, competitor/exposure CRUD - test_risk_approval_lifecycle.py (8 tests): evaluation edge cases, full approval lifecycle - test_trading_extended.py (12 tests): config round-trips, decision filtering, override validation - test_cross_service_roundtrip.py (4 tests): cross-service data consistency - test_error_handling.py (12 tests): 404s, 422s, empty states, health checks Seed script extended with watchlists, approvals, lockouts, notifications, ingestion runs, saved queries, and daily risk snapshots.
254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""Integration tests for Symbol Registry write paths and edge cases.
|
|
|
|
Covers duplicate detection, alias/source creation, watchlist CRUD,
|
|
competitor relationship lifecycle, and exposure profile management.
|
|
|
|
Uses the ``registry_client`` and ``seed_ids`` fixtures from conftest.py.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 6: Write Paths and Edge Cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRegistryDuplicateCompany:
|
|
"""POST /companies — duplicate ticker+exchange returns 409."""
|
|
|
|
async def test_duplicate_company_returns_409(self, registry_client, seed_ids):
|
|
"""POST /companies — duplicate ticker+exchange returns 409."""
|
|
payload = {
|
|
"ticker": "AAPL",
|
|
"legal_name": "Apple Inc Duplicate",
|
|
"exchange": "NASDAQ",
|
|
"sector": "Technology",
|
|
"industry": "Consumer Electronics",
|
|
"market_cap_bucket": "mega",
|
|
}
|
|
resp = await registry_client.post("/companies", json=payload)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
class TestRegistryCompanyNotFound:
|
|
"""GET /companies/{id} — 404 for non-existent UUID."""
|
|
|
|
async def test_company_not_found(self, registry_client):
|
|
"""GET /companies/{id} — 404 for non-existent UUID."""
|
|
fake_id = "00000000-0000-4000-ffff-000000000099"
|
|
resp = await registry_client.get(f"/companies/{fake_id}")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestRegistryAliasCreate:
|
|
"""POST /companies/{id}/aliases — create alias returns 201."""
|
|
|
|
async def test_create_alias(self, registry_client, seed_ids):
|
|
"""POST /companies/{id}/aliases — create alias returns 201."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
payload = {"alias": "Apple Computer", "alias_type": "historical"}
|
|
resp = await registry_client.post(f"/companies/{company_id}/aliases", json=payload)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert "id" in data
|
|
assert data["alias"] == "Apple Computer"
|
|
assert data["alias_type"] == "historical"
|
|
|
|
|
|
class TestRegistrySourceCreate:
|
|
"""POST /companies/{id}/sources — create and validate sources."""
|
|
|
|
async def test_create_source(self, registry_client, seed_ids):
|
|
"""POST /companies/{id}/sources — create source returns 201."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
payload = {
|
|
"source_type": "news_api",
|
|
"source_name": "Test News Source",
|
|
"config": {},
|
|
"credibility_score": 0.7,
|
|
}
|
|
resp = await registry_client.post(f"/companies/{company_id}/sources", json=payload)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert "id" in data
|
|
assert data["source_type"] == "news_api"
|
|
assert data["source_name"] == "Test News Source"
|
|
|
|
async def test_create_source_invalid_type(self, registry_client, seed_ids):
|
|
"""POST /companies/{id}/sources — invalid source_type returns 422."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
payload = {
|
|
"source_type": "invalid_type",
|
|
"source_name": "Bad Source",
|
|
}
|
|
resp = await registry_client.post(f"/companies/{company_id}/sources", json=payload)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
class TestRegistryWatchlistCRUD:
|
|
"""Watchlist creation, member management, and duplicate detection."""
|
|
|
|
async def test_create_watchlist(self, registry_client):
|
|
"""POST /watchlists — create watchlist returns 201."""
|
|
payload = {"name": "IntTest Watchlist", "description": "Integration test"}
|
|
resp = await registry_client.post("/watchlists", json=payload)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert "id" in data
|
|
assert data["name"] == "IntTest Watchlist"
|
|
|
|
async def test_add_watchlist_member(self, registry_client, seed_ids):
|
|
"""POST /watchlists/{id}/members/{company_id} — add member returns 201."""
|
|
wl_id = seed_ids["watchlists"]["WL_01"]
|
|
company_id = seed_ids["companies"]["XOM"]
|
|
resp = await registry_client.post(f"/watchlists/{wl_id}/members/{company_id}")
|
|
assert resp.status_code == 201
|
|
|
|
async def test_list_watchlist_members(self, registry_client, seed_ids):
|
|
"""GET /watchlists/{id}/members — members with ticker and legal_name."""
|
|
wl_id = seed_ids["watchlists"]["WL_01"]
|
|
resp = await registry_client.get(f"/watchlists/{wl_id}/members")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 2 # AAPL and MSFT from seed
|
|
for member in data:
|
|
assert "ticker" in member
|
|
assert "legal_name" in member
|
|
|
|
async def test_duplicate_watchlist_returns_409(self, registry_client, seed_ids):
|
|
"""POST /watchlists — duplicate name returns 409."""
|
|
payload = {"name": "Tech Leaders"} # Already seeded
|
|
resp = await registry_client.post("/watchlists", json=payload)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 7: Competitor and Exposure Write Paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRegistryCompetitorCRUD:
|
|
"""Competitor relationship creation, update, and soft-delete."""
|
|
|
|
async def test_create_competitor(self, registry_client, seed_ids):
|
|
"""POST /companies/{id}/competitors — create returns 201."""
|
|
company_id = seed_ids["companies"]["XOM"]
|
|
competitor_id = seed_ids["companies"]["JNJ"]
|
|
payload = {
|
|
"company_b_id": competitor_id,
|
|
"relationship_type": "same_sector",
|
|
"strength": 0.4,
|
|
"bidirectional": True,
|
|
"source": "manual",
|
|
}
|
|
resp = await registry_client.post(f"/companies/{company_id}/competitors", json=payload)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert "id" in data
|
|
assert data["relationship_type"] == "same_sector"
|
|
assert data["strength"] == 0.4
|
|
assert data["bidirectional"] is True
|
|
|
|
async def test_self_referencing_competitor_returns_400(self, registry_client, seed_ids):
|
|
"""POST /companies/{id}/competitors — self-reference returns 400."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
payload = {
|
|
"company_b_id": company_id,
|
|
"relationship_type": "direct_rival",
|
|
"strength": 0.5,
|
|
}
|
|
resp = await registry_client.post(f"/companies/{company_id}/competitors", json=payload)
|
|
assert resp.status_code == 400
|
|
|
|
async def test_update_competitor(self, registry_client, seed_ids):
|
|
"""PUT /companies/{id}/competitors/{rel_id} — update strength."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
competitor_id = seed_ids["companies"]["JNJ"]
|
|
# Create a new relationship to update
|
|
create_payload = {
|
|
"company_b_id": competitor_id,
|
|
"relationship_type": "overlapping_products",
|
|
"strength": 0.3,
|
|
}
|
|
create_resp = await registry_client.post(
|
|
f"/companies/{company_id}/competitors", json=create_payload
|
|
)
|
|
assert create_resp.status_code == 201
|
|
rel_id = create_resp.json()["id"]
|
|
# Update
|
|
update_payload = {
|
|
"company_b_id": competitor_id,
|
|
"relationship_type": "overlapping_products",
|
|
"strength": 0.7,
|
|
"bidirectional": True,
|
|
"source": "manual",
|
|
}
|
|
resp = await registry_client.put(
|
|
f"/companies/{company_id}/competitors/{rel_id}", json=update_payload
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["strength"] == 0.7
|
|
|
|
async def test_soft_delete_competitor(self, registry_client, seed_ids):
|
|
"""DELETE /companies/{id}/competitors/{rel_id} — soft-delete returns 200."""
|
|
company_id = seed_ids["companies"]["AAPL"]
|
|
competitor_id = seed_ids["companies"]["XOM"]
|
|
# Create a relationship to delete
|
|
create_payload = {
|
|
"company_b_id": competitor_id,
|
|
"relationship_type": "supply_chain_adjacent",
|
|
"strength": 0.2,
|
|
}
|
|
create_resp = await registry_client.post(
|
|
f"/companies/{company_id}/competitors", json=create_payload
|
|
)
|
|
assert create_resp.status_code == 201
|
|
rel_id = create_resp.json()["id"]
|
|
# Delete
|
|
resp = await registry_client.delete(f"/companies/{company_id}/competitors/{rel_id}")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestRegistryExposureCRUD:
|
|
"""Exposure profile upsert, history, and not-found handling."""
|
|
|
|
async def test_upsert_exposure_profile(self, registry_client, seed_ids):
|
|
"""PUT /companies/{id}/exposure — upsert with version increment."""
|
|
company_id = seed_ids["companies"]["MSFT"] # MSFT has no exposure profile
|
|
payload = {
|
|
"geographic_revenue_mix": {"North America": 0.50, "Europe": 0.30, "Asia": 0.20},
|
|
"supply_chain_regions": ["North America", "Europe"],
|
|
"key_input_commodities": [],
|
|
"regulatory_jurisdictions": ["US", "EU"],
|
|
"market_position_tier": "global_leader",
|
|
"export_dependency_pct": 0.40,
|
|
"source": "manual",
|
|
"confidence": 0.9,
|
|
}
|
|
resp = await registry_client.put(f"/companies/{company_id}/exposure", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["version"] >= 1
|
|
assert data["market_position_tier"] == "global_leader"
|
|
assert data["geographic_revenue_mix"]["North America"] == 0.50
|
|
|
|
async def test_exposure_history(self, registry_client, seed_ids):
|
|
"""GET /companies/{id}/exposure/history — all versions."""
|
|
company_id = seed_ids["companies"]["AAPL"] # AAPL has a seeded profile
|
|
resp = await registry_client.get(f"/companies/{company_id}/exposure/history")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
|
|
async def test_exposure_not_found(self, registry_client, seed_ids):
|
|
"""GET /companies/{id}/exposure — 404 for company with no profile."""
|
|
company_id = seed_ids["companies"]["JNJ"] # JNJ has no exposure profile
|
|
resp = await registry_client.get(f"/companies/{company_id}/exposure")
|
|
assert resp.status_code == 404
|