#!/usr/bin/env python3 """Focused driver regression source; no Cargo or live SPT processes required.""" import copy import importlib.util import json from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch _spec = importlib.util.spec_from_file_location("local_lane", Path(__file__).with_name("local-lane.py")) lane = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(lane) # [unit->REQ-LOCAL-GATE-EXECUTION-EVIDENCE] class AllocationTests(unittest.TestCase): def test_admission_and_running_boundaries_are_distinct(self): gib = 1 << 30 self.assertFalse(lane.capacity_verdict(40*gib, 40*gib, 32*gib, 9*gib, initial=True)) self.assertTrue(lane.capacity_verdict(41*gib, 41*gib, 32*gib, 9*gib, initial=True)) self.assertTrue(lane.capacity_verdict(41*gib, 32*gib+1, 32*gib, 9*gib)) self.assertFalse(lane.capacity_verdict(41*gib, 32*gib, 32*gib, 9*gib)) # Growth can consume the allocation while free space remains above the floor. self.assertFalse(lane.capacity_verdict(50*gib, 41*gib, 32*gib, 9*gib)) def test_unbounded_timeout_and_inventory_bypass_are_refused(self): plan = {"reserve_bytes": 32 << 30, "max_growth_bytes": 9 << 30, "rustc_version": "allocated compiler", "steps": [ {"name": "suite", "timeout_seconds": 30, "nextest": {"packages": ["spt-daemon"]}}]} for timeout in (float("nan"), float("inf"), 0): invalid = copy.deepcopy(plan) invalid["steps"][0]["timeout_seconds"] = timeout with self.subTest(timeout=timeout), self.assertRaises(ValueError): lane.validate_plan(invalid) bypass = copy.deepcopy(plan) bypass["steps"][0] = {"name": "suite", "timeout_seconds": 30, "argv": ["cargo", "nextest", "run", "--workspace"]} with self.assertRaises(ValueError): lane.validate_plan(bypass) # [unit->REQ-LOCAL-GATE-EXECUTION-EVIDENCE] class InventoryTests(unittest.TestCase): def inventory(self): return {"test-count": 2, "rust-suites": {"fixture::integration": { "status": "listed", "binary-path": "/allocated/target/fixture", "testcases": { "selected_case": {"ignored": False, "filter-match": {"status": "matches"}}, "excluded_case": {"ignored": False, "filter-match": { "status": "mismatch", "reason": "expression"}}, }}}} def test_execution_inventory_excludes_filtered_cases(self): self.assertEqual(lane.selected_inventory(self.inventory()), [{ "binary_id": "fixture::integration", "binary_path": "/allocated/target/fixture", "tests": ["selected_case"]}]) def test_nonempty_suite_is_not_proof_of_selected_tests(self): inventory = self.inventory() cases = inventory["rust-suites"]["fixture::integration"]["testcases"] cases["selected_case"]["filter-match"] = {"status": "mismatch", "reason": "expression"} with self.assertRaises(ValueError): lane.selected_inventory(inventory) def test_incomplete_inventory_is_not_clearance(self): inventory = self.inventory() inventory["test-count"] = 3 with self.assertRaises(ValueError): lane.selected_inventory(inventory) inventory = self.inventory() del inventory["rust-suites"]["fixture::integration"]["testcases"]["selected_case"]["filter-match"] with self.assertRaises(ValueError): lane.selected_inventory(inventory) # [unit->REQ-LOCAL-GATE-EXECUTION-EVIDENCE] class ProducerTests(unittest.TestCase): def test_json_inventory_stderr_and_actual_failure_are_preserved(self): with tempfile.TemporaryDirectory() as directory: output = Path(directory) code, observed = lane.run_producer(output, output, "fixture", [sys.executable, "-c", "import sys; print('{\"test-count\": 0}'); print('producer diagnostic', file=sys.stderr); sys.exit(7)"], 30, None, lambda: True, json_stdout=True) self.assertEqual(code, 7) self.assertTrue(observed) self.assertEqual(json.loads((output / "fixture.stdout").read_text()), {"test-count": 0}) self.assertIn("producer diagnostic", (output / "fixture.raw").read_text()) self.assertEqual((output / "fixture.exit").read_text().strip(), "7") self.assertEqual(lane.final_status(code, False), 7) def test_cleanup_wait_failure_preserves_actual_producer_exit(self): with tempfile.TemporaryDirectory() as directory: output = Path(directory) with patch.object(lane.psutil, "wait_procs", side_effect=OSError("wait unavailable")): code, observed = lane.run_producer(output, output, "fixture", [sys.executable, "-c", "import sys; sys.exit(7)"], 30, None, lambda: True) self.assertEqual(code, 7) self.assertFalse(observed) self.assertEqual((output / "fixture.exit").read_text().strip(), "7") record = json.loads((output / "fixture.observation.json").read_text()) self.assertEqual(record["producer_exit"], 7) self.assertIn("wait unavailable", str(record["cleanup_errors"])) def test_capacity_refusal_stops_the_owned_producer(self): with tempfile.TemporaryDirectory() as directory: output = Path(directory) code, observed = lane.run_producer(output, output, "fixture", [sys.executable, "-c", "import time; time.sleep(60)"], 30, None, lambda: False) self.assertFalse(observed) self.assertNotEqual(code, 0) record = json.loads((output / "fixture.observation.json").read_text()) self.assertIsNotNone(record["producer_exit"]) self.assertEqual(record["survivors"], []) def test_disk_observation_failure_does_not_skip_cleanup_or_replace_failure(self): with tempfile.TemporaryDirectory() as directory: output = Path(directory) def cleanup(*args, **kwargs): (output / "cleanup-ran").write_text("complete") return True with patch.object(lane, "census", side_effect=cleanup), \ patch.object(lane, "builder_snapshot", return_value={"rows": [], "unreadable": []}): observed = lane.observe_after(output, output, "fixture", {}, lambda name: (_ for _ in ()).throw(OSError("disk unavailable"))) self.assertTrue((output / "cleanup-ran").is_file()) self.assertFalse(observed) self.assertEqual(lane.final_status(101, observed), 101) self.assertEqual(lane.final_status(0, observed), lane.REFUSED) def test_missing_or_unreadable_census_never_means_clear(self): self.assertFalse(lane.census_clear("", 0)) self.assertFalse(lane.census_clear( "CI-CENSUS phase=end box_procs=4 scoped=0 unscoped=3 unreadable_path=1\n", 0)) self.assertFalse(lane.census_clear( "CI-CENSUS phase=end box_procs=4 scoped=1 unscoped=3 unreadable_path=0\n", 0)) self.assertTrue(lane.census_clear( "CI-CENSUS phase=end box_procs=3 scoped=0 unscoped=3 unreadable_path=0\n", 0)) if __name__ == "__main__": unittest.main()