f1065a4aaa
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 3m2s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
|
|
|
|
def _read(relative_path: str) -> str:
|
|
return (REPO_ROOT / relative_path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_runs_page_uses_single_flight_completion_based_polling() -> None:
|
|
source = _read("apps/web/src/app/[locale]/awooop/runs/page.tsx")
|
|
|
|
assert "const AUTO_REFRESH_INTERVAL = 90_000" in source
|
|
assert "fetchRunsInFlightRef.current" in source
|
|
assert "completed ? AUTO_REFRESH_INTERVAL : BUSY_RETRY_INTERVAL" in source
|
|
assert "setInterval(" not in source
|
|
assert 'qualityParams.set("limit", "8")' in source
|
|
|
|
|
|
def test_runs_page_keeps_last_good_readback_on_transient_failures() -> None:
|
|
source = _read("apps/web/src/app/[locale]/awooop/runs/page.tsx")
|
|
|
|
assert "setAiRouteStatus(null)" not in source
|
|
assert "setDossierCoverage(null)" not in source
|
|
assert "setEventRecurrence(null)" not in source
|
|
assert "setAutomationQualitySummary(null)" not in source
|
|
|
|
|
|
def test_autonomous_receipt_panel_serializes_bounded_readbacks() -> None:
|
|
source = _read(
|
|
"apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx"
|
|
)
|
|
|
|
assert "const AUTO_REFRESH_INTERVAL = 90_000" in source
|
|
assert "refreshInFlightRef.current" in source
|
|
assert "window.setTimeout(() => void poll(), AUTO_REFRESH_INTERVAL)" in source
|
|
assert "window.setInterval(" not in source
|
|
assert "if (priority) setPriorityPayload(priority);" in source
|
|
assert "if (consumer) setConsumerPayload(consumer);" in source
|
|
assert (
|
|
"if (telegramVerifier) setTelegramVerifierPayload(telegramVerifier);" in source
|
|
)
|
|
assert (
|
|
"if (telegramCoverage) setTelegramCoveragePayload(telegramCoverage);" in source
|
|
)
|
|
|
|
|
|
def test_quality_summary_cache_outlives_one_ui_poll_cycle() -> None:
|
|
source = _read("apps/api/src/services/awooop_truth_chain_service.py")
|
|
|
|
assert 'os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "120")' in source
|
|
|
|
|
|
def test_priority_work_order_defaults_to_source_backed_readback() -> None:
|
|
source = _read("apps/api/src/api/v1/agents.py")
|
|
|
|
assert "live_overlays: bool = False" in source
|
|
assert "if not live_overlays:" in source
|
|
assert '"delegated_to_dedicated_readbacks"' in source
|
|
assert '"dedicated_runtime_readbacks"' in source
|
|
|
|
|
|
def test_consumer_readback_caches_outlive_one_ui_poll_cycle() -> None:
|
|
consumer_source = _read(
|
|
"apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py"
|
|
)
|
|
runtime_source = _read(
|
|
"apps/api/src/services/ai_agent_autonomous_runtime_control.py"
|
|
)
|
|
|
|
assert "_CONSUMER_READBACK_CACHE_TTL_SECONDS = 120.0" in consumer_source
|
|
assert (
|
|
"_LOG_CONTROLLED_WRITEBACK_CONSUMER_READBACK_CACHE_TTL_SECONDS = 120.0"
|
|
in runtime_source
|
|
)
|