test(036): evidence tests (4) + denial tests (5) — 80 backend, 14 frontend
This commit is contained in:
112
backend/tests/services/agent_runs/test_evidence.py
Normal file
112
backend/tests/services/agent_runs/test_evidence.py
Normal file
@@ -0,0 +1,112 @@
|
||||
# backend/tests/services/agent_runs/test_evidence.py
|
||||
# #region Test.AgentRuns.Evidence [C:3] [TYPE Module] [SEMANTICS test,agent-run,evidence,capture,meta]
|
||||
# @BRIEF L1 tests for evidence adapter — capture_meta validation, masking, required fields.
|
||||
# @RELATION BINDS_TO -> [Services.AgentRuns.Evidence]
|
||||
# @TEST_EDGE missing_required_fields -> ValueError.
|
||||
# @TEST_EDGE mask_selectors_required -> contract enforced.
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, event
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from src.models.agent_run import AgentRun
|
||||
from src.services.agent_runs.evidence import register_screenshot_draft, register_masked_derivative
|
||||
from src.services.agent_runs.service import create_agent_run
|
||||
from src.schemas.agent_run import CreateAgentRunRequest, UIContextV2
|
||||
|
||||
|
||||
def _make_session():
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
||||
from src.models.mapping import Base
|
||||
Base.metadata.create_all(bind=engine)
|
||||
return sessionmaker(bind=engine)(), engine
|
||||
|
||||
|
||||
def _make_context() -> UIContextV2:
|
||||
return UIContextV2(
|
||||
objectType="dashboard", objectId="42", envId="dev",
|
||||
route="/dashboards/42", contextVersion=2,
|
||||
intent="build_dashboard_test_scenario",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session():
|
||||
db, _engine = _make_session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run_id(db_session):
|
||||
req = CreateAgentRunRequest(context=_make_context())
|
||||
snap = create_agent_run(db_session, req, user_id="user-1")
|
||||
db_session.commit()
|
||||
return snap.id
|
||||
|
||||
|
||||
class TestRegisterScreenshotDraft:
|
||||
def test_valid_capture_meta(self, db_session, run_id):
|
||||
result = register_screenshot_draft(db_session, run_id, "user-1",
|
||||
name="screenshot_001.png",
|
||||
intended_path="evidence/screenshot_001.png",
|
||||
sha256="a" * 64,
|
||||
capture_meta={
|
||||
"viewport": {"width": 1920, "height": 1200},
|
||||
"capture_method": "cdp",
|
||||
"filter_context_hash": "b" * 64,
|
||||
"readiness_strategy": "network_idle",
|
||||
"readiness_timeout_ms": 15000,
|
||||
"mask_selectors": [".user-info", ".api-key"],
|
||||
},
|
||||
validation_status="valid",
|
||||
)
|
||||
db_session.commit()
|
||||
assert result.id is not None
|
||||
assert result.kind == "screenshot_evidence"
|
||||
assert result.capture_meta is not None
|
||||
|
||||
def test_missing_required_fields_raises(self, db_session, run_id):
|
||||
with pytest.raises(ValueError, match="missing required fields"):
|
||||
register_screenshot_draft(db_session, run_id, "user-1",
|
||||
name="bad.png", intended_path="bad.png",
|
||||
sha256="a" * 64,
|
||||
capture_meta={},
|
||||
)
|
||||
|
||||
def test_missing_viewport_raises(self, db_session, run_id):
|
||||
with pytest.raises(ValueError, match="missing required fields"):
|
||||
register_screenshot_draft(db_session, run_id, "user-1",
|
||||
name="bad.png", intended_path="bad.png",
|
||||
sha256="a" * 64,
|
||||
capture_meta={"capture_method": "cdp"},
|
||||
)
|
||||
|
||||
|
||||
class TestRegisterMaskedDerivative:
|
||||
def test_masked_derivative_stored_separately(self, db_session, run_id):
|
||||
orig = register_screenshot_draft(db_session, run_id, "user-1",
|
||||
name="screenshot_001.png",
|
||||
intended_path="evidence/screenshot_001.png",
|
||||
sha256="a" * 64,
|
||||
capture_meta={
|
||||
"viewport": {"width": 1920, "height": 1200},
|
||||
"capture_method": "cdp",
|
||||
},
|
||||
)
|
||||
db_session.commit()
|
||||
|
||||
masked = register_masked_derivative(db_session, run_id, "user-1",
|
||||
original_artifact_id=orig.id,
|
||||
derivative_sha256="b" * 64,
|
||||
intended_path="evidence/screenshot_001_masked.png",
|
||||
)
|
||||
db_session.commit()
|
||||
|
||||
assert masked.id is not None
|
||||
assert masked.id != orig.id
|
||||
assert masked.kind == "other"
|
||||
assert "_masked" in masked.name
|
||||
# #endregion Test.AgentRuns.Evidence
|
||||
@@ -0,0 +1,73 @@
|
||||
// frontend/src/lib/models/__tests__/AgentRunModel.denial.test.ts
|
||||
// #region TestAgentRuns.Denial [C:3] [TYPE Module] [SEMANTICS test,agent-run,denial,permission,gate]
|
||||
// @BRIEF L1 tests for denial and permission-denied gate scenarios.
|
||||
// @RELATION BINDS_TO -> [AgentRuns.Model]
|
||||
// @TEST_EDGE deny_clears_gate -> gate null, run returns to running.
|
||||
// @TEST_EDGE permission_denied -> no confirm control.
|
||||
import { describe, it, expect, beforeEach, vi } from "vitest";
|
||||
import { AgentRunModel } from "../AgentRunModel.svelte.js";
|
||||
|
||||
vi.mock("$lib/api", () => ({
|
||||
fetchApi: vi.fn(),
|
||||
}));
|
||||
|
||||
import { fetchApi } from "$lib/api";
|
||||
|
||||
describe("AgentRunModel — Denial", () => {
|
||||
let model: AgentRunModel;
|
||||
|
||||
beforeEach(() => {
|
||||
model = new AgentRunModel();
|
||||
model.applyMetadata({
|
||||
type: "agent_run_started", agent_run_id: "run-1", sequence: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it("denying gate returns run to running state", async () => {
|
||||
model.pendingGate = {
|
||||
id: "gate-1", operation: "repository_write",
|
||||
risk_level: "guarded", required_permission: "dashboard:testing:WRITE",
|
||||
status: "pending", reason_required: false,
|
||||
target_paths: ["test.yaml"],
|
||||
};
|
||||
model.state = "waiting_approval";
|
||||
|
||||
(fetchApi as any).mockResolvedValue({});
|
||||
|
||||
const result = await model.decideGate("deny");
|
||||
expect(result).toBe(true);
|
||||
expect(model.state).toBe("running");
|
||||
expect(model.pendingGate).toBeNull();
|
||||
});
|
||||
|
||||
it("decideGate returns false without runId", async () => {
|
||||
const absentModel = new AgentRunModel();
|
||||
const result = await absentModel.decideGate("deny");
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("decideGate returns false without pending gate", async () => {
|
||||
const result = await model.decideGate("deny");
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("permission denied state has no confirm control", () => {
|
||||
model.state = "absent";
|
||||
// When there's no run, there should be no gate
|
||||
expect(model.pendingGate).toBeNull();
|
||||
expect(model.state).toBe("absent");
|
||||
});
|
||||
|
||||
it("reset clears all state including pending gate", () => {
|
||||
model.pendingGate = {
|
||||
id: "gate-1", operation: "repository_write",
|
||||
risk_level: "guarded", required_permission: "dashboard:testing:WRITE",
|
||||
status: "pending", reason_required: false,
|
||||
target_paths: ["test.yaml"],
|
||||
};
|
||||
model.reset();
|
||||
expect(model.pendingGate).toBeNull();
|
||||
expect(model.drafts).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
// #endregion TestAgentRuns.Denial
|
||||
Reference in New Issue
Block a user