diff --git a/.gitignore b/.gitignore index 4cb57568..9f761a35 100644 --- a/.gitignore +++ b/.gitignore @@ -182,6 +182,7 @@ chipcompiler/tools/ecc_dreamplace/dreamplace .humanize/ humanize-* +.zcode/ docs/superpowers/ findings.md progress.md diff --git a/chipcompiler/engine/flow.py b/chipcompiler/engine/flow.py index 1f61036f..63ed7991 100644 --- a/chipcompiler/engine/flow.py +++ b/chipcompiler/engine/flow.py @@ -304,11 +304,10 @@ def check_step_result(self, workspace_step: WorkspaceStep): ): success = True case StepEnum.RCX.value: - success = True - for spef in ecc_output.spef if ecc_output else []: - if not os.path.exists(spef): - success = False - break + spef_list = ecc_output.spef if ecc_output else [] + success = bool(spef_list) and all( + os.path.isfile(spef) and os.path.getsize(spef) > 0 for spef in spef_list + ) case StepEnum.TIMING_OPT.value: if os.path.exists(output.def_ or "") and os.path.exists(output.verilog or ""): success = True diff --git a/chipcompiler/engine/step_execution.py b/chipcompiler/engine/step_execution.py index 1d7d27ec..0495a8ae 100644 --- a/chipcompiler/engine/step_execution.py +++ b/chipcompiler/engine/step_execution.py @@ -4,7 +4,7 @@ from dataclasses import dataclass from threading import Event, Thread -from chipcompiler.data import Workspace, WorkspaceStep +from chipcompiler.data import StateEnum, Workspace, WorkspaceStep from chipcompiler.engine.db import EngineDB from chipcompiler.utility.log import capture_stdio_to_file, flush_cstdio @@ -87,6 +87,8 @@ def execute_tool_step( ecc_module=engine_db.engine, ) workspace.logger.info(f"[STEP] {step_tag} finished result={result}") + if result is not True and result is not StateEnum.Success: + step_error = f"{step_tag} reported failure (run_step returned {result!r})." except (Exception, SystemExit) as exc: step_error = record_tool_failure(workspace.logger, step_tag, exc) except (Exception, SystemExit) as exc: diff --git a/chipcompiler/tools/ecc/rcx_artifacts.py b/chipcompiler/tools/ecc/rcx_artifacts.py new file mode 100644 index 00000000..c93b03e3 --- /dev/null +++ b/chipcompiler/tools/ecc/rcx_artifacts.py @@ -0,0 +1,126 @@ +"""Publication policy for RCX extraction artifacts. + +iRCX writes extraction results to ``/spef_writer``. This module +resolves the RCX step directories, clears stale extraction artifacts before a +run, and publishes the fresh SPEF set to the step output directory +all-or-nothing: a failed pass restores the previously published SPEFs instead +of destroying them, and a successful pass leaves no backups behind. +""" + +import os +import shutil +from pathlib import Path + +from chipcompiler.data import EccStep, Workspace + + +def _workspace_rcx_dir(path_text: str, workspace_dir: Path) -> Path: + if path_text.startswith("/"): + relative_path = path_text[1:] + if relative_path.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"): + return workspace_dir / relative_path + return Path(path_text) + + +def resolve_rcx_dirs(workspace: Workspace, step: EccStep) -> tuple[Path | None, Path | None]: + """Resolve the RCX extraction data and output directories for a step.""" + workspace_dir = workspace.directory + if workspace_dir is None: + return None, None + + data_dir_text = os.fspath(step.data.dir or "") + output_dir_text = os.fspath(step.output.dir or "") + if not data_dir_text or not output_dir_text: + return None, None + + return ( + _workspace_rcx_dir(data_dir_text, workspace_dir), + _workspace_rcx_dir(output_dir_text, workspace_dir), + ) + + +def wipe_stale_spef_artifacts(data_dir: Path) -> None: + for stale_path in (data_dir / "spef_writer").glob("*.spef"): + stale_path.unlink(missing_ok=True) + + +def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: + data_dir, output_dir = resolve_rcx_dirs(workspace, step) + if data_dir is None or output_dir is None: + workspace.logger.error("RCX data or output directory is not configured") + return False + + spef_writer_dir = data_dir / "spef_writer" + if not spef_writer_dir.is_dir(): + workspace.logger.error("RCX extraction artifacts are missing: %s", spef_writer_dir) + return False + + declared_paths = [spef_path for spef_path in step.output.spef if spef_path] + output_paths = [ + output_dir / spef_path.name + for spef_path in (declared_paths or sorted(spef_writer_dir.glob("*.spef"))) + ] + + if not output_paths: + workspace.logger.error("RCX extraction produced no SPEF artifacts to publish") + return False + + # Validate the whole source set before touching any destination, so a + # partially copied set or a stale destination left by an earlier run can + # never pass for fresh extraction output. + for output_path in output_paths: + source_path = spef_writer_dir / output_path.name + if not (source_path.is_file() and os.path.getsize(source_path) > 0): + workspace.logger.error("RCX extraction artifact is missing or empty: %s", source_path) + return False + + processed: list[tuple[Path, Path | None]] = [] + temp_paths: list[Path] = [] + try: + for output_path in output_paths: + source_path = spef_writer_dir / output_path.name + output_path.parent.mkdir(parents=True, exist_ok=True) + backup_path = ( + output_path.with_name(f".{output_path.name}.prev") if output_path.exists() else None + ) + if backup_path is not None: + # Preserve the previously published SPEF so a failure later + # in this pass can restore the last known-good artifact. + output_path.replace(backup_path) + temp_path = output_path.with_name(f".{output_path.name}.tmp") + processed.append((output_path, backup_path)) + temp_paths.append(temp_path) + shutil.copy2(source_path, temp_path) + temp_path.replace(output_path) + workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) + + for output_path in output_paths: + if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0): + raise OSError(f"Published RCX SPEF is missing or empty: {output_path}") + except Exception as exc: + # Publication is all-or-nothing: a failed pass restores the SPEFs it + # replaced, removes the destinations it created, and drops partial + # temporary copies, leaving the step outputs exactly as before. + for output_path, backup_path in processed: + if backup_path is not None: + backup_path.replace(output_path) + else: + output_path.unlink(missing_ok=True) + for temp_path in temp_paths: + temp_path.unlink(missing_ok=True) + workspace.logger.error("Failed to publish RCX SPEF artifacts: %s", exc) + return False + + # The pass is committed: drop the preserved copies so a successful rerun + # does not accumulate a second SPEF set on disk. A removal failure must + # not fail the already-published step, so it is only reported. + for _output_path, backup_path in processed: + if backup_path is not None: + try: + backup_path.unlink(missing_ok=True) + except OSError as exc: + workspace.logger.warning("Failed to remove RCX SPEF backup: %s", exc) + + if isinstance(step.output.spef, list): + step.output.spef[:] = output_paths + return True diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index 8d4dd9c0..74bde795 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -18,6 +18,11 @@ save_rcx_spef_feature_facts, ) from chipcompiler.tools.ecc.module import ECCToolsModule +from chipcompiler.tools.ecc.rcx_artifacts import ( + copy_rcx_spef_outputs, + resolve_rcx_dirs, + wipe_stale_spef_artifacts, +) from chipcompiler.tools.ecc.sta_artifacts import discard_sta_outputs from chipcompiler.tools.ecc.sta_qor import ( POST_SYNTHESIS_STA_CORNER, @@ -65,49 +70,6 @@ def _workspace_sta_config_path(workspace: Workspace) -> str | None: return os.fspath(config_path) if config_path is not None else None -def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep): - data_dir_text = os.fspath(step.data.dir or "") - output_dir_text = os.fspath(step.output.dir or "") - workspace_dir = workspace.directory - if not data_dir_text or not output_dir_text or workspace_dir is None: - return - - data_dir = Path(data_dir_text) - if data_dir_text.startswith("/"): - relative_data_dir = data_dir_text[1:] - if relative_data_dir.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"): - data_dir = workspace_dir / relative_data_dir - - output_dir = Path(output_dir_text) - if output_dir_text.startswith("/"): - relative_output_dir = output_dir_text[1:] - if relative_output_dir.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"): - output_dir = workspace_dir / relative_output_dir - - spef_writer_dir = data_dir / "spef_writer" - if not spef_writer_dir.is_dir(): - return - - output_paths = [output_dir / spef_path.name for spef_path in step.output.spef if spef_path] - - if not output_paths: - output_paths = [ - output_dir / spef_path.name for spef_path in sorted(spef_writer_dir.glob("*.spef")) - ] - - for output_path in output_paths: - source_path = spef_writer_dir / output_path.name - if not source_path.is_file(): - continue - - output_path.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source_path, output_path) - workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) - - if isinstance(step.output.spef, list): - step.output.spef[:] = output_paths - - def copy_lvs_outputs(workspace: Workspace, step: EccStep): output_dir_text = os.fspath((step.data.steps or {}).get(StepEnum.LVS.value, "")) if not output_dir_text: @@ -841,12 +803,37 @@ def run_rcx(workspace: Workspace, step: EccStep, ecc_module: ECCToolsModule | No if ecc_module is not None: sub_flow.update_step(step_name=EccSubFlowEnum.load_data.value, state=StateEnum.Success) - ecc_module.init_rcx( - config=workspace.config.get(StepEnum.RCX.value, ""), pdk=workspace.pdk.name - ) - ecc_module.run_rcx() - ecc_module.destroy_rcx() - copy_rcx_spef_outputs(workspace, step) + # A rerun keeps the step directory, so drop previously extracted SPEFs + # first; stale artifacts left behind by an earlier run must not be + # mistaken for fresh extraction output. + data_dir, _ = resolve_rcx_dirs(workspace, step) + if data_dir is not None: + wipe_stale_spef_artifacts(data_dir) + + try: + if not ecc_module.init_rcx( + config=workspace.config.get(StepEnum.RCX.value, ""), pdk=workspace.pdk.name + ): + workspace.logger.error("Failed to initialize RCX extraction") + sub_flow.update_step( + step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete + ) + return False + if not ecc_module.run_rcx(): + workspace.logger.error("RCX extraction failed") + sub_flow.update_step( + step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete + ) + return False + finally: + try: + ecc_module.destroy_rcx() + except Exception as exc: + workspace.logger.error("Failed to release the RCX extractor: %s", exc) + + if not copy_rcx_spef_outputs(workspace, step): + sub_flow.update_step(step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete) + return False sub_flow.update_step(step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Success) if not save_data( diff --git a/test/test_engine_flow.py b/test/test_engine_flow.py index 6f2d78dd..d8ed989c 100644 --- a/test/test_engine_flow.py +++ b/test/test_engine_flow.py @@ -25,6 +25,16 @@ from chipcompiler.tools.ecc.signoff_checklist import refresh_step_checklist +class CompletionObserver: + """Record on_step_completed deliveries for assertions.""" + + def __init__(self): + self.completed = [] + + def on_step_completed(self, _step, state, error=None): + self.completed.append((state, error)) + + def test_engine_flow_missing_path_is_not_initialized(): engine_flow = EngineFlow(Workspace()) @@ -332,20 +342,43 @@ def test_rcx_fails_when_spef_missing(self, tmp_path): def test_rcx_succeeds_when_all_spef_exist(self, tmp_path): spef1 = tmp_path / "corner1.spef" spef2 = tmp_path / "corner2.spef" - spef1.write_text("") - spef2.write_text("") + spef1.write_text("*SPEF\n") + spef2.write_text("*SPEF\n") step = EccStep( name=StepEnum.RCX.value, output=EccOutput(spef=[spef1, spef2]), ) assert EngineFlow(Workspace()).check_step_result(step) is True - def test_rcx_succeeds_with_empty_spef_list(self): + def test_rcx_fails_with_empty_spef_list(self): step = EccStep( name=StepEnum.RCX.value, output=EccOutput(spef=[]), ) - assert EngineFlow(Workspace()).check_step_result(step) is True + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_spef_is_zero_byte(self, tmp_path): + spef = tmp_path / "corner1.spef" + spef.write_text("") + step = EccStep( + name=StepEnum.RCX.value, + output=EccOutput(spef=[spef]), + ) + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_one_of_two_spef_missing(self, tmp_path): + spef1 = tmp_path / "corner1.spef" + spef1.write_text("*SPEF\n") + spef2 = tmp_path / "corner2.spef" + step = EccStep( + name=StepEnum.RCX.value, + output=EccOutput(spef=[spef1, spef2]), + ) + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_output_is_not_ecc_output(self): + step = YosysStep(name=StepEnum.RCX.value, tool="yosys") + assert EngineFlow(Workspace()).check_step_result(step) is False class TestStepExceptionForcesIncomplete: @@ -586,6 +619,109 @@ def on_step_completed(self, _step, state, error=None): assert interrupted_step["peak memory (mb)"] >= 0 +class TestRunStepReturnContract: + """Regression: a non-success run_step() return must fail the step, never Success.""" + + @staticmethod + def _make_flow(tmp_path, name, tool, output): + workspace = Workspace() + workspace.flow.path = tmp_path / "flow.json" + flow_data = {"steps": [{"name": name, "tool": tool, "state": "Unstart"}]} + workspace.flow.path.write_text(json.dumps(flow_data), encoding="utf-8") + engine_flow = EngineFlow(workspace) + workspace_step = EccStep(name=name, directory=tmp_path, tool=tool, output=output) + engine_flow.workspace_steps = [workspace_step] + engine_flow.engine_db = SimpleNamespace(engine=None) + return engine_flow, workspace, workspace_step + + @staticmethod + def _valid_route_output(tmp_path): + output_dir = tmp_path / "output" + output_dir.mkdir() + for name in ("route.def", "route.v", "route.gds"): + (output_dir / name).write_text("content\n", encoding="utf-8") + return EccOutput( + def_=output_dir / "route.def", + verilog=output_dir / "route.v", + gds=output_dir / "route.gds", + ) + + @pytest.mark.parametrize( + "returned", + [ + False, + None, + StateEnum.Imcomplete, + StateEnum.Invalid, + StateEnum.Pending, + StateEnum.Unstart, + StateEnum.Ongoing, + ], + ) + def test_non_success_return_forces_incomplete(self, monkeypatch, tmp_path, returned): + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, "route", "ecc", self._valid_route_output(tmp_path) + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Imcomplete + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Imcomplete.value + assert observer.completed == [ + ( + StateEnum.Imcomplete, + f"route(ecc) reported failure (run_step returned {returned!r}).", + ) + ] + + @pytest.mark.parametrize("returned", [True, StateEnum.Success]) + def test_success_returns_reach_success(self, monkeypatch, tmp_path, returned): + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, "route", "ecc", self._valid_route_output(tmp_path) + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + monkeypatch.setattr(tools, "build_step_metrics", lambda **_kwargs: StepMetrics(data={})) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Success + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Success.value + assert observer.completed == [(StateEnum.Success, None)] + + @pytest.mark.parametrize("returned", [True, StateEnum.Success]) + def test_sizer_state_enum_return_keeps_success_contract(self, monkeypatch, tmp_path, returned): + output_dir = tmp_path / "output" + output_dir.mkdir() + def_path = output_dir / "sized.def" + verilog_path = output_dir / "sized.v" + def_path.write_text("VERSION 5.8 ;\nDESIGN gcd ;\nEND DESIGN\n", encoding="utf-8") + verilog_path.write_text("module gcd; endmodule\n", encoding="utf-8") + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, + StepEnum.TIMING_OPT.value, + "sizer", + EccOutput(def_=def_path, verilog=verilog_path), + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + monkeypatch.setattr(tools, "build_step_metrics", lambda **_kwargs: StepMetrics(data={})) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Success + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Success.value + assert observer.completed == [(StateEnum.Success, None)] + + class TestCreateStepFailureBreaksChain: """Regression: create_step(None) must break the flow chain and mark step Incomplete.""" diff --git a/test/tools/ecc/test_rcx_artifacts.py b/test/tools/ecc/test_rcx_artifacts.py new file mode 100644 index 00000000..10d7dde0 --- /dev/null +++ b/test/tools/ecc/test_rcx_artifacts.py @@ -0,0 +1,270 @@ +import errno +import shutil +from pathlib import Path + +import pytest + +from chipcompiler.data import EccData, EccOutput, EccStep, StepEnum, Workspace +from chipcompiler.tools.ecc import rcx_artifacts + + +class FakeLogger: + def __init__(self): + self.infos = [] + self.warnings = [] + self.errors = [] + + def info(self, message, *args): + self.infos.append((message, args)) + + def warning(self, message, *args): + self.warnings.append((message, args)) + + def error(self, message, *args): + self.errors.append((message, args)) + + +def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + source_path = data_dir / "spef_writer" / "gcd_Cworst_125C.spef" + source_path.parent.mkdir(parents=True) + source_path.write_text("*SPEF\n", encoding="utf-8") + spef_outputs = [data_dir / source_path.name] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is True + + destination = output_dir / source_path.name + assert destination.read_text(encoding="utf-8") == "*SPEF\n" + assert not (data_dir / source_path.name).exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [destination] + + +def test_copy_rcx_spef_outputs_fails_when_declared_spef_source_is_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "present.spef").write_text("*SPEF\n", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "missing.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + assert not (output_dir / "missing.spef").exists() + assert not (output_dir / "present.spef").exists() + assert step.output.spef == [output_dir / "missing.spef"] + + +def test_copy_rcx_spef_outputs_fails_when_source_spef_is_zero_byte(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "empty.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + assert not (output_dir / "empty.spef").exists() + assert step.output.spef == [output_dir / "empty.spef"] + + +def test_copy_rcx_spef_outputs_fails_and_keeps_stale_destination_when_source_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\nfresh\n", encoding="utf-8") + stale_destination = output_dir / "missing.spef" + output_dir.mkdir(parents=True) + stale_destination.write_text("*SPEF\nstale\n", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", stale_destination] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert stale_destination.read_text(encoding="utf-8") == "*SPEF\nstale\n" + assert not (output_dir / "good.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", stale_destination] + + +def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_empty(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\n", encoding="utf-8") + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", output_dir / "empty.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "good.spef").exists() + assert not (output_dir / "empty.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] + + +@pytest.mark.parametrize("partial_content", ["", "*SPEF\ntrunc"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails( + tmp_path, monkeypatch, partial_content +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + destination = Path(destination) + if destination.name == ".b.spef.tmp": + destination.write_text(partial_content, encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", fail_second_copy) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +@pytest.mark.parametrize("second_write", ["zero-byte", "skipped"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_validation_fails( + tmp_path, monkeypatch, second_write +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + def copy_with_invalid_second(source, destination, **_kwargs): + destination = Path(destination) + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.name == ".b.spef.tmp" and second_write == "zero-byte": + destination.write_text("", encoding="utf-8") + elif destination.name == ".a.spef.tmp": + destination.write_text("*SPEF\na\n", encoding="utf-8") + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", copy_with_invalid_second) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +def test_copy_rcx_spef_outputs_restores_previous_spef_when_a_later_copy_fails( + tmp_path, monkeypatch +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") + output_dir.mkdir(parents=True) + previous_a = output_dir / "a.spef" + previous_a.write_text("*SPEF\nprevious a\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + destination = Path(destination) + if destination.name == ".b.spef.tmp": + destination.write_text("*SPEF\ntrunc", encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", fail_second_copy) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert previous_a.read_text(encoding="utf-8") == "*SPEF\nprevious a\n" + assert not (output_dir / "b.spef").exists() + assert list(output_dir.iterdir()) == [previous_a] + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +def test_copy_rcx_spef_outputs_removes_backups_after_successful_republish(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") + output_dir.mkdir(parents=True) + (output_dir / "a.spef").write_text("*SPEF\nprevious a\n", encoding="utf-8") + (output_dir / "b.spef").write_text("*SPEF\nprevious b\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is True + + assert (output_dir / "a.spef").read_text(encoding="utf-8") == "*SPEF\nfresh a\n" + assert (output_dir / "b.spef").read_text(encoding="utf-8") == "*SPEF\nfresh b\n" + assert sorted(path.name for path in output_dir.iterdir()) == ["a.spef", "b.spef"] + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index ccb9a7ed..4def6b45 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -13,6 +13,7 @@ EccStep, OriginDesign, Parameters, + StateEnum, StepEnum, StepInput, Workspace, @@ -21,6 +22,7 @@ from chipcompiler.tools.ecc import runner as ecc_runner from chipcompiler.tools.ecc.builder import build_step, build_step_space from chipcompiler.tools.ecc.checklist import EccRcxChecklist +from chipcompiler.tools.ecc.subflow import EccSubFlowEnum class FakeEccModule: @@ -86,6 +88,7 @@ class FakeLogger: def __init__(self): self.infos = [] self.warnings = [] + self.errors = [] def info(self, message, *args): self.infos.append((message, args)) @@ -93,6 +96,9 @@ def info(self, message, *args): def warning(self, message, *args): self.warnings.append((message, args)) + def error(self, message, *args): + self.errors.append((message, args)) + class FakeSubFlow: def __init__(self, *args, **kwargs): @@ -125,6 +131,27 @@ def test_run_analysis_switch(parameters, expected_calls, tmp_path, monkeypatch): assert checklist.return_value.check.call_count == expected_calls +class FakeRcxModule: + def __init__(self, *, init_ok=True, run_ok=True): + self.calls = [] + self.init_ok = init_ok + self.run_ok = run_ok + + def update_step_paths(self, **kwargs): + self.calls.append(("update_step_paths", kwargs)) + + def init_rcx(self, **kwargs): + self.calls.append(("init_rcx", kwargs)) + return self.init_ok + + def run_rcx(self): + self.calls.append(("run_rcx",)) + return self.run_ok + + def destroy_rcx(self): + self.calls.append(("destroy_rcx",)) + + class FakeCtsModule: def __init__(self, timing_quality): self.calls = [] @@ -635,27 +662,118 @@ def test_sta_entry_workspace_uses_declared_spef(tmp_path): assert [item["spef_file"] for item in items] == [str(declared_spef)] -def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - source_path = data_dir / "spef_writer" / "gcd_Cworst_125C.spef" - source_path.parent.mkdir(parents=True) - source_path.write_text("*SPEF\n", encoding="utf-8") - spef_outputs = [data_dir / source_path.name] - step = EccStep( +def _make_rcx_step(tmp_path): + return EccStep( name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), + data=EccData(dir=tmp_path / "RCX_ecc" / "data"), + output=EccOutput(dir=tmp_path / "RCX_ecc" / "output"), ) + + +def test_run_rcx_propagates_init_failure(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule(init_ok=False) + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + call_names = [call[0] for call in module.calls] + assert "init_rcx" in call_names + assert "run_rcx" not in call_names + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_propagates_native_run_failure(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule(run_ok=False) + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + call_names = [call[0] for call in module.calls] + assert "init_rcx" in call_names + assert "run_rcx" in call_names + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_publishes_fresh_spef_and_releases_extractor_once(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + spef_writer = tmp_path / "RCX_ecc" / "data" / "spef_writer" + spef_writer.mkdir(parents=True) + module = FakeRcxModule() + + def run_native_extraction(): + module.calls.append(("run_rcx",)) + (spef_writer / "gcd_Cworst_125C.spef").write_text("*SPEF\n", encoding="utf-8") + return True + + module.run_rcx = run_native_extraction + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + monkeypatch.setattr(ecc_runner, "save_data", lambda **kwargs: True) + monkeypatch.setattr(ecc_runner, "save_rcx_spef_feature_facts", lambda **kwargs: True) + monkeypatch.setattr(ecc_runner, "run_analysis", lambda **kwargs: None) + + assert ecc_runner.run_rcx(workspace, step, module) is True + + published = tmp_path / "RCX_ecc" / "output" / "gcd_Cworst_125C.spef" + assert published.read_text(encoding="utf-8") == "*SPEF\n" + assert step.output.spef == [published] + call_names = [call[0] for call in module.calls] + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Success, + } in sub_flow.updates + + +def test_run_rcx_wipes_stale_spef_and_fails_when_extraction_produces_nothing(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + spef_writer = tmp_path / "RCX_ecc" / "data" / "spef_writer" + spef_writer.mkdir(parents=True) + stale_spef = spef_writer / "stale.spef" + stale_spef.write_text("*SPEF\nstale\n", encoding="utf-8") + module = FakeRcxModule() + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + assert not stale_spef.exists() + assert not (tmp_path / "RCX_ecc" / "output" / "stale.spef").exists() + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_fails_when_spef_writer_dir_is_missing(tmp_path, monkeypatch): workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule() + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) - ecc_runner.copy_rcx_spef_outputs(workspace, step) + assert ecc_runner.run_rcx(workspace, step, module) is False - destination = output_dir / source_path.name - assert destination.read_text(encoding="utf-8") == "*SPEF\n" - assert not (data_dir / source_path.name).exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [destination] + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates def test_run_sta_uses_matched_report_and_feature_corner_directories(tmp_path, monkeypatch):