7577ca6dfd
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m45s
CD Pipeline / build-and-deploy (push) Successful in 17m47s
CD Pipeline / post-deploy-checks (push) Successful in 2m49s
AI 技術雷達監控 / ai-technology-watch (push) Successful in 35s
90 lines
2.3 KiB
Python
90 lines
2.3 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_jsonb_bind_contract(db: _CaptureDb) -> None:
|
|
assert len(db.calls) == 1
|
|
statement, params = db.calls[0]
|
|
normalized_sql = " ".join(str(statement).split())
|
|
assert ":details::jsonb" not in normalized_sql
|
|
assert "CAST(:details AS JSONB)" in normalized_sql
|
|
assert "details" in statement._bindparams
|
|
assert "detail" not in statement._bindparams
|
|
assert json.loads(params["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_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_jsonb_bind_contract(db)
|