Celes Renata
c85c0068a2
- Replace all datetime.utcnow() with datetime.now(tz=timezone.utc) across 8 files - Fix 12 failing tests to match current implementation behavior - Fix pytest_plugins in non-top-level conftest (moved to root conftest.py) - Auto-fix 189 lint issues (import sorting, unused imports) - Add CI/CD pipeline infrastructure (ARC, ArgoCD, Kargo manifests) - Add values-beta.yaml and values-paper.yaml for staged deployments - Update GitHub Actions workflow to use self-hosted-gremlin runners - Add integration-test job to CI pipeline Result: 1596 passed, 0 failed, 0 warnings
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""Tests for the Query API app structure and helper functions."""
|
|
from datetime import datetime, timezone
|
|
|
|
from services.api.app import _parse_jsonb, _row_to_dict, app
|
|
|
|
# --- _parse_jsonb ---
|
|
|
|
def test_parse_jsonb_dict():
|
|
assert _parse_jsonb({"a": 1}) == {"a": 1}
|
|
|
|
|
|
def test_parse_jsonb_list():
|
|
assert _parse_jsonb([1, 2]) == [1, 2]
|
|
|
|
|
|
def test_parse_jsonb_string():
|
|
assert _parse_jsonb('{"x": 1}') == {"x": 1}
|
|
|
|
|
|
def test_parse_jsonb_list_string():
|
|
assert _parse_jsonb('["a", "b"]') == ["a", "b"]
|
|
|
|
|
|
def test_parse_jsonb_none():
|
|
assert _parse_jsonb(None) is None
|
|
|
|
|
|
def test_parse_jsonb_invalid_string():
|
|
assert _parse_jsonb("not json") == "not json"
|
|
|
|
|
|
# --- _row_to_dict ---
|
|
|
|
class FakeRecord(dict):
|
|
"""Mimics asyncpg.Record enough for _row_to_dict."""
|
|
def items(self):
|
|
return super().items()
|
|
|
|
|
|
def test_row_to_dict_converts_datetime():
|
|
dt = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)
|
|
row = FakeRecord({"created_at": dt, "name": "test"})
|
|
result = _row_to_dict(row)
|
|
assert result["created_at"] == dt.isoformat()
|
|
assert result["name"] == "test"
|
|
|
|
|
|
def test_row_to_dict_passes_primitives():
|
|
row = FakeRecord({"count": 42, "active": True, "label": "ok", "val": None})
|
|
result = _row_to_dict(row)
|
|
assert result == {"count": 42, "active": True, "label": "ok", "val": None}
|
|
|
|
|
|
# --- App structure ---
|
|
|
|
def test_app_has_expected_routes():
|
|
paths = [route.path for route in app.routes]
|
|
assert "/health" in paths
|
|
assert "/api/companies" in paths
|
|
assert "/api/companies/{company_id}" in paths
|
|
assert "/api/documents" in paths
|
|
assert "/api/documents/{document_id}" in paths
|
|
assert "/api/trends" in paths
|
|
assert "/api/trends/{trend_id}" in paths
|
|
assert "/api/trends/{trend_id}/evidence" in paths
|
|
assert "/api/recommendations" in paths
|
|
assert "/api/recommendations/{recommendation_id}" in paths
|
|
assert "/api/recommendations/{recommendation_id}/evidence" in paths
|
|
assert "/api/orders" in paths
|
|
assert "/api/orders/{order_id}" in paths
|
|
assert "/api/positions" in paths
|
|
assert "/api/audit/{entity_type}/{entity_id}" in paths
|
|
|
|
|
|
def test_app_has_admin_routes():
|
|
paths = [route.path for route in app.routes]
|
|
# Source health
|
|
assert "/api/admin/sources/health" in paths
|
|
assert "/api/admin/sources/{source_id}/runs" in paths
|
|
assert "/api/admin/sources/{source_id}/toggle" in paths
|
|
assert "/api/admin/sources/{source_id}/credibility" in paths
|
|
# Symbol configs
|
|
assert "/api/admin/companies/{company_id}/toggle" in paths
|
|
assert "/api/admin/companies/{company_id}/sector" in paths
|
|
assert "/api/admin/companies/coverage" in paths
|
|
# Trading mode
|
|
assert "/api/admin/trading/config" in paths
|
|
assert "/api/admin/trading/mode" in paths
|
|
assert "/api/admin/trading/approvals" in paths
|
|
assert "/api/admin/trading/approvals/{approval_id}" in paths
|
|
assert "/api/admin/trading/lockouts" in paths
|
|
# Macro toggle
|
|
assert "/api/admin/macro/status" in paths
|
|
assert "/api/admin/macro/toggle" in paths
|
|
|
|
|
|
def test_app_has_macro_routes():
|
|
paths = [route.path for route in app.routes]
|
|
assert "/api/macro/events" in paths
|
|
assert "/api/macro/events/{event_id}" in paths
|
|
assert "/api/macro/impacts/{ticker}" in paths
|
|
assert "/api/trends/{trend_id}/projection" in paths
|
|
|
|
def test_app_has_ops_dashboard_routes():
|
|
paths = [route.path for route in app.routes]
|
|
assert "/api/ops/ingestion/throughput" in paths
|
|
assert "/api/ops/ingestion/summary" in paths
|
|
assert "/api/ops/model/failures" in paths
|
|
assert "/api/ops/model/performance" in paths
|
|
assert "/api/ops/pipeline/health" in paths
|
|
assert "/api/ops/sources/coverage-gaps" in paths
|