Add parallel Placement pipeline for routing-first topology plans.
Ship a free, analysis-independent job that writes placement_plan.json (domains/satellites, no mm) with API, SSE progress, and a minimal project UI. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -56,3 +56,11 @@ def test_domains_cover_all_ics():
|
||||
covered = {r for d in report.domains for r in d.ic_refs}
|
||||
assert covered == {"U1", "U2", "U3"}
|
||||
assert all(d.assemble_order for d in report.domains)
|
||||
|
||||
|
||||
def test_build_placement_plan_alias():
|
||||
from backend.pinscopex.functional_groups import build_placement_plan
|
||||
report = build_placement_plan(_graph())
|
||||
assert report.objective == "routing"
|
||||
assert {g.ref for g in report.groups} >= {"U1", "U2", "U3"}
|
||||
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Placement pipeline smoke — topology only, no LLM."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.pinscopex.functional_groups import (
|
||||
FunctionalGroupsReport,
|
||||
build_placement_plan,
|
||||
)
|
||||
from backend.pinscopex.models import DesignGraph
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SIMPLE = ROOT / "simple_project"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def graph() -> DesignGraph:
|
||||
path = SIMPLE / "design_graph.json"
|
||||
return DesignGraph.model_validate_json(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_placement_plan_writes_domains_and_groups(graph: DesignGraph, tmp_path: Path):
|
||||
plan = build_placement_plan(graph)
|
||||
assert plan.objective == "routing"
|
||||
assert plan.domains
|
||||
assert plan.groups
|
||||
out = tmp_path / "placement_plan.json"
|
||||
out.write_text(plan.model_dump_json(indent=2) + "\n")
|
||||
loaded = FunctionalGroupsReport.model_validate_json(out.read_text())
|
||||
assert len(loaded.groups) == len(plan.groups)
|
||||
|
||||
|
||||
def test_placement_busy_helpers():
|
||||
from backend.services.projects import ProjectMeta, STATUS_QUEUED, STATUS_RUNNING
|
||||
|
||||
# Mirror placement_pipeline helpers without importing the worker stack
|
||||
# (that pulls Anthropic via services.pipeline in lean test envs).
|
||||
active = frozenset({"queued", "running"})
|
||||
analysis = frozenset({STATUS_QUEUED, STATUS_RUNNING})
|
||||
|
||||
draft = ProjectMeta(id="p", name="t", created="2026-01-01", user_id="u")
|
||||
assert (draft.placement_status or "draft") not in active
|
||||
assert draft.status not in analysis
|
||||
|
||||
draft.placement_status = "queued"
|
||||
assert (draft.placement_status or "draft") in active
|
||||
draft.status = STATUS_RUNNING
|
||||
assert draft.status in analysis
|
||||
Reference in New Issue
Block a user