From 19eab090003c60f4cd854509dc551e906cb85c21 Mon Sep 17 00:00:00 2001 From: Michele Bigi Date: Sun, 20 Sep 2026 15:34:41 +0200 Subject: [PATCH] Native PipelineWorkspace and event broker in periscope/src. PCB and placement jobs import job_workspace instead of PinScope pipeline.py. Analysis run_pipeline re-exports the same broker singleton. --- .../dependency/backend/services/pipeline.py | 214 +----------------- .../dependency/frontend/content/changelog.md | 7 + .../src/backend/services/job_workspace.py | 192 ++++++++++++++++ .../src/backend/services/pcb_pipeline.py | 2 +- .../backend/services/placement_pipeline.py | 2 +- .../development/PINSCOPE_INDEPENDENCE_PLAN.md | 4 +- tests/test_native_job_workspace.py | 113 +++++++++ 7 files changed, 328 insertions(+), 206 deletions(-) create mode 100644 periscope/src/backend/services/job_workspace.py create mode 100644 tests/test_native_job_workspace.py diff --git a/periscope/dependency/backend/services/pipeline.py b/periscope/dependency/backend/services/pipeline.py index 3df8e3e..51a2100 100644 --- a/periscope/dependency/backend/services/pipeline.py +++ b/periscope/dependency/backend/services/pipeline.py @@ -95,58 +95,20 @@ def _git_commit() -> str: # --------------------------------------------------------------------------- -# SSE Event Broker -# --------------------------------------------------------------------------- +# SSE Event Broker + workspace live in periscope/src job_workspace.py. +# Re-export so analysis run_pipeline and pipeline_worker.set_broker stay +# on one singleton. PinScope pipeline.py is not empty-deleted. +from backend.services import job_workspace as _job_ws -class EventBroker: - """In-memory pub/sub for SSE events, keyed by project_id. - - Buffers all events per project so late subscribers (e.g. after a page - navigation) receive the full history before seeing live events. - """ - - def __init__(self): - self._queues: dict[str, list[asyncio.Queue]] = {} - self._history: dict[str, list[dict]] = {} - - def subscribe(self, project_id: str) -> asyncio.Queue: - q: asyncio.Queue = asyncio.Queue() - # Replay buffered events so the subscriber catches up - for msg in self._history.get(project_id, []): - q.put_nowait(msg) - self._queues.setdefault(project_id, []).append(q) - return q - - def unsubscribe(self, project_id: str, q: asyncio.Queue) -> None: - qs = self._queues.get(project_id, []) - if q in qs: - qs.remove(q) - if not qs: - self._queues.pop(project_id, None) - - def clear_history(self, project_id: str) -> None: - self._history.pop(project_id, None) - - def publish(self, project_id: str, event: str, data: dict) -> None: - msg = {"event": event, "data": data} - self._history.setdefault(project_id, []).append(msg) - for q in self._queues.get(project_id, []): - q.put_nowait(msg) +EventBroker = _job_ws.EventBroker +PipelineWorkspace = _job_ws.PipelineWorkspace +broker = _job_ws.broker -broker: EventBroker = EventBroker() - - -def set_broker(b: EventBroker) -> None: - """Replace the module-level broker. - - Called by :mod:`backend.pipeline_worker` at startup to swap in the - GCS-backed broker so events written from the worker are visible to - the API's SSE handler. Must be called *before* :func:`run_pipeline` - or :func:`run_regen_pipeline`. - """ - global broker - broker = b +def set_broker(b) -> None: + """Swap the worker event broker (GCS in prod). Updates native + this module.""" + _job_ws.set_broker(b) + globals()["broker"] = b # Per-process cancel-flag cache: re-reading the project meta from GCS on @@ -189,160 +151,6 @@ def _cancel_gate_check(ctx: PipelineContext) -> None: raise CancelRequested(f"cancel requested for {ctx.project_id}") -# --------------------------------------------------------------------------- -# Pipeline Workspace -# --------------------------------------------------------------------------- - -class PipelineWorkspace: - """Downloads project files from storage to a temp dir for pipeline execution. - - The periscopex core library operates on local paths. This context manager - downloads inputs at enter, provides local paths, and uploads results at exit. - """ - - def __init__( - self, - storage: StorageBackend, - user_id: str, - project_id: str, - ) -> None: - self.storage = storage - self.user_id = user_id - self.project_id = project_id - self.prefix = proj_svc.project_prefix(user_id, project_id) - self._tmpdir: tempfile.TemporaryDirectory | None = None - self.local_dir: Path = Path() - - async def __aenter__(self) -> PipelineWorkspace: - self._tmpdir = tempfile.TemporaryDirectory() - self.local_dir = Path(self._tmpdir.name) - - # Create subdirectories - (self.local_dir / "uploads" / "datasheets").mkdir(parents=True) - (self.local_dir / "extracted").mkdir(parents=True) - (self.local_dir / "patterns").mkdir(parents=True) - (self.local_dir / "models").mkdir(parents=True) - (self.local_dir / "taxonomy").mkdir(parents=True) - - # Download project files from storage - all_keys = self.storage.list_recursive(self.prefix) - for key in all_keys: - # key is like users/{uid}/projects/{pid}/uploads/bom.csv - # We want the relative part after the project prefix - rel = key[len(self.prefix) + 1:] # strip prefix + trailing / - local_path = self.local_dir / rel - self.storage.download_to_local(key, local_path) - - # Download taxonomy files from storage - taxonomy_keys = self.storage.list_prefix("taxonomy/") - for key in taxonomy_keys: - if key.endswith(".json"): - filename = key.rsplit("/", 1)[-1] - self.storage.download_to_local(key, self.local_dir / "taxonomy" / filename) - - # Seed from repo taxonomy if storage had no taxonomy files yet - local_tax = self.local_dir / "taxonomy" - if not any(local_tax.glob("*.json")): - repo_tax = settings.taxonomy_dir - if repo_tax.is_dir(): - for f in repo_tax.glob("*.json"): - shutil.copy2(f, local_tax / f.name) - - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - if exc_type is None: - # Upload outputs back to storage - self._upload_dir("extracted") - self._upload_dir("patterns") - self._upload_dir("models") - self._upload_file("design_graph.json") - self._upload_file("layout_graph.json") - self._upload_file("impedance_nets.json") - self._upload_file("functional_groups.json") - self._upload_file("placement_plan.json") - self._upload_file("bom_summary.json") - self._upload_file("derating.json") - self._upload_file("report.json") - self._upload_file("pcb_report.json") - self._upload_file("pcb_inventory.json") - self._upload_file("periscope-findings.json") - self._upload_file("review_fingerprints.json") - self._upload_file("api_logs.jsonl") - - # Merge taxonomy: read current from storage, add any new entries - # from this run, write back. This avoids clobbering subtypes - # that a concurrent pipeline added while we were running. - tax_dir = self.local_dir / "taxonomy" - if tax_dir.is_dir(): - for f in tax_dir.iterdir(): - if f.is_file() and f.suffix == ".json": - local_data = json.loads(f.read_text()) - local_subtypes = local_data.get("subtypes", {}) - storage_key = f"taxonomy/{f.name}" - - if self.storage.exists(storage_key): - current = self.storage.read_json(storage_key) - merged = current.get("subtypes", {}) - for key, entry in local_subtypes.items(): - if key not in merged: - merged[key] = entry - else: - # Backfill fields the local run generated - # (e.g. specs_schema) that the - # storage copy is missing. - for field, value in entry.items(): - if field not in merged[key]: - merged[key][field] = value - current["subtypes"] = merged - self.storage.write_json(storage_key, current) - else: - self.storage.write_json(storage_key, local_data) - - if self._tmpdir: - self._tmpdir.cleanup() - - def _upload_dir(self, subdir: str) -> None: - """Upload all files in a subdirectory back to storage.""" - local = self.local_dir / subdir - if not local.is_dir(): - return - for f in local.rglob("*"): - if f.is_file(): - rel = f.relative_to(self.local_dir) - key = f"{self.prefix}/{rel}" - self.storage.upload_from_local(f, key) - - def _upload_file(self, name: str) -> None: - """Upload a single file back to storage if it exists.""" - local = self.local_dir / name - if local.is_file(): - self.storage.upload_from_local(local, f"{self.prefix}/{name}") - - def local_path(self, rel: str) -> Path: - """Get a local path within the workspace.""" - return self.local_dir / rel - - def netlist_local_path(self) -> Path: - """Local path of whichever netlist file was synced (``.asc`` or ``.edn``). - - Pipeline workspace mirrors the entire project prefix, so whichever - format the user uploaded lands locally with its original extension. - Falls back to ``uploads/netlist.asc`` if neither exists — downstream - code will raise a clearer error when it tries to read the missing - file than a ``None`` return would. - """ - for ext in ("asc", "edn", "xml", "kicad_net", "kicad_sch"): - p = self.local_dir / "uploads" / f"netlist.{ext}" - if p.exists(): - return p - return self.local_dir / "uploads" / "netlist.asc" - - @property - def taxonomy_dir(self) -> Path: - return self.local_dir / "taxonomy" - - # --------------------------------------------------------------------------- # Pipeline context — shared state threaded through all stage functions # --------------------------------------------------------------------------- diff --git a/periscope/dependency/frontend/content/changelog.md b/periscope/dependency/frontend/content/changelog.md index a9fa6a1..c1c9daf 100644 --- a/periscope/dependency/frontend/content/changelog.md +++ b/periscope/dependency/frontend/content/changelog.md @@ -2,6 +2,13 @@ What's new in Periscope. +## 2.41.0 — 2026-09-20 — Native job workspace for PCB and placement + +`pcb_pipeline` and `placement_pipeline` no longer import PinScope `pipeline.py`. Workspace download/upload and the in-memory event broker live in `periscope/src` (`job_workspace.py`). Analysis `run_pipeline` still lives in `dependency/` and re-exports the same singleton so `pipeline_worker.set_broker` keeps PCB SSE on the GCS event log. Parsers/graph unchanged. `pipeline.py` not deleted. + +- [New] `backend.services.job_workspace`: `PipelineWorkspace`, `EventBroker`, `set_broker`. +- [Changed] PCB/placement import `job_workspace`; `pipeline.py` re-exports the native types. + ## 2.40.0 — 2026-09-20 — Fase C3: native datasheet extraction Pipeline pintable/pattern/specs/auto-resolve run from `periscope/src` (`datasheet_extract.py`): DeepSeek + local SKILL.md, no Anthropic Console skill ids. Inherited `extraction.py` stays in `dependency/`. diff --git a/periscope/src/backend/services/job_workspace.py b/periscope/src/backend/services/job_workspace.py new file mode 100644 index 0000000..aa918f7 --- /dev/null +++ b/periscope/src/backend/services/job_workspace.py @@ -0,0 +1,192 @@ +"""Native project workspace + in-memory job event broker. + +PCB and placement jobs need a local mirror of storage and an SSE broker +without importing PinScope ``pipeline.py`` (parsers, review, extraction). +``pipeline_worker.set_broker`` must update this module's ``broker`` so PCB +events still go to the GCS event log. + +Inherited ``backend.services.pipeline`` re-exports these names after the +slice is proven. Do not empty-delete ``pipeline.py``. +""" + +from __future__ import annotations + +import asyncio +import json +import shutil +import tempfile +from pathlib import Path + +from backend.config import settings +from backend.services import projects as proj_svc +from backend.services.storage import StorageBackend + + +class EventBroker: + """In-memory pub/sub for SSE events, keyed by project_id. + + Buffers all events per project so late subscribers (e.g. after a page + navigation) receive the full history before seeing live events. + """ + + def __init__(self): + self._queues: dict[str, list[asyncio.Queue]] = {} + self._history: dict[str, list[dict]] = {} + + def subscribe(self, project_id: str) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue() + for msg in self._history.get(project_id, []): + q.put_nowait(msg) + self._queues.setdefault(project_id, []).append(q) + return q + + def unsubscribe(self, project_id: str, q: asyncio.Queue) -> None: + qs = self._queues.get(project_id, []) + if q in qs: + qs.remove(q) + if not qs: + self._queues.pop(project_id, None) + + def clear_history(self, project_id: str) -> None: + self._history.pop(project_id, None) + + def publish(self, project_id: str, event: str, data: dict) -> None: + msg = {"event": event, "data": data} + self._history.setdefault(project_id, []).append(msg) + for q in self._queues.get(project_id, []): + q.put_nowait(msg) + + +broker: EventBroker = EventBroker() + + +def set_broker(b: EventBroker) -> None: + """Replace the module-level broker (GCS-backed in the worker).""" + global broker + broker = b + + +class PipelineWorkspace: + """Downloads project files from storage to a temp dir for pipeline execution. + + The periscopex core library operates on local paths. This context manager + downloads inputs at enter, provides local paths, and uploads results at exit. + """ + + def __init__( + self, + storage: StorageBackend, + user_id: str, + project_id: str, + ) -> None: + self.storage = storage + self.user_id = user_id + self.project_id = project_id + self.prefix = proj_svc.project_prefix(user_id, project_id) + self._tmpdir: tempfile.TemporaryDirectory | None = None + self.local_dir: Path = Path() + + async def __aenter__(self) -> PipelineWorkspace: + self._tmpdir = tempfile.TemporaryDirectory() + self.local_dir = Path(self._tmpdir.name) + + (self.local_dir / "uploads" / "datasheets").mkdir(parents=True) + (self.local_dir / "extracted").mkdir(parents=True) + (self.local_dir / "patterns").mkdir(parents=True) + (self.local_dir / "models").mkdir(parents=True) + (self.local_dir / "taxonomy").mkdir(parents=True) + + all_keys = self.storage.list_recursive(self.prefix) + for key in all_keys: + rel = key[len(self.prefix) + 1:] + local_path = self.local_dir / rel + self.storage.download_to_local(key, local_path) + + taxonomy_keys = self.storage.list_prefix("taxonomy/") + for key in taxonomy_keys: + if key.endswith(".json"): + filename = key.rsplit("/", 1)[-1] + self.storage.download_to_local(key, self.local_dir / "taxonomy" / filename) + + local_tax = self.local_dir / "taxonomy" + if not any(local_tax.glob("*.json")): + repo_tax = settings.taxonomy_dir + if repo_tax.is_dir(): + for f in repo_tax.glob("*.json"): + shutil.copy2(f, local_tax / f.name) + + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + if exc_type is None: + self._upload_dir("extracted") + self._upload_dir("patterns") + self._upload_dir("models") + self._upload_file("design_graph.json") + self._upload_file("layout_graph.json") + self._upload_file("impedance_nets.json") + self._upload_file("functional_groups.json") + self._upload_file("placement_plan.json") + self._upload_file("bom_summary.json") + self._upload_file("derating.json") + self._upload_file("report.json") + self._upload_file("pcb_report.json") + self._upload_file("pcb_inventory.json") + self._upload_file("periscope-findings.json") + self._upload_file("review_fingerprints.json") + self._upload_file("api_logs.jsonl") + + tax_dir = self.local_dir / "taxonomy" + if tax_dir.is_dir(): + for f in tax_dir.iterdir(): + if f.is_file() and f.suffix == ".json": + local_data = json.loads(f.read_text()) + local_subtypes = local_data.get("subtypes", {}) + storage_key = f"taxonomy/{f.name}" + + if self.storage.exists(storage_key): + current = self.storage.read_json(storage_key) + merged = current.get("subtypes", {}) + for key, entry in local_subtypes.items(): + if key not in merged: + merged[key] = entry + else: + for field, value in entry.items(): + if field not in merged[key]: + merged[key][field] = value + current["subtypes"] = merged + self.storage.write_json(storage_key, current) + else: + self.storage.write_json(storage_key, local_data) + + if self._tmpdir: + self._tmpdir.cleanup() + + def _upload_dir(self, subdir: str) -> None: + local = self.local_dir / subdir + if not local.is_dir(): + return + for f in local.rglob("*"): + if f.is_file(): + rel = f.relative_to(self.local_dir) + key = f"{self.prefix}/{rel}" + self.storage.upload_from_local(f, key) + + def _upload_file(self, name: str) -> None: + local = self.local_dir / name + if local.is_file(): + self.storage.upload_from_local(local, f"{self.prefix}/{name}") + + def local_path(self, rel: str) -> Path: + return self.local_dir / rel + + def netlist_local_path(self) -> Path: + for ext in ("asc", "edn", "xml", "kicad_net", "kicad_sch"): + p = self.local_dir / "uploads" / f"netlist.{ext}" + if p.exists(): + return p + return self.local_dir / "uploads" / "netlist.asc" + + @property + def taxonomy_dir(self) -> Path: + return self.local_dir / "taxonomy" diff --git a/periscope/src/backend/services/pcb_pipeline.py b/periscope/src/backend/services/pcb_pipeline.py index 361a7a1..4ffb36b 100644 --- a/periscope/src/backend/services/pcb_pipeline.py +++ b/periscope/src/backend/services/pcb_pipeline.py @@ -29,7 +29,7 @@ from backend.periscopex.models import ( from backend.periscopex.pcb_checks import assign_pcb_finding_ids, run_pcb_checks from backend.periscopex.pcb_inventory import build_pcb_inventory from backend.services import projects as proj_svc -from backend.services.pipeline import PipelineWorkspace, broker +from backend.services.job_workspace import PipelineWorkspace, broker from backend.services.storage import StorageBackend logger = logging.getLogger(__name__) diff --git a/periscope/src/backend/services/placement_pipeline.py b/periscope/src/backend/services/placement_pipeline.py index 26f3f81..2921a62 100644 --- a/periscope/src/backend/services/placement_pipeline.py +++ b/periscope/src/backend/services/placement_pipeline.py @@ -16,7 +16,7 @@ from backend.periscopex.graph import build_graph from backend.periscopex.models import ComponentConstraints, DesignGraph, LayoutGraph from backend.periscopex.placement_pack import build_placement_pack from backend.services import projects as proj_svc -from backend.services.pipeline import PipelineWorkspace, broker +from backend.services.job_workspace import PipelineWorkspace, broker from backend.services.storage import StorageBackend logger = logging.getLogger(__name__) diff --git a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md index 3ad507e..42cc9fa 100644 --- a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md +++ b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md @@ -1,6 +1,6 @@ # Piano — indipendenza architettonica e di licenza da PinScope -**Stato:** split **2.38.0**. C2 review **2.39.0/2.39.1** live. C3 extraction **2.40.0**. `validate.py` / `extraction.py` restano in `dependency/` (non chiamati dal live path). Fork non staccato. +**Stato:** split **2.38.0**. C2 review **2.39.x**. C3 extraction **2.40.0**. C4 PCB off `validate.py`. **2.41.0** native `job_workspace` — PCB/placement non importano `pipeline.py`. Parsers/graph still dependency. Fork non staccato. **Gate Michele:** sostituire/smettere di chiamare un modulo `dependency/` solo dopo pytest + deploy smoke. Se la verifica fallisce, resta il path ereditato. **Sequenza:** split → sostituzione incrementale (C2 loop → C3 extraction → C4 PCB off `validate.py`). **Mai** empty-delete. Parsers/graph (C5/E) e auto-place fuori scope. AGPL resta. @@ -60,6 +60,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope: | Auth self-host | `periscope/src/backend/services/local_jwt.py`, `routers/auth.py` | REPLACEMENT | | LLM DeepSeek | `periscope/src/backend/services/llm/deepseek_provider.py`, `local_skill.py`, `pdf_ingest.py` | NEW | | Review loop C2 | `review_session.py`, `review_parse.py`, `review_tools.py`, `review_context.py`, `constraints_lookup.py` | REPLACEMENT 2.39.0; PinScope files kept | +| Job workspace | `periscope/src/backend/services/job_workspace.py` | REPLACEMENT 2.41.0; PCB/placement off `pipeline.py` | | Deploy | `scripts/update-periscope.sh`, `docker-compose.yml`, `periscope/src/backend/Dockerfile` | NEW (nomi `pinscope_*` ancora WEAK) | | Plugin KiCad | `periscope/src/plugins/kicad/` | NEW | | Check deterministici fork | `dnp_check`, `sequencing_check`, `layout_rules`, … under `periscope/src` | NEW ma **INDIRECT**: usano `models` / graph | @@ -226,6 +227,7 @@ Qui sì si riscrive l’engine ereditato. Ordine interno: | C1 | Adapter: `validate.py` emette solo `Finding` grezzi → `complete_finding` | Già parziale; chiudere i campi doppi | | C2 | **REWRITE** loop per-IC native in `periscope/src` (DeepSeek, tools su `DesignGraph`) | **Shipped 2.39.0** live `review_ic_async` → `review_session`; PinScope files **kept**, not called from live loop | | C3 | Extraction: `local_skill.py` + schemi JSON (KEEP schema se identici; REWRITE orchestrazione Anthropic) | **Shipped 2.40.0** live `datasheet_extract`; PinScope `extraction.py` kept | +| C3b | Native `PipelineWorkspace` / event broker | **Shipped 2.41.0** `job_workspace.py`; `pipeline.py` kept as analysis orchestrator | | C4 | Spegnere import da `validate.py` nel PCB (`_parse_review` → parser finding nativo) | **Shipped 2.39.1** `pcb_validation` → `review_parse` / `review_session` | | C5 | Test golden `simple_project` + Emmaforo: parity FACT/REQUIREMENT, non parity prose | | diff --git a/tests/test_native_job_workspace.py b/tests/test_native_job_workspace.py new file mode 100644 index 0000000..d080491 --- /dev/null +++ b/tests/test_native_job_workspace.py @@ -0,0 +1,113 @@ +"""Prove native workspace/broker before PCB/placement stop importing pipeline.py.""" + +from __future__ import annotations + +import ast +import inspect +from pathlib import Path + +import pytest + +from backend.services.job_workspace import ( + EventBroker, + PipelineWorkspace, + set_broker, +) +from backend.services.job_workspace import broker as native_broker +from backend.services.pipeline import EventBroker as InheritedBroker +from backend.services.pipeline import PipelineWorkspace as InheritedWorkspace +from backend.services.storage import LocalStorageBackend + + +def test_native_job_workspace_does_not_import_pipeline(): + import backend.services.job_workspace as jw + + tree = ast.parse(Path(jw.__file__).read_text()) + imported = [ + node.module + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module + ] + assert "backend.services.pipeline" not in imported + assert "backend.periscopex.graph" not in imported + assert "backend.periscopex.parsers" not in imported + + +def test_event_broker_publish_subscribe_and_replay(): + b = EventBroker() + b.publish("p1", "pcb_step_update", {"stage": "parse_pcb"}) + q = b.subscribe("p1") + msg = q.get_nowait() + assert msg == {"event": "pcb_step_update", "data": {"stage": "parse_pcb"}} + b.publish("p1", "pcb_complete", {"findings": 1}) + msg2 = q.get_nowait() + assert msg2["event"] == "pcb_complete" + b.unsubscribe("p1", q) + b.clear_history("p1") + q2 = b.subscribe("p1") + assert q2.empty() + + +def test_set_broker_swaps_module_singleton(): + original = native_broker + replacement = EventBroker() + set_broker(replacement) + import backend.services.job_workspace as jw + assert jw.broker is replacement + set_broker(original) + assert jw.broker is original + + +@pytest.mark.asyncio +async def test_workspace_mirrors_storage_and_netlist_like_inherited(tmp_path): + storage = LocalStorageBackend(tmp_path) + prefix = "users/local/projects/ws1" + storage.write_json(f"{prefix}/project.json", {"id": "ws1"}) + bom = tmp_path / "bom.csv" + bom.write_text("Reference,MPN\nU1,PART\n") + storage.upload_from_local(bom, f"{prefix}/uploads/bom.csv") + net = tmp_path / "netlist.edn" + net.write_text("(edif dummy)") + storage.upload_from_local(net, f"{prefix}/uploads/netlist.edn") + + native_net = None + inherited_net = None + async with PipelineWorkspace(storage, "local", "ws1") as nws: + native_net = nws.netlist_local_path() + assert nws.local_path("uploads/bom.csv").is_file() + assert native_net.name == "netlist.edn" + nws.local_path("pcb_report.json").write_text("{}") + nws._upload_file("pcb_report.json") + async with InheritedWorkspace(storage, "local", "ws1") as iws: + inherited_net = iws.netlist_local_path() + assert inherited_net.name == native_net.name + assert storage.exists(f"{prefix}/pcb_report.json") + + +def test_pcb_and_placement_do_not_import_pipeline_py(): + import backend.services.pcb_pipeline as pcb + import backend.services.placement_pipeline as place + + for mod in (pcb, place): + tree = ast.parse(Path(mod.__file__).read_text()) + imported = [ + node.module + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module + ] + assert "backend.services.pipeline" not in imported + assert "backend.services.job_workspace" in imported + + +def test_pipeline_reexports_native_workspace(): + from backend.services import job_workspace as jw + from backend.services import pipeline as pipe + assert pipe.PipelineWorkspace is jw.PipelineWorkspace + assert pipe.EventBroker is jw.EventBroker + assert inspect.signature(EventBroker.publish) == inspect.signature( + InheritedBroker.publish + ) + native_methods = {"local_path", "netlist_local_path", "_upload_file", "_upload_dir"} + for name in native_methods: + assert hasattr(InheritedWorkspace, name) + assert hasattr(PipelineWorkspace, name)