147 lines
5.5 KiB
Python
147 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
from control_plane.events.models import Event, EventType
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from control_plane.projects.models import Project, ProjectPlan, ProjectStatus, TaskStatus
|
|
from project_brain.planning import (
|
|
PlannedFeature,
|
|
PlannedMilestone,
|
|
PlannedTask,
|
|
PlanValidationError,
|
|
ProjectPlanBuilder,
|
|
ProjectPlanContract,
|
|
parse_project_plan_response,
|
|
)
|
|
|
|
|
|
def test_m3_project_plan_builder_creates_task_dag_seed() -> None:
|
|
project = Project.objects.create(name="Planned App", goal="Draft")
|
|
contract = ProjectPlanContract(
|
|
goal="Build a tiny Django app",
|
|
scope="One endpoint with tests",
|
|
stack={"backend": "Django"},
|
|
acceptance_criteria=["project tests pass"],
|
|
milestones=[
|
|
PlannedMilestone(
|
|
key="M1",
|
|
title="Health Endpoint",
|
|
goal="Expose health check",
|
|
features=[
|
|
PlannedFeature(
|
|
key="F1",
|
|
title="Health API",
|
|
tasks=[
|
|
PlannedTask(
|
|
task_id="T1",
|
|
goal="Add a /health endpoint returning JSON ok",
|
|
acceptance_criteria=["/health returns ok", "tests pass"],
|
|
)
|
|
],
|
|
)
|
|
],
|
|
)
|
|
],
|
|
)
|
|
|
|
plan = ProjectPlanBuilder().apply(project, contract)
|
|
|
|
project.refresh_from_db()
|
|
assert project.status == ProjectStatus.READY
|
|
assert project.current_plan_version == plan.version
|
|
assert project.milestones.count() == 1
|
|
assert project.features.count() == 1
|
|
assert project.tasks.filter(status=TaskStatus.READY).count() == 1
|
|
event_types = set(Event.objects.filter(project=project).values_list("event_type", flat=True))
|
|
assert EventType.PLAN_APPROVED in event_types
|
|
assert EventType.MILESTONE_CREATED in event_types
|
|
assert EventType.TASK_CREATED in event_types
|
|
assert EventType.TASK_READY in event_types
|
|
|
|
|
|
def valid_plan_payload() -> dict[str, object]:
|
|
return {
|
|
"project_plan": {
|
|
"goal": "Build a tiny Django app",
|
|
"scope": "One endpoint",
|
|
"acceptance_criteria": ["project tests pass"],
|
|
"milestones": [
|
|
{
|
|
"key": "M1",
|
|
"title": "API",
|
|
"goal": "Build API",
|
|
"features": [
|
|
{
|
|
"key": "F1",
|
|
"title": "Health",
|
|
"tasks": [
|
|
{
|
|
"id": "T1",
|
|
"goal": "Add health endpoint",
|
|
"acceptance_criteria": ["health endpoint passes tests"],
|
|
"dependencies": [],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
|
|
def test_m3_parse_valid_sol_plan_response() -> None:
|
|
contract = parse_project_plan_response(json.dumps(valid_plan_payload()))
|
|
|
|
assert contract.goal == "Build a tiny Django app"
|
|
assert contract.milestones[0].features[0].tasks[0].task_id == "T1"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutate, message",
|
|
[
|
|
(lambda p: p.update({"project_plan": "bad"}), "project_plan"),
|
|
(lambda p: p["project_plan"].pop("acceptance_criteria"), "acceptance_criteria"),
|
|
(lambda p: p["project_plan"]["milestones"][0]["features"][0]["tasks"][0].pop("acceptance_criteria"), "acceptance_criteria"),
|
|
(lambda p: p["project_plan"]["milestones"][0]["features"][0]["tasks"].append(dict(p["project_plan"]["milestones"][0]["features"][0]["tasks"][0])), "Duplicate task"),
|
|
(lambda p: p["project_plan"]["milestones"][0]["features"][0]["tasks"][0].update({"dependencies": ["NOPE"]}), "nonexistent"),
|
|
],
|
|
)
|
|
def test_m3_rejects_malformed_sol_plans_atomically(mutate, message: str) -> None:
|
|
payload = valid_plan_payload()
|
|
mutate(payload)
|
|
project = Project.objects.create(name="Bad Plan", goal="Draft")
|
|
|
|
with pytest.raises(PlanValidationError, match=message):
|
|
contract = parse_project_plan_response(json.dumps(payload))
|
|
ProjectPlanBuilder().apply(project, contract)
|
|
|
|
assert ProjectPlan.objects.filter(project=project).count() == 0
|
|
|
|
|
|
def test_m3_rejects_dependency_cycles_atomically() -> None:
|
|
payload = valid_plan_payload()
|
|
tasks = payload["project_plan"]["milestones"][0]["features"][0]["tasks"]
|
|
tasks[0]["dependencies"] = ["T2"]
|
|
tasks.append({"id": "T2", "goal": "Second", "acceptance_criteria": ["done"], "dependencies": ["T1"]})
|
|
project = Project.objects.create(name="Cycle", goal="Draft")
|
|
|
|
with pytest.raises(PlanValidationError, match="cycle"):
|
|
contract = parse_project_plan_response(json.dumps(payload))
|
|
ProjectPlanBuilder().apply(project, contract)
|
|
|
|
assert ProjectPlan.objects.filter(project=project).count() == 0
|
|
|
|
|
|
def test_m3_rejects_invalid_feature_references_atomically() -> None:
|
|
payload = valid_plan_payload()
|
|
payload["project_plan"]["milestones"][0]["features"][0]["key"] = ""
|
|
project = Project.objects.create(name="Bad Feature", goal="Draft")
|
|
|
|
with pytest.raises(PlanValidationError, match="key"):
|
|
contract = parse_project_plan_response(json.dumps(payload))
|
|
ProjectPlanBuilder().apply(project, contract)
|
|
|
|
assert ProjectPlan.objects.filter(project=project).count() == 0
|