test(036): repository (11 tests) + agent context v2 (9 tests) + tool filter (6 tests)

agent/tests/test_agent/test_agent_context_v2.py

@@ -0,0 +1,86 @@
+# agent/tests/test_agent/test_agent_context_v2.py
+# #region TestAgent.ContextV2 [C:2] [TYPE Module] [SEMANTICS test,agent,context,v2]
+# @BRIEF Tests for UIContext v2 validation — backward compat, scenario intent, invalid combos.
+# @RELATION BINDS_TO -> [AgentChat.Context]
+# @TEST_EDGE v1_ordinary -> passes unchanged.
+# @TEST_EDGE v2_scenario -> passes with intent.
+# @TEST_EDGE v1_with_intent -> fails.
+# @TEST_EDGE v2_with_wrong_objectType -> fails.
+import pytest
+from ss_tools.agent._context import validate_uicontext, UIContextValidationError
+
+
+class TestUIContextV1BackwardCompat:
+    """v1 contexts must still pass unchanged."""
+
+    def test_v1_dashboard(self):
+        ctx = validate_uicontext({
+            "contextVersion": 1, "objectType": "dashboard",
+            "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+        })
+        assert ctx["contextVersion"] == 1
+
+    def test_v1_dataset(self):
+        ctx = validate_uicontext({
+            "contextVersion": 1, "objectType": "dataset",
+            "objectId": "1", "envId": "dev", "route": "/datasets/1",
+        })
+        assert ctx["objectType"] == "dataset"
+
+    def test_v1_preserves_extra_fields(self):
+        ctx = validate_uicontext({
+            "contextVersion": 1, "objectType": "dashboard",
+            "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+            "customField": "preserved",
+        })
+        assert ctx["customField"] == "preserved"
+
+
+class TestUIContextV2Scenario:
+    """v2 scenario context requires intent and objectType=dashboard."""
+
+    def test_v2_scenario_passes(self):
+        ctx = validate_uicontext({
+            "contextVersion": 2, "objectType": "dashboard",
+            "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+            "intent": "build_dashboard_test_scenario",
+        })
+        assert ctx["intent"] == "build_dashboard_test_scenario"
+
+    def test_v2_without_intent_passes(self):
+        """v2 without intent is valid — just not a scenario."""
+        ctx = validate_uicontext({
+            "contextVersion": 2, "objectType": "dashboard",
+            "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+        })
+        assert ctx["contextVersion"] == 2
+
+    def test_v2_dataset_rejected(self):
+        with pytest.raises(UIContextValidationError, match="objectType must be 'dashboard'"):
+            validate_uicontext({
+                "contextVersion": 2, "objectType": "dataset",
+                "objectId": "1", "envId": "dev", "route": "/datasets/1",
+            })
+
+    def test_v2_unknown_intent_rejected(self):
+        with pytest.raises(UIContextValidationError, match="unsupported intent"):
+            validate_uicontext({
+                "contextVersion": 2, "objectType": "dashboard",
+                "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+                "intent": "run_arbitrary_sql",
+            })
+
+    def test_v3_version_rejected(self):
+        with pytest.raises(UIContextValidationError, match="unsupported contextVersion"):
+            validate_uicontext({
+                "contextVersion": 3, "objectType": "dashboard",
+                "objectId": "42", "envId": "dev", "route": "/dashboards/42",
+            })
+
+    def test_missing_version_rejected(self):
+        with pytest.raises(UIContextValidationError, match="contextVersion is required"):
+            validate_uicontext({
+                "objectType": "dashboard", "objectId": "42",
+                "envId": "dev", "route": "/dashboards/42",
+            })
+# #endregion TestAgent.ContextV2

agent/tests/test_agent/test_scenario_tool_filter.py

@@ -0,0 +1,75 @@
+# agent/tests/test_agent/test_scenario_tool_filter.py
+# #region TestAgent.ScenarioToolFilter [C:2] [TYPE Module] [SEMANTICS test,agent,tool,filter,scenario]
+# @BRIEF Tests for scenario allowlist — excludes SQL tools, preserves mandatory tools.
+# @RELATION BINDS_TO -> [AgentChat.ToolFilter]
+# @TEST_EDGE superset_execute_sql -> excluded in scenario mode.
+# @TEST_EDGE show_capabilities -> always passes.
+# @TEST_EDGE normal_mode -> SQL allowed via dataset affinity.
+from collections import namedtuple
+import pytest
+from ss_tools.agent._tool_filter import build_tool_pipeline, _SCENARIO_TOOL_ALLOWLIST
+
+Tool = namedtuple("Tool", ["name"])
+
+
+SCENARIO_SAFE = [
+    "show_capabilities", "search_dashboards", "get_health_summary",
+    "superset_list_databases", "superset_explore_database",
+    "get_task_status", "list_environments",
+    "create_branch", "commit_changes", "deploy_dashboard",
+    "run_llm_validation", "run_llm_documentation",
+]
+SQL_TOOLS = [
+    "superset_execute_sql", "superset_format_sql", "superset_create_dataset",
+]
+
+
+def _names(result):
+    return [t.name for t in result]
+
+
+class TestScenarioAllowlist:
+    def test_all_scenario_safe_tools_pass(self):
+        tools = [Tool(n) for n in SCENARIO_SAFE + SQL_TOOLS]
+        result = build_tool_pipeline(tools, "admin", "dashboard", "build_dashboard_test_scenario")
+        names = _names(result)
+        for safe in SCENARIO_SAFE:
+            assert safe in names, f"{safe} should pass allowlist"
+
+    def test_sql_tools_blocked_in_scenario(self):
+        tools = [Tool(n) for n in SQL_TOOLS + ["show_capabilities"]]
+        result = build_tool_pipeline(tools, "admin", "dashboard", "build_dashboard_test_scenario")
+        names = _names(result)
+        for sql in SQL_TOOLS:
+            assert sql not in names, f"{sql} must be blocked in scenario mode"
+
+    def test_superset_execute_sql_excluded(self):
+        assert "superset_execute_sql" not in _SCENARIO_TOOL_ALLOWLIST
+        assert "superset_format_sql" not in _SCENARIO_TOOL_ALLOWLIST
+
+    def test_mandatory_tools_always_pass(self):
+        tools = [Tool("show_capabilities")]
+        result = build_tool_pipeline(tools, "user", None, "build_dashboard_test_scenario")
+        assert _names(result) == ["show_capabilities"]
+
+
+class TestNormalMode:
+    """SQL tools are allowed outside scenario mode."""
+
+    def test_sql_allowed_in_dataset_context(self):
+        tools = [Tool(n) for n in ["show_capabilities", "superset_execute_sql", "superset_list_databases"]]
+        result = build_tool_pipeline(tools, "admin", "dataset")
+        names = _names(result)
+        assert "superset_execute_sql" in names, "SQL allowed in normal dataset mode"
+
+
+class TestRBAC:
+    """RBAC runs first, before allowlist."""
+
+    def test_admin_tools_blocked_by_rbac(self):
+        tools = [Tool("deploy_dashboard"), Tool("show_capabilities")]
+        result = build_tool_pipeline(tools, "user", "dashboard")
+        names = _names(result)
+        assert "deploy_dashboard" not in names, "deploy_dashboard requires admin role"
+        assert "show_capabilities" in names
+# #endregion TestAgent.ScenarioToolFilter

backend/tests/services/agent_runs/test_repository.py

@@ -0,0 +1,205 @@
+# backend/tests/services/agent_runs/test_repository.py
+# #region Test.AgentRuns.Repository [C:3] [TYPE Module] [SEMANTICS test,agent-run,repository,crud]
+# @BRIEF L1 unit tests for AgentRunRepository — ownership, terminal immutability, sequence uniqueness.
+# @RELATION BINDS_TO -> [Services.AgentRuns.Repository]
+# @TEST_EDGE foreign_run_access -> None returned.
+# @TEST_EDGE terminal_immutability -> is_terminal returns True.
+# @TEST_EDGE same_sequence_same_hash -> existing event returned.
+# @TEST_EDGE same_sequence_different_hash -> ValueError raised.
+import pytest
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from src.models.agent_run import AgentRun, AgentRunEvent, DraftArtifact, ApprovalGate
+from src.services.agent_runs.repository import AgentRunRepository
+
+
+def _make_session():
+    """Create a per-test SQLite in-memory session with FK enforcement."""
+    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
+
+
+@pytest.fixture
+def repo():
+    db, _engine = _make_session()
+    try:
+        yield AgentRunRepository(db)
+    finally:
+        db.close()
+
+
+def _make_run(user_id: str = "user-1", dashboard_id: str = "42") -> AgentRun:
+    return AgentRun(
+        id=None, user_id=user_id, dashboard_id=dashboard_id,
+        environment_id="dev", context_snapshot={}, status="CREATED",
+    )
+
+
+class TestRunCRUD:
+    def test_create_and_get(self, repo):
+        run = _make_run()
+        repo.create(run)
+        assert run.id is not None
+
+        found = repo.get(run.id, "user-1")
+        assert found is not None
+        assert found.id == run.id
+
+    def test_foreign_owner_returns_none(self, repo):
+        run = _make_run(user_id="user-1")
+        repo.create(run)
+
+        found = repo.get(run.id, "user-2")
+        assert found is None
+
+    def test_terminal_detection(self, repo):
+        run = _make_run()
+        repo.create(run)
+        assert not repo.is_terminal(run)
+
+        run.status = "COMPLETED"
+        assert repo.is_terminal(run)
+
+
+class TestEventSequence:
+    def test_append_event(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        evt = AgentRunEvent(
+            id=None, run_id=run.id, sequence=1,
+            event_type="progress", stage="context", status="active",
+            payload_hash="abc123",
+        )
+        result = repo.append_event(run.id, evt)
+        repo.db.flush()
+        assert result.id is not None
+
+        events = repo.get_events(run.id)
+        assert len(events) == 1
+        assert events[0].sequence == 1
+
+    def test_same_sequence_same_hash_idempotent(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        evt1 = AgentRunEvent(
+            id=None, run_id=run.id, sequence=1,
+            event_type="progress", stage="context", status="active",
+            payload_hash="abc123",
+        )
+        repo.append_event(run.id, evt1)
+
+        evt2 = AgentRunEvent(
+            id=None, run_id=run.id, sequence=1,
+            event_type="progress", stage="context", status="active",
+            payload_hash="abc123",
+        )
+        result = repo.append_event(run.id, evt2)
+        assert result.id == evt1.id  # Same event returned
+
+    def test_same_sequence_different_hash_raises(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        evt1 = AgentRunEvent(
+            id=None, run_id=run.id, sequence=1,
+            event_type="progress", stage="context", status="active",
+            payload_hash="abc123",
+        )
+        repo.append_event(run.id, evt1)
+
+        evt2 = AgentRunEvent(
+            id=None, run_id=run.id, sequence=1,
+            event_type="progress", stage="context", status="active",
+            payload_hash="different",
+        )
+        with pytest.raises(ValueError, match="exists with different payload"):
+            repo.append_event(run.id, evt2)
+
+    def test_get_events_after_sequence(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        for seq in range(1, 5):
+            repo.append_event(run.id, AgentRunEvent(
+                id=None, run_id=run.id, sequence=seq,
+                event_type="progress", payload_hash=f"hash{seq}",
+            ))
+
+        after = repo.get_events(run.id, after_sequence=2)
+        assert len(after) == 2
+        assert after[0].sequence == 3
+
+
+class TestDrafts:
+    def test_register_and_get(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        draft = DraftArtifact(
+            id=None, run_id=run.id, kind="scenario", name="test.yaml",
+            intended_path="test.yaml", content_ref="draft:ref:abc",
+            sha256="a" * 64, validation_status="pending",
+        )
+        repo.register_draft(draft)
+        assert draft.id is not None
+
+        drafts = repo.get_drafts(run.id)
+        assert len(drafts) == 1
+
+    def test_get_draft_by_id(self, repo):
+        run = _make_run()
+        repo.create(run)
+
+        draft = DraftArtifact(
+            id=None, run_id=run.id, kind="scenario", name="test.yaml",
+            intended_path="test.yaml", content_ref="draft:ref:abc",
+            sha256="a" * 64,
+        )
+        repo.register_draft(draft)
+
+        found = repo.get_draft(draft.id, run.id)
+        assert found is not None
+        assert found.name == "test.yaml"
+
+
+class TestApprovalGates:
+    def test_create_and_get_pending(self, repo):
+        run = _make_run()
+        repo.create(run)
+        from datetime import datetime, timezone, timedelta
+
+        gate = ApprovalGate(
+            id=None, run_id=run.id, operation="repository_write",
+            request_hash="x" * 64, target_paths=["test.yaml"],
+            required_permission="dashboard:testing:WRITE",
+            expires_at=datetime.now(timezone.utc) + timedelta(seconds=300),
+        )
+        repo.create_gate(gate)
+        assert gate.id is not None
+
+        pending = repo.get_pending_gate(run.id)
+        assert pending is not None
+        assert pending.status == "pending"
+
+    def test_get_gate_by_id(self, repo):
+        run = _make_run()
+        repo.create(run)
+        from datetime import datetime, timezone, timedelta
+
+        gate = ApprovalGate(
+            id=None, run_id=run.id, operation="repository_write",
+            request_hash="x" * 64, target_paths=["test.yaml"],
+            required_permission="dashboard:testing:WRITE",
+            expires_at=datetime.now(timezone.utc) + timedelta(seconds=300),
+        )
+        repo.create_gate(gate)
+
+        found = repo.get_gate(gate.id, run.id)
+        assert found is not None
+# #endregion Test.AgentRuns.Repository