Update phase state (thread-safe, atomic write). Args: phase: Current build/test phase test_name: Name of test being executed (optional) target: Compilation target (optional) path: Execution path ("sequential" or "streaming")
(
self,
phase: Literal["CONFIGURE", "COMPILE", "LINK", "EXECUTE"],
test_name: Optional[str] = None,
target: Optional[str] = None,
path: Literal["sequential", "streaming"] = "sequential",
)
| 55 | self._register_cleanup() |
| 56 | |
| 57 | def set_phase( |
| 58 | self, |
| 59 | phase: Literal["CONFIGURE", "COMPILE", "LINK", "EXECUTE"], |
| 60 | test_name: Optional[str] = None, |
| 61 | target: Optional[str] = None, |
| 62 | path: Literal["sequential", "streaming"] = "sequential", |
| 63 | ) -> None: |
| 64 | """ |
| 65 | Update phase state (thread-safe, atomic write). |
| 66 | |
| 67 | Args: |
| 68 | phase: Current build/test phase |
| 69 | test_name: Name of test being executed (optional) |
| 70 | target: Compilation target (optional) |
| 71 | path: Execution path ("sequential" or "streaming") |
| 72 | """ |
| 73 | with self._lock: |
| 74 | state = { |
| 75 | "phase": phase, |
| 76 | "test_name": test_name, |
| 77 | "target": target, |
| 78 | "timestamp": datetime.now().isoformat(), |
| 79 | "pid": os.getpid(), |
| 80 | "build_mode": self.build_mode, |
| 81 | "path": path, |
| 82 | } |
| 83 | |
| 84 | # Atomic write: write to temp file, then rename |
| 85 | temp_file = self.state_file.with_suffix(".tmp") |
| 86 | with open(temp_file, "w", encoding="utf-8") as f: |
| 87 | json.dump(state, f, indent=2) |
| 88 | |
| 89 | # Atomic rename (replaces existing file) |
| 90 | temp_file.replace(self.state_file) |
| 91 | |
| 92 | def get_phase(self) -> Optional[dict[str, Any]]: |
| 93 | """ |
no test coverage detected