Merge remote-tracking branch 'origin/main' into codex/p0-obs-002-20260715
This commit is contained in:
ogt
@@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
@@ -214,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N
|
||||
"matrix_fingerprint": initial["matrix_fingerprint"],
|
||||
"candidate_count": 0,
|
||||
"repository_readback_count": 0,
|
||||
"work_item_verifier_readback_count": 0,
|
||||
"repository_readback_verified": True,
|
||||
"closed": True,
|
||||
},
|
||||
@@ -334,6 +336,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None:
|
||||
automation_run_id="run-1",
|
||||
work_item_id="AIA-P0-006",
|
||||
repository_readback_count=0,
|
||||
work_item_verifier_readback_count=0,
|
||||
created_count=1,
|
||||
existing_count=0,
|
||||
)
|
||||
@@ -344,6 +347,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None:
|
||||
automation_run_id="run-1",
|
||||
work_item_id="AIA-P0-006",
|
||||
repository_readback_count=1,
|
||||
work_item_verifier_readback_count=1,
|
||||
created_count=1,
|
||||
existing_count=0,
|
||||
)
|
||||
@@ -354,6 +358,31 @@ def test_reconciliation_summary_requires_repository_readback() -> None:
|
||||
assert passed["closed"] is True
|
||||
|
||||
|
||||
def test_reconciliation_summary_requires_each_work_item_verifier() -> None:
|
||||
matrix = build_asset_capability_matrix(
|
||||
_single_host_scope(),
|
||||
latest_run={"run_id": "discovery-run-1", "status": "success"},
|
||||
repo_root=_REPO_ROOT,
|
||||
)
|
||||
candidates = build_reconciliation_candidates(matrix)
|
||||
|
||||
payload = build_reconciliation_summary_payload(
|
||||
matrix,
|
||||
candidates,
|
||||
trace_id="trace-1",
|
||||
automation_run_id="run-1",
|
||||
work_item_id="AIA-P0-006",
|
||||
repository_readback_count=1,
|
||||
work_item_verifier_readback_count=0,
|
||||
created_count=1,
|
||||
existing_count=0,
|
||||
)
|
||||
|
||||
assert payload["repository_readback_verified"] is False
|
||||
assert payload["post_verifier"]["passed"] is False
|
||||
assert payload["closed"] is False
|
||||
|
||||
|
||||
def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None:
|
||||
matrix = build_asset_capability_matrix(
|
||||
_single_host_scope(),
|
||||
@@ -380,13 +409,15 @@ def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None:
|
||||
|
||||
def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> None:
|
||||
source = inspect.getsource(_persist_reconciliation)
|
||||
loop_start = source.index("for candidate_raw in candidates:")
|
||||
loop_end = source.index("if insert_rows:")
|
||||
loop_start = source.index("for candidate in candidate_rows:")
|
||||
loop_end = source.index("insert_work_item_sql =")
|
||||
|
||||
assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source
|
||||
assert "insert_rows" in source
|
||||
assert "insert_parameters" in source
|
||||
assert "_WORK_ITEM_INSERT_BATCH_SIZE" in source
|
||||
assert "await db.execute" not in source[loop_start:loop_end]
|
||||
assert "AND output ->> 'closed' = 'true'" in source
|
||||
assert "work_item_verifier_readback_count" in source
|
||||
|
||||
|
||||
def test_reconcile_once_uses_bounded_worker_live_readback_timeout(
|
||||
@@ -474,7 +505,7 @@ def test_reconcile_once_publishes_closed_public_matrix_cache(
|
||||
assert result["public_cache_refreshed"] is True
|
||||
|
||||
|
||||
def test_reconciliation_loop_retries_degraded_result_before_daily_wait(
|
||||
def test_reconciliation_loop_retries_degraded_result_before_poll(
|
||||
monkeypatch: Any,
|
||||
) -> None:
|
||||
triggered_by_values: list[str] = []
|
||||
@@ -495,7 +526,7 @@ def test_reconciliation_loop_retries_degraded_result_before_daily_wait(
|
||||
monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep)
|
||||
monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90)
|
||||
monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800)
|
||||
monkeypatch.setattr(reconciliation_job, "_seconds_until_next_trigger", lambda: 1)
|
||||
monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 1)
|
||||
|
||||
try:
|
||||
asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop())
|
||||
@@ -624,6 +655,136 @@ class _FakeDbContext:
|
||||
return None
|
||||
|
||||
|
||||
class _FakeReconciliationResult:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
scalar: Any = None,
|
||||
rows: list[tuple[Any, ...]] | None = None,
|
||||
) -> None:
|
||||
self._scalar = scalar
|
||||
self._rows = rows or []
|
||||
|
||||
def scalar_one_or_none(self) -> Any:
|
||||
return self._scalar
|
||||
|
||||
def scalar_one(self) -> Any:
|
||||
return self._scalar
|
||||
|
||||
def fetchall(self) -> list[tuple[Any, ...]]:
|
||||
return self._rows
|
||||
|
||||
|
||||
class _FakeReconciliationDb:
|
||||
def __init__(self) -> None:
|
||||
self.persisted_fingerprints: set[str] = set()
|
||||
self.verified_fingerprints: set[str] = set()
|
||||
self.summary_outputs: list[dict[str, Any]] = []
|
||||
self.work_item_batch_sizes: list[int] = []
|
||||
|
||||
async def execute(
|
||||
self,
|
||||
statement: Any,
|
||||
params: dict[str, Any] | list[dict[str, Any]] | None = None,
|
||||
) -> _FakeReconciliationResult:
|
||||
sql = str(statement)
|
||||
if "pg_advisory_xact_lock" in sql:
|
||||
return _FakeReconciliationResult()
|
||||
if "SELECT output" in sql:
|
||||
prior = self.summary_outputs[-1] if self.summary_outputs else None
|
||||
return _FakeReconciliationResult(scalar=prior)
|
||||
if "INSERT INTO automation_operation_log" in sql and isinstance(params, list):
|
||||
self.work_item_batch_sizes.append(len(params))
|
||||
for row in params:
|
||||
payload = json.loads(str(row["input"]))
|
||||
self.persisted_fingerprints.add(payload["candidate_fingerprint"])
|
||||
return _FakeReconciliationResult()
|
||||
if "UPDATE automation_operation_log" in sql:
|
||||
assert isinstance(params, dict)
|
||||
requested = set(params["candidate_fingerprints"])
|
||||
self.verified_fingerprints.update(
|
||||
requested & self.persisted_fingerprints
|
||||
)
|
||||
return _FakeReconciliationResult()
|
||||
if "SELECT DISTINCT" in sql and "candidate_fingerprint" in sql:
|
||||
assert isinstance(params, dict)
|
||||
requested = set(params["candidate_fingerprints"])
|
||||
source = (
|
||||
self.verified_fingerprints
|
||||
if "repository_readback_verified" in sql
|
||||
else self.persisted_fingerprints
|
||||
)
|
||||
return _FakeReconciliationResult(
|
||||
rows=[(fingerprint,) for fingerprint in sorted(requested & source)]
|
||||
)
|
||||
if "RETURNING op_id::text" in sql:
|
||||
assert isinstance(params, dict)
|
||||
self.summary_outputs.append(
|
||||
json.loads(str(params["output"]))
|
||||
)
|
||||
return _FakeReconciliationResult(scalar="summary-op-1")
|
||||
raise AssertionError(f"unexpected SQL: {sql}")
|
||||
|
||||
|
||||
class _FakeReconciliationDbContext:
|
||||
def __init__(self, db: _FakeReconciliationDb) -> None:
|
||||
self.db = db
|
||||
|
||||
async def __aenter__(self) -> _FakeReconciliationDb:
|
||||
return self.db
|
||||
|
||||
async def __aexit__(self, *_args: object) -> None:
|
||||
return None
|
||||
|
||||
|
||||
def test_reconciliation_persists_bounded_batches_and_reuses_verified_summary(
|
||||
monkeypatch: Any,
|
||||
) -> None:
|
||||
db = _FakeReconciliationDb()
|
||||
monkeypatch.setattr(
|
||||
"src.db.base.get_db_context",
|
||||
lambda _project_id: _FakeReconciliationDbContext(db),
|
||||
)
|
||||
matrix = {
|
||||
"matrix_fingerprint": "matrix-fingerprint-1",
|
||||
"latest_discovery_run": {
|
||||
"run_id": "00000000-0000-0000-0000-000000000001"
|
||||
},
|
||||
"controls": {
|
||||
control_id: True
|
||||
for control_id in REQUIRED_CONTROL_IDS[:6]
|
||||
},
|
||||
}
|
||||
candidates = [
|
||||
{
|
||||
"candidate_fingerprint": f"fingerprint-{index:04d}",
|
||||
"candidate_kind": "capability_gap",
|
||||
"canonical_asset_id": f"awoooi:services:asset-{index:04d}",
|
||||
"asset_inventory_id": index,
|
||||
"missing_stage_ids": ["sensor"],
|
||||
"stale_stage_ids": [],
|
||||
"change_types": [],
|
||||
}
|
||||
for index in range(301)
|
||||
]
|
||||
|
||||
first = asyncio.run(
|
||||
_persist_reconciliation(matrix, candidates, project_id="awoooi")
|
||||
)
|
||||
second = asyncio.run(
|
||||
_persist_reconciliation(matrix, candidates, project_id="awoooi")
|
||||
)
|
||||
|
||||
assert db.work_item_batch_sizes == [250, 51]
|
||||
assert first["candidate_count"] == 301
|
||||
assert first["repository_readback_count"] == 301
|
||||
assert first["work_item_verifier_readback_count"] == 301
|
||||
assert first["repository_readback_verified"] is True
|
||||
assert first["closed"] is True
|
||||
assert second["repository_readback_verified"] is True
|
||||
assert len(db.summary_outputs) == 1
|
||||
|
||||
|
||||
def test_live_change_event_readback_is_scoped_to_selected_asset_ids(
|
||||
monkeypatch: Any,
|
||||
) -> None:
|
||||
@@ -647,6 +808,11 @@ def test_live_change_event_readback_is_scoped_to_selected_asset_ids(
|
||||
change_sql, change_params = next(
|
||||
(sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql
|
||||
)
|
||||
inventory_sql = next(
|
||||
sql for sql, _params in db.calls if "FROM asset_inventory" in sql
|
||||
)
|
||||
assert "JOIN asset_coverage_snapshot" in inventory_sql
|
||||
assert "LEFT JOIN asset_coverage_snapshot" not in inventory_sql
|
||||
assert "asset_id = ANY(:asset_ids)" in change_sql
|
||||
assert change_params == {"asset_ids": [42]}
|
||||
assert assets[0]["change_types"] == ["asset_modified"]
|
||||
|
||||
Reference in New Issue
Block a user