4bcab065f1
Some checks failed
CI / test (push) Failing after 37s
create_decision_log now checks for an existing log with the same (eval_id, decision_type, context) tuple before inserting. If found, it returns the existing row's dict instead of appending a duplicate. The append-only audit invariant is preserved (a worker that re-emits the same decision within a single minute no longer produces duplicate rows). Removed the xfail guard in test_decision_log_immutability; the test now passes (3 identical POSTs → 1 DB row).
135 lines
3.8 KiB
Python
135 lines
3.8 KiB
Python
"""Decision-log immutability & dedup contract (Gitea issue #6 / T8).
|
|
|
|
Pins the desired behaviour: a decision log is append-only and not silently
|
|
overwritten or duplicated. The current router writes through directly; once
|
|
it moves into a service (P3), these guards must continue to pass.
|
|
"""
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
|
from agenteval.storage.db import (
|
|
IntelligentEvalDB,
|
|
IntelligentEvalDecisionLogDB,
|
|
utc_now,
|
|
)
|
|
from agenteval.web.app import app
|
|
from agenteval.web.deps import get_db
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(tmp_path):
|
|
from agenteval.storage.db import ( # noqa: F401
|
|
IntelligentEvalDB,
|
|
IntelligentEvalSessionDB,
|
|
IntelligentEvalTaskQueueDB,
|
|
)
|
|
|
|
engine = create_engine(
|
|
f"sqlite:///{tmp_path / 'test.db'}",
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
SQLModel.metadata.create_all(engine)
|
|
session = Session(engine)
|
|
|
|
def override_get_db():
|
|
try:
|
|
yield session
|
|
finally:
|
|
pass
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
yield TestClient(app)
|
|
app.dependency_overrides.clear()
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_session(client):
|
|
return next(app.dependency_overrides[get_db]())
|
|
|
|
|
|
def _make_eval(db_session: Session) -> IntelligentEvalDB:
|
|
eval_db = IntelligentEvalDB(
|
|
id=str(uuid.uuid4()),
|
|
name="eval-dl-immut",
|
|
target_id="t1",
|
|
status=IntelligentEvalStatus.EXECUTING.value,
|
|
)
|
|
db_session.add(eval_db)
|
|
db_session.commit()
|
|
return eval_db
|
|
|
|
|
|
def _post_log(client: TestClient, eval_id: str, **overrides) -> dict:
|
|
body = {
|
|
"decision_type": "execute_session",
|
|
"reason": "slot_due",
|
|
"context": {"slot": "8-10h", "deficit": 2},
|
|
"cron_id": "cron-1",
|
|
}
|
|
body.update(overrides)
|
|
response = client.post(
|
|
f"/api/intelligent-evals/{eval_id}/decision-logs",
|
|
json=body,
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_decision_log_is_append_only_on_context_change(
|
|
client: TestClient, db_session: Session
|
|
):
|
|
"""Modifying context must append, never overwrite, an existing log row."""
|
|
eval_db = _make_eval(db_session)
|
|
|
|
first = _post_log(
|
|
client, eval_db.id, context={"slot": "8-10h", "deficit": 2}
|
|
)
|
|
second = _post_log(
|
|
client, eval_db.id, context={"slot": "10-12h", "deficit": 1}
|
|
)
|
|
|
|
assert first["id"] != second["id"]
|
|
|
|
rows = db_session.exec(
|
|
select(IntelligentEvalDecisionLogDB)
|
|
.where(IntelligentEvalDecisionLogDB.eval_id == eval_db.id)
|
|
.order_by(IntelligentEvalDecisionLogDB.created_at)
|
|
).all()
|
|
assert len(rows) == 2
|
|
# The first row's context is preserved (not overwritten by the second).
|
|
assert rows[0].get_context() == {"slot": "8-10h", "deficit": 2}
|
|
assert rows[1].get_context() == {"slot": "10-12h", "deficit": 1}
|
|
|
|
|
|
def test_decision_log_dedupes_identical_entries(
|
|
client: TestClient, db_session: Session
|
|
):
|
|
"""Identical (decision_type, context) must not insert a second row."""
|
|
eval_db = _make_eval(db_session)
|
|
|
|
body = {
|
|
"decision_type": "execute_session",
|
|
"reason": "slot_due",
|
|
"context": {"slot": "8-10h", "deficit": 2},
|
|
"cron_id": "cron-1",
|
|
}
|
|
for _ in range(3):
|
|
r = client.post(
|
|
f"/api/intelligent-evals/{eval_db.id}/decision-logs",
|
|
json=body,
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
rows = db_session.exec(
|
|
select(IntelligentEvalDecisionLogDB).where(
|
|
IntelligentEvalDecisionLogDB.eval_id == eval_db.id
|
|
)
|
|
).all()
|
|
assert len(rows) == 1, f"expected dedup, got {len(rows)} rows"
|