123 lines
8.5 KiB
Python
123 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
|
|
from agents.control_room import AgentControlRoomService
|
|
from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
|
|
from control_plane.agents.models import AgentRole, AgentVersion, PromotionStatus, ProgenySignal
|
|
from control_plane.events.bus import EventBus
|
|
from control_plane.events.models import EventType
|
|
from control_plane.projects.models import Exploration, ExplorationOpportunity, Project, ProjectPlan, Milestone, RoadmapItem, Scenario, ScenarioFinding, ScenarioRun, ScenarioSuite, StewardFinding, Task, TaskAttempt, TaskStatus
|
|
from control_plane.resources.models import Resource, ResourceKind
|
|
from control_plane.verification.models import Review, TestRun, Verification, VerificationLevel, VerificationResult
|
|
from graph.bootstrap import champion_project_exploration_graph_v1, champion_task_execution_graph_v1
|
|
from graph.models import GraphApproval, GraphRun, GraphRunStatus
|
|
|
|
|
|
def fixture() -> dict[str, object]:
|
|
SeedAgentsCommand().handle()
|
|
AgentControlRoomService().bootstrap_frontend_agents()
|
|
team = AgentControlRoomService().create_software_feature_team()
|
|
Resource.objects.create(name="Qwen", kind=ResourceKind.MODEL, provider="local_inference", roles=["CODING"], health_status="AVAILABLE", config={"model": "qwen38"})
|
|
project = Project.objects.create(name="UI Dogfood", goal="Operate Artifex through the UI", repository_path="/tmp/ui")
|
|
plan = ProjectPlan.objects.create(project=project, version=1, goal=project.goal, scope="UI V1")
|
|
milestone = Milestone.objects.create(project=project, plan=plan, key="UI", title="UI", goal="Build UI")
|
|
task = Task.objects.create(project=project, milestone=milestone, task_type="implementation", status=TaskStatus.COMPLETE, goal="Render dashboard", acceptance_criteria=["loads"])
|
|
coder = AgentVersion.objects.get(agent__role=AgentRole.CODER, promotion_status=PromotionStatus.CHAMPION)
|
|
reviewer = AgentVersion.objects.get(agent__role=AgentRole.REVIEWER, promotion_status=PromotionStatus.CHAMPION)
|
|
judge = AgentVersion.objects.get(agent__role=AgentRole.PROJECT_JUDGE, promotion_status=PromotionStatus.CHAMPION)
|
|
TaskAttempt.objects.create(task=task, attempt_number=1, coder=coder, status="COMPLETE")
|
|
TestRun.objects.create(project=project, task=task, command="pytest", status="PASS")
|
|
Review.objects.create(task=task, reviewer=reviewer, status="PASS")
|
|
Verification.objects.create(project=project, task=task, judge=judge, level=VerificationLevel.TASK, result=VerificationResult.PASS)
|
|
graph_version = champion_task_execution_graph_v1()
|
|
graph_run = GraphRun.objects.create(project=project, task=task, execution_graph_version=graph_version, current_node="review", status=GraphRunStatus.PAUSED, metadata={"telemetry": {}})
|
|
node_run = graph_run.node_runs.create(node_id="review", node_type="review", status="PAUSED", visit_index=1, agent_version=reviewer, output_metadata={"review_status": "RUNNING"})
|
|
approval = GraphApproval.objects.create(graph_run=graph_run, node_run=node_run, reason="AWAITING_UI_APPROVAL")
|
|
exploration = Exploration.objects.create(project=project, status="COMPLETE")
|
|
opportunity = ExplorationOpportunity.objects.create(exploration=exploration, project=project, title="Add status filter", description="Filter dashboard cards", opportunity_type="UX", recommended_action="EXTEND", value_score=0.8, effort_score=0.2, risk_score=0.2, confidence=0.8, technical_fit=0.8, strategic_fit=0.8, composite_score=0.75, grouping_key="ui-opp")
|
|
roadmap_item = RoadmapItem.objects.create(project=project, title="Improve graph viewer", description="Expose node details", horizon="NOW", target_action="EXTEND", grouping_key="ui-roadmap")
|
|
suite = ScenarioSuite.objects.create(project=project, name="UI Suite")
|
|
scenario = Scenario.objects.create(project=project, suite=suite, name="Permission denial", title="Permission denial", scenario_type="PERMISSION", target_type="PROJECT", target_id=str(project.id), injected_condition={"mechanism": "permission_denial"}, expected_invariants=["safe"], success_criteria=["pass"], resource_budget={"max_seconds": 5})
|
|
run = ScenarioRun.objects.create(project=project, scenario=scenario, result="FAIL", status="COMPLETE", failure_evidence={"summary": "denied"})
|
|
finding = ScenarioFinding.objects.create(project=project, scenario=scenario, scenario_run=run, title="Scenario failed", summary="permission denied", failure_category="PERMISSION", recommended_action="EXTEND", recommended_route="RoadmapItem", grouping_key="ui-scenario")
|
|
StewardFinding.objects.create(project=project, finding_type="RELIABILITY", title="Open issue", summary="Needs attention", severity="HIGH", confidence=0.8, recommended_action="REPAIR", grouping_key="ui-steward")
|
|
ProgenySignal.objects.create(project=project, agent_version=coder, source="ui", severity="MEDIUM", failure_category="MALFORMED", summary="Signal", grouping_key="ui-signal")
|
|
EventBus().publish(EventType.TASK_COMPLETED, project=project, task=task, actor="test")
|
|
return {"project": project, "task": task, "graph_run": graph_run, "approval": approval, "opportunity": opportunity, "roadmap_item": roadmap_item, "finding": finding, "coder": coder, "team": team}
|
|
|
|
|
|
def assert_ok(client: Client, url: str, text: str) -> None:
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert text.encode() in response.content
|
|
|
|
|
|
def test_control_plane_core_pages_render_and_link() -> None:
|
|
data = fixture()
|
|
client = Client()
|
|
project = data["project"]
|
|
task = data["task"]
|
|
graph_run = data["graph_run"]
|
|
coder = data["coder"]
|
|
|
|
assert_ok(client, reverse("dashboard"), "Dashboard")
|
|
assert_ok(client, reverse("projects"), "UI Dogfood")
|
|
assert_ok(client, reverse("project_workspace", args=[project.id]), "Project DAG")
|
|
assert_ok(client, reverse("project_brain", args=[project.id]), "Project Brain")
|
|
assert_ok(client, reverse("project_archaeologist", args=[project.id]), "Archaeologist")
|
|
assert_ok(client, reverse("graph_run_detail", args=[graph_run.id]), "Live execution graph viewer")
|
|
assert_ok(client, reverse("task_detail", args=[task.id]), "Acceptance")
|
|
assert_ok(client, reverse("steward"), "Open issue")
|
|
assert_ok(client, reverse("explore"), "Add status filter")
|
|
assert_ok(client, reverse("roadmap"), "Improve graph viewer")
|
|
assert_ok(client, reverse("scenarios"), "Permission denial")
|
|
assert_ok(client, reverse("progeny"), "Signal")
|
|
assert_ok(client, reverse("agent_control_room"), "Agent Control Room")
|
|
assert_ok(client, reverse("agent_detail", args=[coder.id]), "Competencies")
|
|
assert_ok(client, reverse("resources"), "Qwen")
|
|
assert_ok(client, reverse("approvals"), "AWAITING_UI_APPROVAL")
|
|
assert_ok(client, reverse("activity"), "TASK_COMPLETED")
|
|
|
|
|
|
def test_json_endpoints_and_ui_actions_work() -> None:
|
|
data = fixture()
|
|
client = Client()
|
|
project = data["project"]
|
|
graph_run = data["graph_run"]
|
|
approval = data["approval"]
|
|
opportunity = data["opportunity"]
|
|
roadmap_item = data["roadmap_item"]
|
|
finding = data["finding"]
|
|
|
|
assert client.get(reverse("graph_run_json", args=[graph_run.id])).json()["status"] == "PAUSED"
|
|
assert client.get(reverse("project_dag_json", args=[project.id])).json()["tasks"]
|
|
assert client.get(reverse("agent_performance_json", args=[data["coder"].id])).json()["quality"]
|
|
|
|
response = client.post(reverse("opportunity_action", args=[opportunity.id]), {"action": "roadmap"})
|
|
assert response.status_code == 302
|
|
assert project.roadmap_items.filter(title="Add status filter").exists()
|
|
|
|
response = client.post(reverse("roadmap_action", args=[roadmap_item.id]), {"action": "NEXT"})
|
|
assert response.status_code == 302
|
|
roadmap_item.refresh_from_db()
|
|
assert roadmap_item.horizon == "NEXT"
|
|
|
|
response = client.post(reverse("scenario_finding_action", args=[finding.id]), {"action": "roadmap"})
|
|
assert response.status_code == 302
|
|
|
|
response = client.post(reverse("approval_action", args=[approval.id]), {"action": "approve"})
|
|
assert response.status_code == 302
|
|
approval.refresh_from_db()
|
|
graph_run.refresh_from_db()
|
|
assert approval.status == "APPROVED"
|
|
assert graph_run.status == "RUNNING"
|
|
|
|
|
|
def test_ui_can_start_safe_explore_graph() -> None:
|
|
data = fixture()
|
|
client = Client()
|
|
response = client.post(reverse("run_explore", args=[data["project"].id]))
|
|
assert response.status_code == 302
|
|
assert GraphRun.objects.filter(execution_graph_version__graph__name="project_exploration", project=data["project"]).exists()
|