dbed105066
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 6m36s
CD Pipeline / build-and-deploy (push) Successful in 15m16s
CD Pipeline / post-deploy-checks (push) Failing after 10m29s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 30s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Failing after 12m42s
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault(
|
|
"DATABASE_URL",
|
|
"postgresql+asyncpg://test:test@localhost/test",
|
|
)
|
|
|
|
from src.db import base as db_base # noqa: E402
|
|
from src.services import audit_sink, contract_service # noqa: E402
|
|
|
|
|
|
class _CaptureDb:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[Any, dict[str, Any]]] = []
|
|
|
|
async def execute(
|
|
self,
|
|
statement: Any,
|
|
params: dict[str, Any],
|
|
) -> None:
|
|
self.calls.append((statement, params))
|
|
|
|
|
|
def _assert_immutable_audit_jsonb_bind_contract(db: _CaptureDb) -> None:
|
|
assert len(db.calls) == 1
|
|
statement, params = db.calls[0]
|
|
normalized_sql = " ".join(str(statement).split())
|
|
assert "INSERT INTO alert_operation_log" in normalized_sql
|
|
assert "INSERT INTO audit_logs" not in normalized_sql
|
|
assert ":context::jsonb" not in normalized_sql
|
|
assert "CAST(:context AS JSONB)" in normalized_sql
|
|
assert "context" in statement._bindparams
|
|
assert "details" not in statement._bindparams
|
|
context = json.loads(params["context"])
|
|
assert context["schema_version"] == "awooop_sanitized_audit_v1"
|
|
assert context["details"]["result"] == "verified"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_sink_uses_asyncpg_safe_jsonb_bind(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _CaptureDb()
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str):
|
|
assert project_id == "awoooi"
|
|
yield db
|
|
|
|
monkeypatch.setattr(db_base, "get_db_context", fake_db_context)
|
|
|
|
await audit_sink._write_audit_impl(
|
|
project_id="awoooi",
|
|
action="run.approval.approve",
|
|
resource_type="run",
|
|
resource_id="run-1",
|
|
details={"result": "verified"},
|
|
run_id="run-1",
|
|
trace_id="trace-1",
|
|
)
|
|
|
|
_assert_immutable_audit_jsonb_bind_contract(db)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contract_service_uses_asyncpg_safe_jsonb_bind(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _CaptureDb()
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str):
|
|
assert project_id == "awoooi"
|
|
yield db
|
|
|
|
monkeypatch.setattr(db_base, "get_db_context", fake_db_context)
|
|
|
|
await contract_service._write_audit(
|
|
project_id="awoooi",
|
|
action="contract.revision.activated",
|
|
resource_type="contract",
|
|
resource_id="contract-1",
|
|
details={"result": "verified"},
|
|
)
|
|
|
|
_assert_immutable_audit_jsonb_bind_contract(db)
|