153 lines
7.2 KiB
Python
153 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from agents.steward import StewardService
|
|
from control_plane.projects.models import CommitRecord, Milestone, Project, ProjectPlan, StewardAction, StewardFinding, StewardPolicy, TaskStatus
|
|
from graph.bootstrap import champion_steward_run_graph_v1
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from graph.models import GraphApprovalStatus, GraphRun, GraphRunStatus
|
|
from graph.steward import steward_registry
|
|
from tests.test_m2_autonomous_loop import create_disposable_django_repo
|
|
|
|
|
|
def enrolled_project(tmp_path: Path, *, metadata: dict[str, object] | None = None, approval_requirements: dict[str, str] | None = None) -> tuple[Project, StewardService]:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
project = Project.objects.create(name="Stewarded", goal="Keep system healthy", repository_path=str(repo))
|
|
policy = StewardPolicy.objects.create(
|
|
name="Steward Test Policy",
|
|
enabled_checks=["DEPENDENCY_DRIFT", "RUNTIME_CI", "SECURITY", "SECRET_EXPIRY", "PERFORMANCE"],
|
|
metadata=metadata or {},
|
|
severity_thresholds={"performance_delta_percent": 20},
|
|
approval_requirements=approval_requirements or {"REPAIR": "CRITICAL"},
|
|
)
|
|
service = StewardService()
|
|
service.enroll_project(project, policy)
|
|
return project, service
|
|
|
|
|
|
def run_steward_graph(project: Project, service: StewardService):
|
|
enrollment = project.steward_enrollments.get(status="ACTIVE")
|
|
version = champion_steward_run_graph_v1()
|
|
steward_run = service.start_run(enrollment, version)
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, project=project, current_node=version.graph_spec["entry"], metadata={"steward_run_id": str(steward_run.id)})
|
|
runtime = LangGraphRuntime(steward_registry(service))
|
|
runtime.run_until_terminal_or_paused(graph_run)
|
|
steward_run.refresh_from_db()
|
|
graph_run.refresh_from_db()
|
|
return steward_run, graph_run
|
|
|
|
|
|
def test_steward_enrollment_checks_and_repair_routing(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(
|
|
tmp_path,
|
|
metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}], "secret_expiry_metadata": [{"name": "api_cert", "expires_at": "2026-09-01", "severity": "MEDIUM"}]},
|
|
)
|
|
|
|
steward_run, graph_run = run_steward_graph(project, service)
|
|
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert steward_run.status == "COMPLETE"
|
|
assert steward_run.checks.count() == 5
|
|
findings = list(steward_run.findings.order_by("finding_type"))
|
|
assert [finding.finding_type for finding in findings] == ["SECRET_EXPIRY", "SECURITY_SIGNAL"]
|
|
actions = list(StewardAction.objects.select_related("task", "finding"))
|
|
assert len(actions) == 2
|
|
assert all(action.action_type == "REPAIR" for action in actions)
|
|
assert all(action.task and action.task.status == TaskStatus.READY for action in actions)
|
|
assert "api_cert" in actions[0].finding.evidence.get("expiring_secrets", [{}])[0].get("name", "") or "api_cert" in actions[1].finding.evidence.get("expiring_secrets", [{}])[0].get("name", "")
|
|
|
|
|
|
def test_steward_deduplicates_findings_across_runs(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})
|
|
|
|
first_run, _ = run_steward_graph(project, service)
|
|
second_run, _ = run_steward_graph(project, service)
|
|
|
|
assert StewardFinding.objects.count() == 1
|
|
finding = StewardFinding.objects.get()
|
|
assert finding.occurrence_count == 2
|
|
assert finding.steward_run == second_run
|
|
assert first_run.findings.count() == 0
|
|
|
|
|
|
def test_high_severity_repair_pauses_until_approval(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-2", "severity": "HIGH"}]}, approval_requirements={"REPAIR": "HIGH"})
|
|
|
|
steward_run, graph_run = run_steward_graph(project, service)
|
|
|
|
assert graph_run.status == GraphRunStatus.PAUSED
|
|
assert graph_run.approvals.get().status == GraphApprovalStatus.PENDING
|
|
action = StewardAction.objects.get()
|
|
assert action.status == "APPROVAL_REQUIRED"
|
|
assert action.task is None
|
|
|
|
approval = graph_run.approvals.get()
|
|
approval.status = GraphApprovalStatus.APPROVED
|
|
approval.decided_by = "tester"
|
|
approval.save(update_fields=["status", "decided_by", "updated_at"])
|
|
graph_run.status = GraphRunStatus.RUNNING
|
|
graph_run.save(update_fields=["status", "updated_at"])
|
|
LangGraphRuntime(steward_registry(service)).run_until_terminal_or_paused(graph_run)
|
|
|
|
graph_run.refresh_from_db()
|
|
steward_run.refresh_from_db()
|
|
action.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert steward_run.status == "COMPLETE"
|
|
assert action.status == "ROUTED"
|
|
assert action.task is not None
|
|
|
|
|
|
def test_steward_routes_systemic_runtime_signal_to_progeny_investigation(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(tmp_path)
|
|
enrollment = project.steward_enrollments.get()
|
|
steward_run = service.start_run(enrollment)
|
|
check = service._check(steward_run, "RUNTIME_CI", "FAIL", {"events": [{"event_type": "CI_FAILED"}]}, "HIGH")
|
|
raw = service._findings_for_check(check)[0]
|
|
finding = service.upsert_finding(steward_run, check, raw)
|
|
finding.occurrence_count = 2
|
|
finding.save(update_fields=["occurrence_count"])
|
|
service.classify_findings(steward_run)
|
|
|
|
action = service.route_findings(steward_run)[0]
|
|
|
|
assert action.action_type == "INVESTIGATE"
|
|
assert action.investigation is not None
|
|
assert action.task is None
|
|
|
|
|
|
def test_repair_resolution_links_commit_to_steward_finding(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})
|
|
run_steward_graph(project, service)
|
|
action = StewardAction.objects.select_related("task", "finding").get()
|
|
task = action.task
|
|
assert task is not None
|
|
task.status = TaskStatus.COMPLETE
|
|
task.save(update_fields=["status", "updated_at"])
|
|
CommitRecord.objects.create(project=project, task=task, sha="abc123", branch_name="repair", message="Repair Steward finding")
|
|
|
|
assert service.resolve_after_repair(action.finding) is True
|
|
|
|
action.finding.refresh_from_db()
|
|
assert action.finding.status == "RESOLVED"
|
|
assert CommitRecord.objects.get(task=task).steward_finding == action.finding
|
|
|
|
|
|
def test_resolved_identical_findings_do_not_create_duplicate_repairs(tmp_path: Path) -> None:
|
|
project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})
|
|
run_steward_graph(project, service)
|
|
action = StewardAction.objects.select_related("task", "finding").get()
|
|
task = action.task
|
|
assert task is not None
|
|
task.status = TaskStatus.COMPLETE
|
|
task.save(update_fields=["status", "updated_at"])
|
|
CommitRecord.objects.create(project=project, task=task, sha="abc123", branch_name="repair", message="Repair Steward finding")
|
|
service.resolve_after_repair(action.finding)
|
|
|
|
run_steward_graph(project, service)
|
|
|
|
assert StewardFinding.objects.count() == 1
|
|
assert StewardAction.objects.count() == 1
|
|
assert project.tasks.filter(task_type="repair").count() == 1
|
|
assert StewardFinding.objects.get().status == "RESOLVED"
|