Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
121 kB · 2682 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683#!/usr/bin/env python3"""Structural checks for the local Tangled CI scaffold.
These tests never run Docker, never touch a workflow engine, and never contact aspindle, a device, or the network. They parse the workflow files with a smallstdlib-only YAML-subset loader and exercise ``tools/run_ci.py`` as a librarywith an injected command runner.
The loader supports only the subset the workflows use: block mappings, blocksequences, inline mappings on a sequence item, flow sequences, quoted and plainscalars, and booleans. There is no dependency on PyYAML."""
from __future__ import annotations
import contextlibimport ioimport itertoolsimport jsonimport osimport signalimport subprocessimport sysimport tempfileimport timeimport unittestfrom unittest import mockfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS))
import run_ci # noqa: E402
WORKFLOW_DIR = ROOT / ".tangled" / "workflows"WORKFLOW_FILES = { "smoke.yml": "smoke", "check.yml": "check", "emulator.yml": "emulator",}
# --------------------------------------------------------------------------# Minimal YAML-subset loader (no PyYAML)# --------------------------------------------------------------------------
def _clean_lines(text: str) -> list[tuple[int, str]]: lines: list[tuple[int, str]] = [] for raw in text.split("\n"): if not raw.strip(): continue body = raw.lstrip(" ") if body.startswith("#"): continue lines.append((len(raw) - len(body), body.rstrip())) return lines
def _try_split_kv(text: str): depth = 0 quote = None for index, char in enumerate(text): if quote: if char == quote: quote = None continue if char in "\"'": quote = char elif char in "[{": depth += 1 elif char in "]}": depth -= 1 elif char == ":" and depth == 0: if index + 1 == len(text) or text[index + 1] == " ": return text[:index].strip(), text[index + 1:].strip() return None, None
def _split_flow(inner: str) -> list[str]: parts: list[str] = [] depth = 0 quote = None current = "" for char in inner: if quote: current += char if char == quote: quote = None continue if char in "\"'": quote = char current += char elif char in "[{": depth += 1 current += char elif char in "]}": depth -= 1 current += char elif char == "," and depth == 0: parts.append(current.strip()) current = "" else: current += char if current.strip(): parts.append(current.strip()) return parts
def _scalar(text: str): text = text.strip() if text.startswith("[") and text.endswith("]"): inner = text[1:-1].strip() return [] if not inner else [_scalar(part) for part in _split_flow(inner)] if len(text) >= 2 and text[0] == text[-1] and text[0] in "\"'": return text[1:-1] lowered = text.lower() if lowered == "true": return True if lowered == "false": return False if lowered in ("null", "~"): return None return text
def _parse_node(lines, index, indent): if lines[index][1].startswith("- "): return _parse_seq(lines, index, indent) return _parse_map(lines, index, indent)
def _parse_map(lines, index, indent): result: dict = {} while index < len(lines): current_indent, text = lines[index] if current_indent < indent: break if current_indent > indent: raise ValueError(f"unexpected indent: {text!r}") if text.startswith("- "): break key, value = _try_split_kv(text) if key is None: raise ValueError(f"expected 'key: value', got {text!r}") if value != "": result[key] = _scalar(value) index += 1 elif index + 1 < len(lines) and lines[index + 1][0] > indent: child, index = _parse_node(lines, index + 1, lines[index + 1][0]) result[key] = child else: result[key] = None index += 1 return result, index
def _parse_seq(lines, index, indent): items: list = [] while index < len(lines): current_indent, text = lines[index] if current_indent != indent or not text.startswith("- "): break item = text[2:].strip() if item == "": if index + 1 < len(lines) and lines[index + 1][0] > indent: child, index = _parse_node(lines, index + 1, lines[index + 1][0]) items.append(child) else: items.append(None) index += 1 continue key, value = _try_split_kv(item) if key is None: items.append(_scalar(item)) index += 1 continue mapping: dict = {} if value != "": mapping[key] = _scalar(value) index += 1 elif index + 1 < len(lines) and lines[index + 1][0] > indent: child, index = _parse_node(lines, index + 1, lines[index + 1][0]) mapping[key] = child else: mapping[key] = None index += 1 if index < len(lines) and lines[index][0] > indent: more, index = _parse_map(lines, index, lines[index][0]) mapping.update(more) items.append(mapping) return items, index
def load_yaml(text: str): lines = _clean_lines(text) if not lines: return None value, index = _parse_node(lines, 0, lines[0][0]) if index != len(lines): raise ValueError(f"trailing content at line {index}") return value
# --------------------------------------------------------------------------# Test helpers# --------------------------------------------------------------------------
def workflow_text(name: str) -> str: return (WORKFLOW_DIR / name).read_text(encoding="utf-8")
def scripted_runner(statuses): """A runner whose nth call returns the nth status; extra calls return 0.""" calls: list[dict] = []
def runner(argv, *, env, cwd, timeout=None): calls.append({"argv": list(argv), "env": env, "cwd": cwd, "timeout": timeout}) position = len(calls) - 1 status = statuses[position] if position < len(statuses) else 0 return status, ""
return runner, calls
def flatten(commands): return [part for command in commands for part in command.argv]
def all_commands(): """Every command definition the helper can run, for blanket assertions.""" dummy = "/usr/bin/x64sc" commands = [] for action in run_ci.CONTAINER_ACTIONS: commands.extend(run_ci.container_steps(action)) commands.extend(run_ci.emulator_steps(dummy)) commands.extend(run_ci.runtime_steps("1234-abcd")) commands.extend(run_ci.release_steps()) return commands
# --------------------------------------------------------------------------# Loader self-test# --------------------------------------------------------------------------
class LoaderTests(unittest.TestCase): def test_inline_mapping_sequence(self): spec = load_yaml( "when:\n" ' - event: ["manual", "push"]\n' 'engine: "microvm"\n' ) self.assertEqual(spec["when"], [{"event": ["manual", "push"]}]) self.assertEqual(spec["engine"], "microvm")
def test_nested_mapping_and_sequence_of_mappings(self): spec = load_yaml( "steps:\n" ' - name: "one"\n' ' command: "do it"\n' "virtualisation:\n" " docker: true\n" ) self.assertEqual(spec["steps"], [{"name": "one", "command": "do it"}]) self.assertEqual(spec["virtualisation"], {"docker": True})
def test_comments_and_blank_lines_ignored(self): spec = load_yaml("# comment\n\nengine: x\n") self.assertEqual(spec, {"engine": "x"})
# --------------------------------------------------------------------------# Workflow specifications# --------------------------------------------------------------------------
class WorkflowSpecTests(unittest.TestCase): def test_all_workflow_files_exist(self): for name in WORKFLOW_FILES: self.assertTrue( (WORKFLOW_DIR / name).is_file(), f"missing workflow {name}" )
def test_files_are_valid_in_subset(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) self.assertIsInstance(spec, dict, name)
def test_engine_is_microvm(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) self.assertEqual(spec.get("engine"), "microvm", name)
def test_pilot_uses_observed_image_other_jobs_remain_gated(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) if name in ("smoke.yml", "check.yml"): self.assertEqual(spec.get("image"), "nixos") else: self.assertNotIn("image", spec, name)
def test_pilot_declares_only_required_guest_dependencies(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) if name in ("smoke.yml", "check.yml"): self.assertEqual(spec.get("dependencies"), ["python3", "git", "gnumake"]) else: self.assertNotIn("dependencies", spec, name)
def test_trigger_is_manual_only(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) when = spec.get("when") self.assertIsInstance(when, list, name) self.assertTrue(when, name) for condition in when: self.assertEqual(condition.get("event"), ["manual"], name) self.assertNotIn("branch", condition, name) self.assertNotIn("tag", condition, name) self.assertNotIn("schedule", condition, name) lowered = workflow_text(name).lower() self.assertNotIn("pull_request", lowered, name) self.assertNotIn("event: [\"push", lowered, name)
def test_docker_virtualisation_enabled(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) self.assertEqual(spec["virtualisation"], {"docker": True}, name)
def test_first_step_asserts_guest_preconditions(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) steps = spec.get("steps") self.assertIsInstance(steps, list, name) self.assertTrue(steps, name) self.assertEqual( steps[0].get("command"), "python3 tools/run_ci.py guest-check", name, )
def test_action_step_calls_the_helper(self): for name, action in WORKFLOW_FILES.items(): spec = load_yaml(workflow_text(name)) commands = [step.get("command", "") for step in spec["steps"]] self.assertTrue( any( command.startswith(f"python3 tools/run_ci.py {action}") for command in commands ), f"{name} must invoke the {action} action", )
def test_no_credentials_or_secrets(self): forbidden = ("c64u", "password", "token", "secret", "credential", "${{") for name in WORKFLOW_FILES: lowered = workflow_text(name).lower() for needle in forbidden: self.assertNotIn(needle, lowered, f"{name} contains {needle!r}")
def test_no_environment_block(self): for name in WORKFLOW_FILES: spec = load_yaml(workflow_text(name)) self.assertNotIn("environment", spec, name)
def test_no_publish_or_release_trigger_inside_workflows(self): for name in WORKFLOW_FILES: lowered = workflow_text(name).lower() self.assertNotIn("publish", lowered, name) self.assertNotIn("--mode release", lowered, name)
def test_no_physical_device_target(self): for name in WORKFLOW_FILES: self.assertNotIn("--target ultimate", workflow_text(name), name)
# --------------------------------------------------------------------------# run_ci.py as a library# --------------------------------------------------------------------------
class SafeEnvTests(unittest.TestCase): def test_allowlisted_variables_pass_through(self): env = run_ci.safe_env( {"PATH": "/bin", "HOME": "/root", "LANG": "C.UTF-8"} ) self.assertEqual(env["PATH"], "/bin") self.assertEqual(env["HOME"], "/root") self.assertEqual(env["CI"], "true")
def test_c64u_and_credential_variables_are_dropped(self): env = run_ci.safe_env( { "PATH": "/bin", "C64U_PASSWORD": "hunter2", "C64U_HOST": "192.168.8.64", "GITHUB_TOKEN": "ghp_x", "MY_SECRET": "s", "TANGLED_REPO_NAME": "ultimate-tool", } ) self.assertNotIn("C64U_PASSWORD", env) self.assertNotIn("C64U_HOST", env) self.assertNotIn("GITHUB_TOKEN", env) self.assertNotIn("MY_SECRET", env) self.assertEqual(env["TANGLED_REPO_NAME"], "ultimate-tool")
def test_personal_unrelated_variables_are_not_forwarded(self): env = run_ci.safe_env({"PATH": "/bin", "EDITOR": "vim", "USER": "david"}) self.assertNotIn("EDITOR", env) self.assertNotIn("USER", env)
class StepGuardTests(unittest.TestCase): def test_real_steps_are_accepted(self): run_ci.assert_safe_steps(run_ci.container_steps("smoke")) run_ci.assert_safe_steps(run_ci.container_steps("check")) run_ci.assert_safe_steps(run_ci.container_steps("package")) run_ci.assert_safe_steps(run_ci.emulator_steps("/usr/bin/x64sc")) run_ci.assert_safe_steps(run_ci.runtime_steps("1234-abcd"))
def test_every_command_has_a_bounded_timeout(self): for command in all_commands(): self.assertIsInstance(command.timeout_seconds, int) self.assertGreater(command.timeout_seconds, 0) # the shared helper budget is the absolute cap every ceiling nests in self.assertLessEqual(command.timeout_seconds, run_ci.CI_BUDGET_SECONDS)
def test_device_target_is_refused(self): with self.assertRaises(run_ci.ConfigError): run_ci.assert_safe_steps( [["ultimate-tool", "test", "smoke", "--target", "ultimate"]] )
def test_fake_and_vice_targets_are_allowed(self): run_ci.assert_safe_steps( [["ultimate-tool", "test", "smoke", "--target", "fake"]] ) run_ci.assert_safe_steps( [["ultimate-tool", "test", "smoke", "--target", "vice"]] )
def test_shell_string_command_is_refused(self): with self.assertRaises(run_ci.ConfigError): run_ci.assert_safe_steps(["make check-all && rm -rf /"])
def test_non_string_argument_is_refused(self): with self.assertRaises(run_ci.ConfigError): run_ci.assert_safe_steps([["echo", 1]])
def test_credentialish_token_is_refused(self): with self.assertRaises(run_ci.ConfigError): run_ci.assert_safe_steps([["curl", "-H", "Authorization: token x"]])
class StepExecutionTests(unittest.TestCase): def test_execute_steps_passes_env_and_timeout(self): runner, calls = scripted_runner([0, 0]) commands = [run_ci.cmd(["echo", "a"], 10), run_ci.cmd(["echo", "b"], 20)] run_ci.execute_steps(commands, runner, {"CI": "true"}) self.assertEqual(len(calls), 2) self.assertEqual(calls[0]["env"], {"CI": "true"}) self.assertEqual([call["timeout"] for call in calls], [10, 20])
def test_execute_steps_fails_closed_on_nonzero(self): runner, calls = scripted_runner([0, 1]) commands = [ run_ci.cmd(["echo", "a"], 10), run_ci.cmd(["echo", "b"], 10), run_ci.cmd(["echo", "c"], 10), ] with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.execute_steps(commands, runner, {"CI": "true"}) self.assertEqual(caught.exception.code, 1) self.assertEqual(len(calls), 2)
def test_timeout_status_is_a_failure(self): runner, _ = scripted_runner([run_ci.EXIT_TIMEOUT]) with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.execute_steps([run_ci.cmd(["slow"], 5)], runner, {"CI": "true"}) self.assertEqual(caught.exception.code, run_ci.EXIT_TIMEOUT)
def test_run_checked_reports_status(self): runner, _ = scripted_runner([3]) with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.run_checked(run_ci.cmd(["false"], 5), runner, {"CI": "true"}) self.assertEqual(caught.exception.code, 3)
class OutputRetainerTests(unittest.TestCase): """The bounded live echo and the bounded retained head and tail."""
def test_small_output_is_whole_and_echoed_exactly_once(self): sink = io.StringIO() retainer = run_ci.OutputRetainer(stream=sink, live_bytes=1000, head_bytes=1000, tail_bytes=1000) retainer.feed(b"hello world\n") retainer.finish() self.assertEqual(retainer.text(), "hello world\n") self.assertFalse(retainer.suppressed) retainer.report("stage") # the live echo and the diagnostic never both print the same line self.assertEqual(sink.getvalue().count("hello world"), 1) self.assertIn("all shown above", sink.getvalue())
def test_large_output_retains_a_bounded_head_and_tail(self): sink = io.StringIO() retainer = run_ci.OutputRetainer( stream=sink, live_bytes=200, head_bytes=400, tail_bytes=400 ) for _ in range(5000): retainer.feed(b"x" * 10) retainer.finish() text = retainer.text() self.assertLessEqual(len(text), 400 + 400 + 200) self.assertIn("truncated", text) self.assertTrue(text.startswith("x")) self.assertTrue(text.endswith("x")) self.assertEqual(retainer.total, 50000) self.assertEqual(retainer.echoed, 200) self.assertTrue(retainer.suppressed)
def test_diagnostic_only_prints_the_part_the_live_echo_did_not_show(self): sink = io.StringIO() retainer = run_ci.OutputRetainer( stream=sink, live_bytes=200, head_bytes=400, tail_bytes=400 ) for _ in range(5000): retainer.feed(b"x" * 10) retainer.finish() echoed = sink.getvalue() retainer.report("stage") diagnostic = sink.getvalue()[len(echoed):] # the retained tail is 400 bytes, the echo covered the first 200 self.assertIn("not shown live", diagnostic) self.assertLess(len(diagnostic), 400 + 400) # the dropped middle is reported, never silently lost self.assertIn("not shown", diagnostic)
def test_overlapping_head_and_tail_are_not_duplicated(self): retainer = run_ci.OutputRetainer(head_bytes=100, tail_bytes=100, live_bytes=10) retainer.feed(b"a" * 150) retainer.finish() self.assertEqual(retainer.dropped, 0) self.assertEqual(retainer.text(), "a" * 150)
def test_a_child_that_never_emits_a_newline_still_shows_progress(self): sink = io.StringIO() retainer = run_ci.OutputRetainer( stream=sink, live_bytes=100_000, head_bytes=1000, tail_bytes=1000 ) retainer.feed(b"y" * (run_ci.LIVE_LINE_FLUSH_BYTES * 2)) self.assertTrue(sink.getvalue()) self.assertTrue(sink.getvalue().startswith("y"))
def test_no_environment_is_ever_read_or_dumped(self): sink = io.StringIO() retainer = run_ci.OutputRetainer(stream=sink) retainer.feed(b"PATH=/bin\nC64U_PASSWORD=hunter2\n") retainer.finish() retainer.report("stage") # output is echoed as the child wrote it; the helper adds no environment self.assertNotIn("environ", sink.getvalue())
def _child_env() -> dict[str, str]: return {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "CI": "true"}
class StreamingRunnerTests(unittest.TestCase): """The real runner streams a child's output while it is still running."""
def test_short_child_output_is_streamed_before_the_child_exits(self): with tempfile.TemporaryDirectory() as tmp: marker = Path(tmp, "released") script = Path(tmp, "child.py") script.write_text( "import pathlib, sys, time\n" "marker = pathlib.Path(sys.argv[1])\n" "print('first', flush=True)\n" "deadline = time.monotonic() + 20\n" "while not marker.exists():\n" " if time.monotonic() > deadline:\n" " print('blocked', flush=True)\n" " raise SystemExit(9)\n" " time.sleep(0.01)\n" "print('second', flush=True)\n", encoding="utf-8", ) sink = _TriggerSink(marker, "first") with contextlib.redirect_stdout(sink): code, output = run_ci.default_runner( [sys.executable, "-u", str(script), str(marker)], env=_child_env(), cwd=ROOT, timeout=30, ) self.assertEqual(code, 0) # the child only prints "second" once the sink has seen "first"; if the # runner held output until exit the child would exit 9 instead self.assertNotIn("blocked", output) self.assertIn("first", output) self.assertIn("second", output) # each line reaches the log once: no replay at exit self.assertEqual(sink.getvalue().count("second"), 1)
def test_large_child_output_is_bounded_on_the_stream_and_in_memory(self): child = "for i in range(60000): print('line-%d-' % i + 'z' * 40)" sink = io.StringIO() with contextlib.redirect_stdout(sink): code, output = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=120, ) self.assertEqual(code, 0) live = len(sink.getvalue()) self.assertLessEqual( live, run_ci.LIVE_OUTPUT_BYTES + run_ci.RETAINED_TAIL_BYTES + run_ci.LIVE_LINE_FLUSH_BYTES + 4096, ) self.assertLessEqual( len(output), run_ci.RETAINED_HEAD_BYTES + run_ci.RETAINED_TAIL_BYTES + 256, ) self.assertIn("truncated", output) self.assertIn("not shown live", sink.getvalue())
def test_failure_status_and_output_are_returned(self): with contextlib.redirect_stdout(io.StringIO()): code, output = run_ci.default_runner( [sys.executable, "-u", "-c", "import sys; print('boom'); raise SystemExit(3)"], env=_child_env(), cwd=ROOT, timeout=30, ) self.assertEqual(code, 3) self.assertIn("boom", output)
def _resource_fixture(root: Path) -> Path: """A fake /proc/pressure + cgroup tree, so no test reads the real system.""" pressure = root / "pressure" cgroup = root / "cgroup" pressure.mkdir(parents=True, exist_ok=True) cgroup.mkdir(parents=True, exist_ok=True) (pressure / "cpu").write_text( "some avg10=0.15 avg60=0.02 avg300=0.00 total=98765\n", encoding="utf-8" ) (pressure / "io").write_text( "some avg10=1.00 avg60=2.00 avg300=3.00 total=111\n" "full avg10=0.50 avg60=0.60 avg300=0.70 total=99\n", encoding="utf-8", ) # A line the allowlist does not name, carrying text that must not be echoed. (pressure / "memory").write_text( "some avg10=0.00 avg60=0.00 avg300=0.00 total=5\n" "extra injected=hunter2\n", encoding="utf-8", ) (cgroup / "cpu.pressure").write_text( "some avg10=9.99 avg60=9.99 avg300=9.99 total=42\n", encoding="utf-8" ) (cgroup / "memory.events").write_text( "low 0\nhigh 1\nmax 2\noom 3\noom_kill 4\nunlisted 999\n", encoding="utf-8" ) (cgroup / "cpu.stat").write_text( "usage_usec 123456\nnr_periods 10\nnr_throttled 2\n" "throttled_usec 300\ncore_sched.force_idle_usec 7\n", encoding="utf-8", ) (cgroup / "memory.current").write_text("104857600\n", encoding="utf-8") (cgroup / "memory.max").write_text("max\n", encoding="utf-8") return root
def _fixture_probe(root: Path, **kwargs) -> run_ci.ResourceProbe: return run_ci.ResourceProbe( pressure_root=root / "pressure", cgroup_root=root / "cgroup", **kwargs )
class _ScriptedProbe: """A probe that hands out prepared snapshots, so both edges can be checked."""
def __init__(self, snapshots): self._snapshots = list(snapshots) self.calls = 0
def snapshot(self): snap = self._snapshots[min(self.calls, len(self._snapshots) - 1)] self.calls += 1 return snap
class ResourceSnapshotTests(unittest.TestCase): """The snapshots are fixed-path, bounded, read-only, and numeric-only."""
def test_only_the_allowlisted_fields_are_reported(self): with tempfile.TemporaryDirectory() as tmp: root = _resource_fixture(Path(tmp)) snapshot = _fixture_probe(root).snapshot() names = [name for name, _ in snapshot.values] self.assertIn("psi.cpu.some.avg10", names) self.assertIn("psi.io.full.total", names) self.assertIn("cgroup.cpu.pressure.some.total", names) self.assertIn("cgroup.memory.events.oom_kill", names) self.assertIn("cgroup.cpu.stat.throttled_usec", names) self.assertIn("cgroup.memory.current", names) self.assertEqual(snapshot.get("cgroup.memory.max"), "max") # nothing outside the allowlist, and no free text reaches the report self.assertNotIn("unlisted", names) self.assertNotIn("core_sched.force_idle_usec", str(names)) self.assertNotIn("hunter2", snapshot.summary()) self.assertNotIn("injected=", snapshot.summary())
def test_a_symlinked_source_is_not_followed(self): with tempfile.TemporaryDirectory() as tmp: root = _resource_fixture(Path(tmp)) # a plausible-looking link where a real cgroup file would be os.symlink(root / "pressure" / "cpu", root / "cgroup" / "io.pressure") snapshot = _fixture_probe(root).snapshot() names = [name for name, _ in snapshot.values] self.assertFalse([name for name in names if name.startswith("cgroup.io.pressure")]) # the real files are still read self.assertIn("psi.cpu.some.total", names)
def test_missing_sources_report_nothing_instead_of_raising(self): with tempfile.TemporaryDirectory() as tmp: empty = Path(tmp, "nothing") empty.mkdir() probe = run_ci.ResourceProbe( pressure_root=empty / "pressure", cgroup_root=empty / "cgroup" ) snapshot = probe.snapshot() self.assertEqual(snapshot.values, ()) self.assertEqual(snapshot.summary(), "") line = run_ci.format_elapsed( "make check-all", 1.0, 0, before=snapshot, after=snapshot ) self.assertIn("resources: unavailable", line) self.assertNotIn("delta:", line)
def test_a_capped_read_drops_the_incomplete_last_line(self): with tempfile.TemporaryDirectory() as tmp: root = _resource_fixture(Path(tmp)) (root / "pressure" / "cpu").write_text( "some avg10=0.15 avg60=0.02 avg300=0.00 total=98765\n" "full avg10=9.90 avg60=9.80 avg300=9.70 total=9\n", encoding="utf-8", ) # 80 bytes holds the first line and cuts the second one mid-value snapshot = _fixture_probe(root, budget_bytes=80).snapshot() self.assertEqual(snapshot.get("psi.cpu.some.total"), 98765) # the value the cap cut through is not reported as if it were written self.assertIsNone(snapshot.get("psi.cpu.full.total")) self.assertIsNone(snapshot.get("psi.cpu.full.avg300"))
def test_non_numeric_values_are_dropped(self): with tempfile.TemporaryDirectory() as tmp: root = _resource_fixture(Path(tmp)) (root / "pressure" / "cpu").write_text( "some avg10=0.15 avg60=hunter2 total=12\n", encoding="utf-8" ) (root / "cgroup" / "memory.current").write_text("secret\n", encoding="utf-8") snapshot = _fixture_probe(root).snapshot() self.assertEqual(snapshot.get("psi.cpu.some.avg10"), 0.15) self.assertEqual(snapshot.get("psi.cpu.some.total"), 12) self.assertIsNone(snapshot.get("psi.cpu.some.avg60")) self.assertIsNone(snapshot.get("cgroup.memory.current")) self.assertNotIn("hunter2", snapshot.summary())
def test_a_bounded_report_states_what_it_withheld(self): pairs = tuple((f"cgroup.cpu.stat.counter{index}", index) for index in range(40)) block = run_ci.format_resource_fields(pairs, limit=80) self.assertIn("fields not shown", block) self.assertTrue(block.startswith("cgroup.cpu.stat.counter0=0 ")) self.assertLessEqual(len(block), 80 + 40) self.assertNotIn("counter39", block)
def test_only_counters_are_differenced(self): before = run_ci.ResourceSnapshot( (("psi.cpu.some.avg10", 0.5), ("psi.cpu.some.total", 100), ("cgroup.memory.events.oom_kill", 0)) ) after = run_ci.ResourceSnapshot( (("psi.cpu.some.avg10", 9.0), ("psi.cpu.some.total", 640), ("cgroup.memory.events.oom_kill", 1), ("cgroup.cpu.stat.usage_usec", 50)) ) self.assertEqual( run_ci.resource_delta(before, after), (("psi.cpu.some.total", 540), ("cgroup.memory.events.oom_kill", 1)), )
def test_the_measured_stage_is_the_helpers_own_check_stage(self): self.assertTrue(run_ci.measured_stage(["make", "check-all"])) self.assertFalse(run_ci.measured_stage(["make", "check"])) self.assertFalse(run_ci.measured_stage(["npm", "ci"])) # the marker names a stage the helper actually defines, so it cannot drift check_argv = [command.argv for command in run_ci.container_steps("check")] self.assertIn(list(run_ci.MEASURED_ARGV[0]), check_argv)
class StageMeasurementTests(unittest.TestCase): """The runner prints one elapsed line per stage, and reads only when asked."""
def test_elapsed_and_both_edges_are_printed_once(self): before = run_ci.ResourceSnapshot((("psi.cpu.some.total", 100),)) after = run_ci.ResourceSnapshot((("psi.cpu.some.total", 140),)) probe = _ScriptedProbe([before, after]) sink = io.StringIO() with contextlib.redirect_stdout(sink): code, output = run_ci.default_runner( [sys.executable, "-u", "-c", "print('measured child')"], env=_child_env(), cwd=ROOT, timeout=60, measure=True, probe=probe, ) log = sink.getvalue() self.assertEqual(code, 0) self.assertIn("measured child", output) self.assertEqual(log.count("[run_ci] resources before"), 1) self.assertEqual(log.count("[run_ci] elapsed"), 1) self.assertEqual(probe.calls, 2) self.assertRegex(log, r"\[run_ci\] elapsed \d+\.\ds status=0 ") self.assertIn("psi.cpu.some.total=140", log) self.assertIn("delta: psi.cpu.some.total=+40", log)
def test_a_stage_that_is_not_measured_prints_only_its_elapsed_line(self): sink = io.StringIO() with mock.patch.object(run_ci, "ResourceProbe") as probe_class: with contextlib.redirect_stdout(sink): code, _ = run_ci.default_runner( [sys.executable, "-u", "-c", "print('plain child')"], env=_child_env(), cwd=ROOT, timeout=60, ) log = sink.getvalue() self.assertEqual(code, 0) # no global path is read for an ordinary stage probe_class.assert_not_called() self.assertIn("[run_ci] elapsed", log) self.assertNotIn("resources before", log) self.assertNotIn("resources:", log)
def test_a_failing_stage_reports_its_status_and_keeps_the_outcome_shape(self): sink = io.StringIO() with contextlib.redirect_stdout(sink): result = run_ci.default_runner( [sys.executable, "-u", "-c", "raise SystemExit(7)"], env=_child_env(), cwd=ROOT, timeout=60, ) self.assertIn("status=7", sink.getvalue()) # the reporting adds no field to what a caller already unpacks code, output = result self.assertEqual(code, 7) self.assertEqual(output, result.output)
def test_a_stopped_stage_says_so_on_its_elapsed_line(self): sink = io.StringIO() with contextlib.redirect_stdout(sink): code, _ = run_ci.default_runner( [sys.executable, "-u", "-c", "import time; time.sleep(60)"], env=_child_env(), cwd=ROOT, timeout=1, cleanup_seconds=15, ) self.assertEqual(code, run_ci.EXIT_TIMEOUT) self.assertIn("(timed out)", sink.getvalue()) self.assertIn(f"status={run_ci.EXIT_TIMEOUT}", sink.getvalue())
def test_the_snapshot_roots_are_fixed_constants_and_never_discovered(self): self.assertEqual(run_ci.RESOURCE_PRESSURE_ROOT, "/proc/pressure") self.assertEqual(run_ci.RESOURCE_CGROUP_ROOT, "/sys/fs/cgroup") self.assertLessEqual(run_ci.RESOURCE_READ_BYTES, 4096) self.assertLessEqual(run_ci.RESOURCE_REPORT_BYTES, 4096) source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") for discovery in ("os.walk", "os.listdir", "glob.glob", "Path.glob"): self.assertNotIn(discovery, source)
class _TriggerSink(io.StringIO): """A stdout sink that releases the child once its first line appears."""
def __init__(self, marker: Path, needle: str) -> None: super().__init__() self.marker = marker self.needle = needle self.released = False
def write(self, text: str) -> int: written = super().write(text) if not self.released and self.needle in self.getvalue(): self.released = True self.marker.write_text("released", encoding="utf-8") return written
class UnreapableChild: """A stand-in child that never exits and never gives up its exit status.
It models a process stuck in an uninterruptible sleep: it cannot be reaped, so nothing may claim it is gone. ``wait`` fails the test if it is ever called without a bound. """
def __init__(self, sink: io.StringIO) -> None: read_fd, write_fd = os.pipe() os.close(write_fd) # the pipe is already at end of file self.sink = sink self.stdout = os.fdopen(read_fd, "rb", buffering=0) self.pid = os.getpid() self.returncode = None self.signals: list[int] = [] self.bounds: list[float] = [] self.sink_at_first_signal: str | None = None
def poll(self): return None
def wait(self, timeout=None): assert timeout is not None, "wait() must always carry a bound" self.bounds.append(timeout) raise subprocess.TimeoutExpired("child", timeout)
def send_signal(self, signum): self.signals.append(signum) if self.sink_at_first_signal is None: self.sink_at_first_signal = self.sink.getvalue()
def kill(self): # pragma: no cover - only used if send_signal is unavailable self.signals.append(signal.SIGKILL)
# --------------------------------------------------------------------------# Bounded process-state probes for the cleanup checks## A killed child that nothing has adopted and reaped yet keeps its pid, so# ``kill(pid, 0)`` keeps succeeding while the task sits in state ``Z``. That task# holds no CPU and touches nothing, so ``/proc`` decides, and the states are# reported distinctly: a zombie is stopped, an unreadable status is not called# gone, and neither is confused with a live process.# --------------------------------------------------------------------------
PROC_ROOT = Path("/proc")STARTUP_SECONDS = 10 # bounded window for a real child to report it startedSTATE_SECONDS = 10 # bounded window for a stopped pid to stop executingCHILD_DEADLINE_SECONDS = 5 # a child's own budget: it contains interpreter startupSTATE_GONE = "gone" # no status entry at all: the task has exitedSTATE_UNKNOWN = "unknown" # the status could not be read: never assume it is goneZOMBIE_STATES = frozenset({"Z", "X", "x"})
def process_state(pid: int, *, root=None) -> str: """Return the kernel state letter for ``pid``, or a 'gone'/'unknown' marker.
``stat`` is ``pid (comm) state ...`` and ``comm`` may itself contain spaces and closing parentheses, so the state is read after the last ``)``. """ base = PROC_ROOT if root is None else Path(root) try: text = (base / str(int(pid)) / "stat").read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return STATE_GONE except OSError: return STATE_UNKNOWN try: return text[text.rindex(")") + 2] except (ValueError, IndexError): return STATE_UNKNOWN
def process_is_running(pid: int, *, root=None) -> bool: """True while ``pid`` can still execute; False once it exited or is a zombie.
A state that cannot be read counts as still running, because nothing observed says otherwise. Where ``/proc`` is not mounted the signal-0 probe is the only evidence available and is used as the fallback. """ base = PROC_ROOT if root is None else Path(root) if not (base / "self" / "stat").exists(): try: os.kill(int(pid), 0) except ProcessLookupError: return False except PermissionError: # pragma: no cover - not expected here return True return True state = process_state(pid, root=base) if state == STATE_GONE: return False return state not in ZOMBIE_STATES
def already_stopped(pid: int, *, root=None) -> bool: """True when a fresh read shows ``pid`` has already stopped by itself.
A pin can fail because the task exited and was reaped between the check above it and the pin itself, and such a task owes no signal. The pin failure alone never says that: it is the same ``None`` for a live pid this process may not pin. Only this read decides, and only a missing status entry or an unreaped zombie counts -- a state that cannot be read is not evidence of anything. """ base = PROC_ROOT if root is None else Path(root) if not (base / "self" / "stat").exists(): # Without ``/proc`` a missing entry proves nothing about the pid. return False state = process_state(pid, root=base) return state == STATE_GONE or state in ZOMBIE_STATES
def process_identity(pid: int, *, root=None): """Return the kernel identity of ``pid``, or None when it has no status entry.
``stat`` is ``pid (comm) state ppid pgrp session ... starttime ...`` and ``comm`` may contain spaces and parentheses, so every field is read after the last ``)``. ``start_ticks`` is the kernel start time in clock ticks since boot: together with ``pgrp`` it names one process for as long as the boot lasts, which a bare pid does not. """ base = PROC_ROOT if root is None else Path(root) try: text = (base / str(int(pid)) / "stat").read_text(encoding="utf-8", errors="replace") except OSError: return None try: fields = text[text.rindex(")") + 2:].split() return {"state": fields[0], "pgrp": int(fields[2]), "start_ticks": fields[19]} except (ValueError, IndexError): return None
def open_identity_fd(pid): """Pin ``pid`` to the process it names right now, or return None.
A pidfd follows its process rather than its pid, so a signal sent through it cannot reach a stranger that inherited the pid afterwards. """ opener = getattr(os, "pidfd_open", None) if opener is None: # pragma: no cover - kernels before 5.3 return None try: return opener(int(pid), 0) except OSError: return None
def signal_identity_fd(fd, signum) -> bool: """Send ``signum`` through a pidfd; True when the kernel accepted it.""" sender = getattr(signal, "pidfd_send_signal", None) if sender is None: # pragma: no cover - kernels before 5.3 return False try: sender(fd, signum) return True except OSError: return False
def wait_until(predicate, *, timeout: float, interval: float = 0.02) -> bool: """Poll ``predicate`` for a bounded window and return its last value.""" deadline = time.monotonic() + float(timeout) while True: if predicate(): return True if time.monotonic() >= deadline: return False time.sleep(interval)
def wait_until_not_running(pid: int, *, timeout: float = STATE_SECONDS, interval: float = 0.05, root=None) -> str: """Wait a bounded window for ``pid`` to stop executing; return its last state.""" wait_until(lambda: not process_is_running(pid, root=root), timeout=timeout, interval=interval) return process_state(pid, root=root)
def reported_tree(pid_file: Path): """Return the process tree the child reported, or None while it is incomplete.
Each line is ``role pid start_ticks pgrp``; the start ticks and group travel with the pid so a later cleanup can tell the reported process from an unrelated one that inherited the pid after it exited. """ try: text = pid_file.read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return None entries = {} for line in text.splitlines(): parts = line.split() if len(parts) != 4: return None role, pid, start_ticks, pgrp = parts if role not in ("child", "grand") or role in entries: return None if not (pid.isdigit() and start_ticks.isdigit() and pgrp.isdigit()): return None if int(pid) < 2: return None entries[role] = {"pid": int(pid), "start_ticks": start_ticks, "pgrp": int(pgrp)} if set(entries) != {"child", "grand"}: return None return {"pids": [entries["child"]["pid"], entries["grand"]["pid"]], "entries": entries}
def read_reported_pids(pid_file: Path) -> list[int] | None: """Return the pids a child reported, or None while that report is incomplete.""" tree = reported_tree(pid_file) return None if tree is None else tree["pids"]
def child_source(pid_file: Path, *, grand_pid=None) -> str: """Source for a child that reports identity for itself and its grandchild, then sleeps.
With ``grand_pid`` the child spawns nothing at all: the test owns that process and this child only records its identity, so ownership never depends on this child reaching the point where it writes the report. Without it the child starts its own grandchild in the same group, which a group signal covers. """ lines = [ "import os, pathlib, subprocess, sys, time\n", f"pid_file = pathlib.Path({str(pid_file)!r})\n", "def ident(pid):\n", " fields = pathlib.Path('/proc/%d/stat' % pid).read_text().rsplit(')', 1)[1].split()\n", " return fields[19], fields[2]\n", ] if grand_pid is None: lines.append("grand = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n") lines.append("grand_pid = grand.pid\n") else: lines.append(f"grand_pid = {int(grand_pid)}\n") lines.extend([ "here, there = ident(os.getpid()), ident(grand_pid)\n", "pid_file.write_text('child %d %s %s\\ngrand %d %s %s\\n'\n", " % (os.getpid(), here[0], here[1], grand_pid, there[0], there[1]))\n", "print('started', flush=True)\n", "time.sleep(60)\n", ]) return "".join(lines)
def paused_child_source() -> str: """Source for a child that stops itself before it can report anything.""" return "import os, signal, time\nos.kill(os.getpid(), signal.SIGSTOP)\ntime.sleep(60)\n"
class _UnkillableChild: """A real child presented with a process group the helper cannot reach.
``pid`` is outside any real pid range and ``send_signal`` reports delivery without delivering anything, so every stop the helper attempts lands nowhere while the real process tree keeps running. Everything else delegates to the real handle, so the production code path is unchanged. The regression below needs exactly that: a stop that fails after the child has reported its pids. """
def __init__(self, proc): self.proc = proc self.pid = 4194304 self.stdout = proc.stdout
@property def returncode(self): return self.proc.returncode
def wait(self, timeout=None): return self.proc.wait(timeout=timeout)
def poll(self): return self.proc.poll()
def send_signal(self, signum): return None
def stop_owned_tree(spawned, pid_file, *, timeout=STATE_SECONDS): """Stop whatever the helper left behind; return ``(confirmed, detail)``.
The captured handles are the ownership token. A leader is stopped through its own group only while it is still executing, and reaped because it is this process's own child; a group whose leader has been reaped is never signalled, because that group id can already belong to someone else.
A descendant is never stopped by group at all. It is signalled through a pidfd taken now, or as a single pid whose kernel start time and group still match what the child reported. A pid that cannot be identified, a signal that cannot be delivered, and a pid that outlives the bound are all reported as unconfirmed: nothing here quietly waits on a survivor and calls it clean. """ tree = reported_tree(pid_file) entries = {} if tree is None else tree["entries"] unconfirmed = [] if tree is None and not spawned: # No report and no handle: this call observed nothing it could verify, so # it may not claim the tree is gone. return False, "no report and no owned handle: nothing could be verified" for proc in spawned: if process_is_running(proc.pid): try: os.killpg(proc.pid, signal.SIGKILL) except OSError: pass for proc in spawned: try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: unconfirmed.append(f"the child {proc.pid} could not be reaped") leaders = {proc.pid for proc in spawned} for entry in entries.values(): pid = entry["pid"] if pid in leaders or not process_is_running(pid): continue # Pin the process first, then read the identity the kernel holds for the # pinned pid: a check that ran before the pin could be stale by the time # any signal was sent, and a plain pid signal could reach a stranger. fd = open_identity_fd(pid) if fd is None: # The pin also fails when the pid is reaped between the check above and # the pin itself, and a task that already stopped owes no signal. Any # other failure -- no pidfd support, a live pid this process may not # pin -- still needed a stop it did not get, so a fresh read showing it # already stopped is the only thing that confirms it. if not already_stopped(pid): unconfirmed.append(f"pid {pid} could not be pinned, so it was not signalled") continue try: current = process_identity(pid) if current is None: # The same exit, now between the pin and this read: a status entry # that is already gone owes no signal either. if not already_stopped(pid): unconfirmed.append(f"pid {pid} identity could not be read after pinning") continue if current["start_ticks"] != entry["start_ticks"] or current["pgrp"] != entry["pgrp"]: unconfirmed.append(f"pid {pid} is no longer the process the child reported") continue if not signal_identity_fd(fd, signal.SIGKILL): unconfirmed.append(f"pid {pid} could not be signalled through its pidfd") finally: os.close(fd) for pid in (tree["pids"] if tree else []): if process_is_running(pid): wait_until_not_running(pid, timeout=timeout) if process_is_running(pid): unconfirmed.append(f"pid {pid} was still executing {timeout:g}s after the cleanup") return not unconfirmed, "; ".join(unconfirmed)
class TimeoutCleanupTests(unittest.TestCase): """A bounded deadline terminates the whole child process group, once."""
def _spy_popen(self, spawned, wrap=None): """Return a patch target that captures the real handle before the helper sees it.""" real_popen = subprocess.Popen
def spawn(argv, **kwargs): proc = real_popen(argv, **kwargs) spawned.append(proc) return proc if wrap is None else wrap(proc)
return spawn
def test_timeout_kills_children_and_returns_a_bounded_failure(self): with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") child = child_source(pid_file) spawned = [] started = time.monotonic() # The handle is captured at spawn, so the teardown in ``finally`` owns # the tree even when an assertion fails before the pids are reported. with mock.patch.object(run_ci.subprocess, "Popen", self._spy_popen(spawned)): try: with contextlib.redirect_stdout(io.StringIO()) as sink: code, output = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=CHILD_DEADLINE_SECONDS, ) elapsed = time.monotonic() - started self.assertEqual(code, run_ci.EXIT_TIMEOUT) self.assertLess(elapsed, 30, "the deadline must return promptly") self.assertIn("started", output) self.assertIn("its process group was stopped", output) self.assertIn("its process group was stopped", sink.getvalue()) # The child reports both pids before it sleeps, so an # incomplete report is a startup problem to name rather than # to read past. self.assertTrue( wait_until(lambda: read_reported_pids(pid_file) is not None, timeout=STARTUP_SECONDS), f"the child never reported its pids (code={code}); output={output[-400:]!r}", ) child_pid, grandchild_pid = read_reported_pids(pid_file) self.assertNotEqual(child_pid, grandchild_pid) for pid in (child_pid, grandchild_pid): state = wait_until_not_running(pid) # A grandchild killed with the group is reparented and # stays a zombie until something adopts and reaps it. It # executes nothing, so it has not outlived the stop. self.assertFalse( process_is_running(pid), f"pid {pid} was still executing (state {state!r}) " f"{STATE_SECONDS:g}s after the deadline", ) finally: cleanup = stop_owned_tree(spawned, pid_file) # the teardown must be confirmed, not merely attempted self.assertTrue(cleanup[0], cleanup[1])
def test_the_owned_cleanup_stops_the_tree_when_the_production_stop_fails(self): """A stop that never reaches the group is still fully cleaned up.
The child reports its pids first; then the helper's group signals are redirected at a group that does not exist, so the tree survives the stop and the runner can only report it as unconfirmed. The assertions below state that; the teardown in ``finally`` must still leave nothing executing. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") child = child_source(pid_file) spawned = [] with mock.patch.object( run_ci.subprocess, "Popen", self._spy_popen(spawned, wrap=_UnkillableChild) ): try: with contextlib.redirect_stdout(io.StringIO()): outcome = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=1, cleanup_seconds=2, ) self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) # the stop could not reach the group, and nothing pretends it did self.assertTrue(outcome.cleanup_unconfirmed) self.assertTrue( wait_until(lambda: read_reported_pids(pid_file) is not None, timeout=STARTUP_SECONDS), "the child never reported its pids", ) child_pid, grandchild_pid = read_reported_pids(pid_file) self.assertTrue(process_is_running(child_pid), "the production stop reached the group") self.assertTrue(process_is_running(grandchild_pid), "the production stop reached the group") finally: cleanup = stop_owned_tree(spawned, pid_file) self.assertTrue(cleanup[0], cleanup[1]) for pid in read_reported_pids(pid_file): self.assertFalse(process_is_running(pid), f"pid {pid} survived the owned cleanup")
def _start_owned_sleep(self): """Start the process this test owns; its handle exists before anything else.""" return subprocess.Popen( [sys.executable, "-c", "import time; time.sleep(60)"], start_new_session=True )
def test_the_owned_cleanup_stops_an_owned_process_the_group_could_not_reach(self): """A process this test owns outlives the leader's group and is still stopped.
The test starts the detached sleep itself, so its handle exists before the child runs and ownership never depends on a report; the child only records its identity. The runner's stop reaches the child's group, which this process is not part of, so the cleanup has to stop it from the handle. """ detached = self._start_owned_sleep() spawned = [detached] try: with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") child = child_source(pid_file, grand_pid=detached.pid) with mock.patch.object(run_ci.subprocess, "Popen", self._spy_popen(spawned)): try: with contextlib.redirect_stdout(io.StringIO()): outcome = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=2, ) self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) self.assertFalse(outcome.cleanup_unconfirmed) self.assertTrue( wait_until(lambda: read_reported_pids(pid_file) is not None, timeout=STARTUP_SECONDS), "the child never reported its pids", ) child_pid, reported_pid = read_reported_pids(pid_file) self.assertEqual(reported_pid, detached.pid) self.assertFalse(process_is_running(child_pid), "the leader should have been stopped") self.assertTrue( process_is_running(detached.pid), "the owned process should have survived" ) # it leads its own group, so the leader's group signal never reached it self.assertEqual(process_identity(detached.pid)["pgrp"], detached.pid) finally: cleanup = stop_owned_tree(spawned, pid_file) self.assertTrue(cleanup[0], cleanup[1]) self.assertFalse(process_is_running(detached.pid), "the owned process outlived its cleanup") finally: if detached.poll() is None: detached.kill() detached.wait(timeout=STATE_SECONDS)
def test_the_owned_cleanup_stops_an_owned_process_when_the_child_never_reported(self): """A child that pauses before reporting leaves no report; ownership survives.
The child stops itself before it writes anything, so there is no identity to compare and no pid to read. The process this test owns is still stopped from the handle captured before the child ran, and the cleanup is never confirmed on the strength of a missing report alone. """ detached = self._start_owned_sleep() spawned = [detached] try: with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") child = paused_child_source() with mock.patch.object(run_ci.subprocess, "Popen", self._spy_popen(spawned)): try: with contextlib.redirect_stdout(io.StringIO()): outcome = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=2, ) self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) # the child was paused before it could report anything at all self.assertIsNone(read_reported_pids(pid_file)) self.assertTrue( process_is_running(detached.pid), "the owned process should have survived" ) finally: cleanup = stop_owned_tree(spawned, pid_file) self.assertTrue(cleanup[0], cleanup[1]) self.assertFalse(process_is_running(detached.pid), "the owned process outlived its cleanup") finally: if detached.poll() is None: detached.kill() detached.wait(timeout=STATE_SECONDS)
def test_a_child_that_closes_stdout_still_respects_the_deadline(self): child = ( "import os, time\n" "os.dup2(os.open(os.devnull, os.O_WRONLY), 1)\n" "time.sleep(60)\n" ) started = time.monotonic() with contextlib.redirect_stdout(io.StringIO()): code, _ = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=2, ) elapsed = time.monotonic() - started self.assertEqual(code, run_ci.EXIT_TIMEOUT) self.assertLess(elapsed, 30, "a closed pipe must not defeat the deadline")
def test_a_cooperative_child_is_stopped_on_sigterm_first(self): with tempfile.TemporaryDirectory() as tmp: marker = Path(tmp, "released") script = Path(tmp, "child.py") script.write_text( "import pathlib, signal, sys, time\n" "marker = pathlib.Path(sys.argv[1])\n" "def on_term(signum, frame):\n" " marker.write_text('released', encoding='utf-8')\n" " raise SystemExit(0)\n" "signal.signal(signal.SIGTERM, on_term)\n" "print('ready', flush=True)\n" "time.sleep(60)\n", encoding="utf-8", ) with contextlib.redirect_stdout(io.StringIO()): outcome = run_ci.default_runner( [sys.executable, "-u", str(script), str(marker)], env=_child_env(), cwd=ROOT, timeout=2, ) self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) self.assertFalse(outcome.cleanup_unconfirmed) self.assertIn("SIGTERM", outcome.detail) # the child ran its own release path: that is what the grace is for self.assertTrue(marker.is_file())
def test_a_child_that_ignores_sigterm_is_forced_and_confirmed(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) ready = Path(directory.name, "ready") child = ( "import pathlib, signal, sys, time\n" "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" "pathlib.Path(sys.argv[1]).write_text('ready', encoding='utf-8')\n" "time.sleep(60)\n" ) with contextlib.redirect_stdout(io.StringIO()): proc = subprocess.Popen( [sys.executable, "-u", "-c", child, str(ready)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) self.addCleanup(lambda: proc.poll() is None and proc.kill()) # Wait for the child's own report that the handler is installed. A fixed # sleep loses this race on a loaded runner, and losing it would look like # a child that never ignored SIGTERM. self.assertTrue( wait_until(ready.is_file, timeout=STARTUP_SECONDS), "the child never reported that it installed the SIGTERM handler", ) budget = run_ci.Deadline(30, label="test cleanup") confirmed, forced, detail = run_ci.stop_process_group(proc, budget=budget, grace=1, reap=5) self.assertTrue(confirmed) self.assertTrue(forced, "the escalation must be reported, not assumed") self.assertIn("SIGKILL", detail) self.assertIsNotNone(proc.poll())
def test_a_child_that_cannot_be_reaped_reports_cleanup_unconfirmed(self): sink = io.StringIO() created: list[UnreapableChild] = []
def factory(*args, **kwargs): created.append(UnreapableChild(sink)) return created[-1]
with mock.patch.object(run_ci.subprocess, "Popen", factory), \ mock.patch.object(run_ci.os, "getpgid", side_effect=ProcessLookupError), \ contextlib.redirect_stdout(sink): outcome = run_ci.default_runner(["/nonexistent-child"], env={}, cwd=ROOT, timeout=1) child = created[0] self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) self.assertTrue(outcome.cleanup_unconfirmed) self.assertIn("could NOT be confirmed", outcome.output) self.assertIn("may still exist", outcome.output) self.assertEqual(child.signals[:2], [signal.SIGTERM, signal.SIGKILL]) # the diagnostic was printed before the process group was signalled self.assertIn("exceeded its", child.sink_at_first_signal or "") # and nothing waited without a bound self.assertTrue(child.bounds) self.assertTrue(all(bound is not None for bound in child.bounds))
def test_a_failing_stage_reports_cleanup_unconfirmed_through_run_checked(self): def runner(argv, *, env, cwd, timeout=None): return run_ci.RunOutcome(run_ci.EXIT_TIMEOUT, "", True, "still running 5s after SIGKILL") with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.run_checked(run_ci.cmd(["slow"], 5), runner, {"CI": "true"}, stage="slow stage") self.assertTrue(caught.exception.cleanup_unconfirmed) self.assertIn("may still exist", str(caught.exception))
def test_no_wait_is_issued_without_a_bound(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn(".wait()", source) self.assertNotIn("wait(timeout=None)", source) self.assertIn("def wait_bounded", source)
def test_one_cleanup_deadline_bounds_every_step(self): # The per-step ceilings must fit inside the single deadline that covers # the stop, the drain and the container work. self.assertLessEqual( run_ci.CHILD_DRAIN_SECONDS + run_ci.CHILD_GRACE_SECONDS + run_ci.CHILD_REAP_SECONDS, run_ci.CLEANUP_RESERVE_SECONDS, ) self.assertLessEqual(run_ci.T_CONTAINER_TEARDOWN, run_ci.CLEANUP_RESERVE_SECONDS) self.assertGreater(run_ci.MIN_CLEANUP_SECONDS, 0)
def test_a_timed_out_command_spends_no_more_than_its_cleanup_deadline(self): child = ( "import signal, time\n" "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" "print('ready', flush=True)\n" "time.sleep(60)\n" ) started = time.monotonic() with contextlib.redirect_stdout(io.StringIO()): outcome = run_ci.default_runner( [sys.executable, "-u", "-c", child], env=_child_env(), cwd=ROOT, timeout=1, cleanup_seconds=3, ) elapsed = time.monotonic() - started self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) self.assertTrue(outcome.forced) self.assertFalse(outcome.cleanup_unconfirmed) self.assertIsNotNone(outcome.cleanup) self.assertLessEqual(outcome.cleanup.elapsed(), 3 + 0.5, "cleanup stayed inside its deadline") self.assertLess(elapsed, 1 + 3 + 5, "the whole command is bound plus one cleanup deadline")
def test_the_runner_does_not_retry_a_timed_out_stage(self): calls = [] def runner(argv, *, env, cwd, timeout=None): calls.append(timeout) return run_ci.EXIT_TIMEOUT, "" with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.execute_steps( [run_ci.cmd(["slow"], 5), run_ci.cmd(["later"], 5)], runner, {"CI": "true"}, ) self.assertTrue(caught.exception.timed_out) self.assertEqual(len(calls), 1)
class ProcessStateTests(unittest.TestCase): """The cleanup probes read the kernel state, not just whether a pid exists."""
def _proc_tree(self, directory, entries): """Build a stand-in /proc tree so the probes read real files.""" root = Path(directory) (root / "self").mkdir() (root / "self" / "stat").write_text("1 (python) S 0 1\n", encoding="utf-8") for pid, text in entries.items(): (root / str(pid)).mkdir() (root / str(pid) / "stat").write_text(text + "\n", encoding="utf-8") return root
def test_states_are_read_from_the_field_after_the_last_paren(self): with tempfile.TemporaryDirectory() as tmp: root = self._proc_tree(tmp, { 41: "41 (python) S 1 41 41", 42: "42 (python) Z 1 42 42", 43: "43 (worker ) Z) R 1 43 43", 44: "44 (dead) X 1 44 44", }) self.assertEqual(process_state(41, root=root), "S") self.assertEqual(process_state(42, root=root), "Z") # a comm containing ')' and a space must not shift the state letter self.assertEqual(process_state(43, root=root), "R") self.assertEqual(process_state(44, root=root), "X")
def test_a_zombie_has_stopped_even_though_its_pid_is_still_there(self): with tempfile.TemporaryDirectory() as tmp: root = self._proc_tree(tmp, { 41: "41 (python) S 1 41 41", 42: "42 (python) Z 1 42 42", }) self.assertTrue(process_is_running(41, root=root)) # kill(pid, 0) would still succeed here: the state letter decides self.assertFalse(process_is_running(42, root=root)) self.assertEqual(process_state(9999, root=root), STATE_GONE) self.assertFalse(process_is_running(9999, root=root))
def test_an_unreadable_state_is_not_called_gone(self): with tempfile.TemporaryDirectory() as tmp: root = self._proc_tree(tmp, {41: "not a stat line"}) self.assertEqual(process_state(41, root=root), STATE_UNKNOWN) self.assertTrue( process_is_running(41, root=root), "only a missing entry or a zombie state means stopped", )
def test_the_state_wait_is_bounded_and_reports_the_last_state(self): with tempfile.TemporaryDirectory() as tmp: root = self._proc_tree(tmp, {}) states = iter(["S", "S", "Z"]) with mock.patch.object(sys.modules[__name__], "process_state", lambda pid, **_: next(states, "Z")): self.assertEqual(wait_until_not_running(1, timeout=1, interval=0, root=root), "Z") started = time.monotonic() self.assertFalse(wait_until(lambda: False, timeout=0)) self.assertLess(time.monotonic() - started, 5) self.assertTrue(wait_until(lambda: True, timeout=0))
def test_an_incomplete_pid_report_is_not_trusted(self): with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") self.assertIsNone(read_reported_pids(pid_file)) pid_file.write_text("", encoding="utf-8") self.assertIsNone(read_reported_pids(pid_file)) # a report missing either process, or missing its identity, is unused pid_file.write_text("child 4321 91 4321\n", encoding="utf-8") self.assertIsNone(read_reported_pids(pid_file)) pid_file.write_text("child 4321 91\ngrand 4322 93\n", encoding="utf-8") self.assertIsNone(read_reported_pids(pid_file)) pid_file.write_text("child 4321 91 4321\ngrand 4322 93 4321\n", encoding="utf-8") self.assertEqual(read_reported_pids(pid_file), [4321, 4322]) # the identity travels with each pid tree = reported_tree(pid_file) self.assertEqual(tree["entries"]["grand"], {"pid": 4322, "start_ticks": "93", "pgrp": 4321})
def test_a_pid_that_no_longer_matches_the_report_is_never_signalled(self): """A reused pid is refused rather than killed: the identity has to match.
Both entries name this very process but claim a later start time, so a cleanup that signalled on the pid alone would kill the test run. The refusal has to be reported as unconfirmed instead. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") me = os.getpid() mine = process_identity(me) self.assertIsNotNone(mine) later = int(mine["start_ticks"]) + 1 pid_file.write_text( f"child {me} {later} {mine['pgrp']}\ngrand {me} {later} {mine['pgrp']}\n", encoding="utf-8", ) confirmed, detail = stop_owned_tree([], pid_file, timeout=0.2) self.assertFalse(confirmed) self.assertIn("no longer the process the child reported", detail) self.assertTrue(process_is_running(me), "a pid that cannot be identified is never signalled")
def test_an_identity_swapped_while_pinning_is_not_signalled(self): """The identity is read after the pin, so a swap is refused without a signal.
The pin itself succeeds, so only the identity read can show that the pid no longer names the reported process. Nothing may be signalled then, and the pinned descriptors have to be closed rather than leaked. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") me = os.getpid() mine = process_identity(me) later = int(mine["start_ticks"]) + 1 pid_file.write_text( f"child {me} {later} {mine['pgrp']}\ngrand {me} {later} {mine['pgrp']}\n", encoding="utf-8", ) pinned, signalled = [], []
def pin(pid): read_fd, write_fd = os.pipe() os.close(write_fd) pinned.append(read_fd) return read_fd
before = len(os.listdir("/proc/self/fd")) with mock.patch.object(sys.modules[__name__], "open_identity_fd", pin), \ mock.patch.object(sys.modules[__name__], "signal_identity_fd", lambda fd, sig: signalled.append((fd, sig)) or True): confirmed, detail = stop_owned_tree([], pid_file, timeout=0.2) self.assertFalse(confirmed) self.assertIn("no longer the process the child reported", detail) self.assertEqual(len(pinned), 2, "each reported pid is pinned once") self.assertEqual(signalled, [], "a pid whose identity changed is never signalled") self.assertEqual(len(os.listdir("/proc/self/fd")), before, "the pinned fds were closed, not leaked") self.assertTrue(process_is_running(me))
def test_a_pid_that_cannot_be_pinned_is_never_signalled_numerically(self): """Without a pidfd nothing proves which process the pid names, so nothing is sent.
The report matches this very process, so a numeric fallback signal would end the test run. The refusal has to be reported instead. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") me = os.getpid() mine = process_identity(me) pid_file.write_text( f"child {me} {mine['start_ticks']} {mine['pgrp']}\n" f"grand {me} {mine['start_ticks']} {mine['pgrp']}\n", encoding="utf-8", ) kills = [] with mock.patch.object(sys.modules[__name__], "open_identity_fd", lambda pid: None), \ mock.patch.object(os, "kill", lambda pid, sig: kills.append(pid)): confirmed, detail = stop_owned_tree([], pid_file, timeout=0.2) self.assertFalse(confirmed) self.assertIn("could not be pinned", detail) self.assertEqual(kills, [], "there is no numeric fallback signal") self.assertTrue(process_is_running(me))
def _reaping_window(self, pid_file, first_checks, *, state): """Report this very process, checked for a stop first and read as ``state``.
The report names this process because nothing is ever signalled here: a signal addressed to these pids would end the test run. ``first_checks`` True values put the pin inside the window where the pid still looks running, which is the window the cleanup has to survive. """ me = os.getpid() mine = process_identity(me) pid_file.write_text( f"child {me} {mine['start_ticks']} {mine['pgrp']}\n" f"grand {me} {mine['start_ticks']} {mine['pgrp']}\n", encoding="utf-8", ) checks = [True] * first_checks # after the scripted checks the pid keeps answering exactly as ``state`` # says, so the cleanup's own bounded confirming wait reads the same process live = state != STATE_GONE and state not in ZOMBIE_STATES signalled = []
def still_running(pid, *, root=None): # every pid still looks like it needs a stop until the checks run out, # which is when the cleanup makes its own confirming read return checks.pop(0) if checks else live
patches = ( mock.patch.object(sys.modules[__name__], "process_is_running", still_running), mock.patch.object(sys.modules[__name__], "open_identity_fd", lambda pid: None), mock.patch.object(sys.modules[__name__], "process_state", lambda pid, **kwargs: state), mock.patch.object( sys.modules[__name__], "signal_identity_fd", lambda fd, sig: signalled.append((fd, sig)) or True, ), ) return checks, signalled, patches
def test_a_pid_reaped_before_its_pin_is_confirmed_stopped(self): """A descendant that exits between the check and the pin needs no signal.
The group stop can reach the descendant and the descendant can be reaped in the window between "still running" and the pin. That pin fails exactly like an unpinnable live process, so only the fresh state read may decide -- and a pid with no status entry has stopped, which confirms the cleanup instead of reporting a survivor. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") checks, signalled, patches = self._reaping_window(pid_file, 2, state=STATE_GONE) with patches[0], patches[1], patches[2], patches[3]: confirmed, detail = stop_owned_tree([], pid_file, timeout=0.2) self.assertEqual(checks, [], "each reported pid was checked for a stop before the read") self.assertTrue(confirmed, detail) self.assertEqual(signalled, [], "a pid that already stopped is never signalled") self.assertTrue(process_is_running(os.getpid()))
def test_a_live_pid_that_cannot_be_pinned_stays_unconfirmed(self): """The same pin failure stays unconfirmed while the pid can still run.
Only the read after the pin separates the two cases. Here it reports a live process, so the exemption that covers an already reaped pid cannot be claimed for a survivor that simply could not be pinned. """ with tempfile.TemporaryDirectory() as tmp: pid_file = Path(tmp, "pids") checks, signalled, patches = self._reaping_window(pid_file, 2, state="S") with patches[0], patches[1], patches[2], patches[3]: confirmed, detail = stop_owned_tree([], pid_file, timeout=0.2) self.assertEqual(checks, [], "each reported pid was checked for a stop before the read") self.assertFalse(confirmed) self.assertIn("could not be pinned", detail) self.assertIn("was still executing", detail) self.assertEqual(signalled, [], "there is no numeric fallback signal") self.assertTrue(process_is_running(os.getpid()))
def test_no_report_and_no_handle_cannot_be_called_clean(self): """Nothing observed means nothing confirmed, even though nothing was left.""" with tempfile.TemporaryDirectory() as tmp: confirmed, detail = stop_owned_tree([], Path(tmp, "missing.pid"), timeout=0.2) self.assertFalse(confirmed) self.assertIn("nothing could be verified", detail)
class FakeClock: def __init__(self, start: float = 0.0) -> None: self.value = start
def __call__(self) -> float: return self.value
def advance(self, seconds: float) -> None: self.value += max(0.0, float(seconds))
def cleanup_budget(clock=None, seconds: int = run_ci.CLEANUP_RESERVE_SECONDS): """The single cleanup deadline a test hands to a teardown.""" return run_ci.Deadline(seconds, clock=clock or time.monotonic, label="test cleanup")
class ClockedChild: """A fake child that consumes the injected clock and can ignore SIGTERM.
``wait`` always takes a bound and spends part of it, exactly like a real command against a slow daemon. ``returncode=None`` means the child ignores SIGTERM and only stops when SIGKILL arrives. """
def __init__(self, clock, *, output: bytes = b"", exit_code=0, wait_cost: float = 0.0) -> None: read_fd, write_fd = os.pipe() if output: os.write(write_fd, output) os.close(write_fd) self.clock = clock self.stdout = os.fdopen(read_fd, "rb", buffering=0) self.pid = os.getpid() self.returncode = exit_code self.wait_cost = wait_cost self.signals: list[int] = [] self.waits: list[object] = []
def poll(self): return self.returncode
def wait(self, timeout=None): assert timeout is not None, "wait() must always carry a bound" offered = float(timeout) self.waits.append(offered) if self.returncode is None: self.clock.advance(offered) # still running: it uses what it was given raise subprocess.TimeoutExpired("child", timeout) self.clock.advance(min(offered, self.wait_cost)) return self.returncode
def send_signal(self, signum): self.signals.append(signum) if signum == signal.SIGKILL: self.returncode = -signal.SIGKILL self.wait_cost = 2.0 # a killed child still takes a moment to reap
class FakeDaemon: """A fake Docker daemon whose containers outlive the client that made them.
``visible_after_seconds`` models a client killed while the daemon was still creating the container: nothing is listable until that much simulated time has passed, so a single probe right after the kill sees nothing. ``call_seconds`` charges every successful list/remove call against the injected clock, the way a slow daemon consumes a real deadline. """
def __init__(self, *, clock=time.monotonic, advance=None, visible_after_seconds: float = 0.0, run_status: int = 0, call_seconds: float = 0.0) -> None: self.clock = clock self.advance = advance or (lambda seconds: None) self.visible_after_seconds = visible_after_seconds self.run_status = run_status self.call_seconds = call_seconds self.containers: dict[str, dict] = {} self.calls: list[tuple[list[str], object]] = [] self.removed: list[str] = [] self.selectors: list[str] = []
def start(self, selector: str, name: str = "container") -> str: """Register a container that exists independently of any client.""" identifier = os.urandom(8).hex() self.containers[identifier] = { "name": name, "label": selector, "visible_at": self.clock() + self.visible_after_seconds, } return identifier
def _charge(self) -> None: if self.call_seconds: self.advance(self.call_seconds)
# -- the fake docker CLI --------------------------------------------- def runner(self, argv, *, env, cwd, timeout=None): argv = list(argv) self.calls.append((argv, timeout)) if "ps" in argv: selector = argv[argv.index("--filter") + 1][len("label="):] self.selectors.append(selector) listed = "".join( f"{identifier}\n" for identifier, info in self.containers.items() if info["label"] == selector and self.clock() >= info["visible_at"] ) self._charge() return 0, listed if "rm" in argv: identifiers = argv[argv.index("--force") + 1:] status = 0 for identifier in identifiers: if identifier in self.containers: self.removed.append(self.containers.pop(identifier)["name"]) else: status = 1 self._charge() return status, "" if "run" in argv: name = argv[argv.index("--name") + 1] labels = [argv[index + 1] for index, value in enumerate(argv) if value == "--label"] self.start(labels[0], name) return self.run_status, "" return 0, ""
def stages(self): """The stage commands, without the teardown probes.""" return [(argv, timeout) for argv, timeout in self.calls if "ps" not in argv]
class ContainerOwnershipTests(unittest.TestCase): """M1: killing the client must not orphan the daemon's container."""
def test_the_container_is_created_with_a_unique_name_and_label(self): first = run_ci.new_ownership_token() second = run_ci.new_ownership_token() self.assertNotEqual(first, second) name = f"{run_ci.CONTAINER_NAME_PREFIX}-check-{first}" argv = run_ci.docker_run_argv( "check", 600, name=name, labels=[f"{run_ci.CONTAINER_LABEL_KEY}={first}", f"{run_ci.CONTAINER_STAGE_LABEL_KEY}=check"], ) self.assertEqual(argv[argv.index("--name") + 1], name) labels = [argv[index + 1] for index, value in enumerate(argv) if value == "--label"] self.assertIn(f"{run_ci.CONTAINER_LABEL_KEY}={first}", labels) self.assertIn(f"{run_ci.CONTAINER_STAGE_LABEL_KEY}=check", labels) # the ownership marks are this invocation's, not a fixed name other = run_ci.docker_run_argv( "check", 600, name=f"{run_ci.CONTAINER_NAME_PREFIX}-check-{second}", labels=[f"{run_ci.CONTAINER_LABEL_KEY}={second}"], ) self.assertNotEqual(argv[argv.index("--name") + 1], other[other.index("--name") + 1])
def test_the_two_label_shapes_are_allowlisted_and_the_runtime_one_is_key_only(self): token = "1234-abcd" self.assertEqual(run_ci.ci_owned_label(token).selector(), f"{run_ci.CONTAINER_LABEL_KEY}={token}") # the harness labels its containers with a key only, not key=value self.assertEqual( run_ci.runtime_owned_label(token).selector(), f"{run_ci.RUNTIME_LABEL_KEY_PREFIX}.{token}", ) with self.assertRaises(run_ci.ConfigError): run_ci.OwnedLabel("some-other-project", token) with self.assertRaises(run_ci.ConfigError): # an unsafe token is refused before any docker command is built run_ci.ci_owned_label("../../etc")
def test_the_runtime_label_prefix_is_the_harness_constant_and_the_token_is_valid(self): # The source builder changes its imported harness's label namespace. # Load this contract independently, as a runtime subprocess would. import importlib.util
name = "runtime_ownership_contract_under_test" spec = importlib.util.spec_from_file_location(name, TOOLS / "verify_release_local.py") harness = importlib.util.module_from_spec(spec) with mock.patch.dict(sys.modules, {name: harness}): spec.loader.exec_module(harness) self.assertEqual(run_ci.RUNTIME_LABEL_KEY_PREFIX, harness.CONTAINER_LABEL) token = run_ci.new_ownership_token() self.assertEqual(harness.validate_invocation_id(token), token)
def test_teardown_removes_exactly_the_container_this_invocation_labelled(self): daemon = FakeDaemon() token = "1234-abcd" daemon.start(run_ci.ci_owned_label(token).selector(), "owned") # another invocation's container must survive untouched daemon.start(run_ci.ci_owned_label("9999-ffff").selector(), "other") result = run_ci.teardown_owned_containers( run_ci.ci_owned_label(token), runner=daemon.runner, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "removed") self.assertEqual(daemon.removed, ["owned"]) self.assertEqual(len(daemon.containers), 1, "only the owned container may be removed") self.assertEqual([info["name"] for info in daemon.containers.values()], ["other"]) self.assertEqual(daemon.selectors[0], run_ci.ci_owned_label(token).selector())
def test_teardown_never_prunes_and_only_addresses_identifiers_it_was_given(self): daemon = FakeDaemon() token = "5555-eeee" daemon.start(run_ci.ci_owned_label(token).selector(), "owned") run_ci.teardown_owned_containers( run_ci.ci_owned_label(token), runner=daemon.runner, env={}, budget=cleanup_budget() ) for argv, _ in daemon.calls: self.assertNotIn("prune", argv) self.assertNotIn("system", argv) self.assertNotIn("container", argv) for part in argv: self.assertNotIn("$(", part) self.assertNotIn("*", part) listing = [argv for argv, _ in daemon.calls if "ps" in argv][0] self.assertIn("--no-trunc", listing, "full identifiers, so removal names the same container")
def test_teardown_follows_a_delayed_start_inside_the_same_deadline(self): clock = FakeClock() # the daemon registers the container 35s after the client was killed daemon = FakeDaemon(clock=clock, visible_after_seconds=35) token = "7777-1111" daemon.start(run_ci.ci_owned_label(token).selector(), "late") budget = cleanup_budget(clock) result = run_ci.teardown_owned_containers( run_ci.ci_owned_label(token), runner=daemon.runner, env={}, budget=budget, settle=True, sleep=clock.advance, ) self.assertEqual(result.outcome, "removed") self.assertGreaterEqual(clock.value, 35, "the observation covered a delayed start") self.assertLessEqual(budget.elapsed(), run_ci.CLEANUP_RESERVE_SECONDS) self.assertEqual(daemon.containers, {})
def test_teardown_reports_absent_without_touching_anything(self): daemon = FakeDaemon() result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("nothing-here"), runner=daemon.runner, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "absent") self.assertTrue(result.confirmed) self.assertEqual([argv for argv, _ in daemon.calls if "rm" in argv], [])
def test_a_listing_that_cannot_be_trusted_is_reported_not_used(self): def unreachable(argv, *, env, cwd, timeout=None): return 1, "Cannot connect to the Docker daemon"
def garbage(argv, *, env, cwd, timeout=None): return 0, "all\n"
def truncated(argv, *, env, cwd, timeout=None): return run_ci.RunOutcome(0, "aabbccddee", truncated=True)
for runner, expected in ( (unreachable, "could not list"), (garbage, "not a container identifier"), (truncated, "was truncated"), ): seen = []
def recording(argv, *, env, cwd, timeout=None, _inner=runner): seen.append(list(argv)) return _inner(argv, env=env, cwd=cwd, timeout=timeout)
result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("x1"), runner=recording, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "unconfirmed", expected) self.assertIn(expected, result.detail) self.assertEqual([argv for argv in seen if "rm" in argv], [], "nothing removed on a guess")
def test_a_container_that_survives_its_removal_is_reported_not_chased(self): calls = []
def runner(argv, *, env, cwd, timeout=None): calls.append(list(argv)) if "ps" in argv: return 0, "aabbccddeeff\n" # the daemon keeps listing it return 0, "" result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "unconfirmed") self.assertIn("still listed after removal", result.detail) self.assertEqual(len([argv for argv in calls if "rm" in argv]), 1, "no chase loop")
def test_an_ambiguous_removal_result_is_not_retried(self): def runner(argv, *, env, cwd, timeout=None): if "ps" in argv: return 0, "aabbccddeeff\n" return 1, "Error response from daemon: No such container" # who removed it? result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "unconfirmed") self.assertIn("ambiguous removal result", result.detail) self.assertIn("not retried", result.detail)
def test_a_daemon_that_keeps_minting_containers_stops_at_the_round_cap(self): counter = itertools.count()
def runner(argv, *, env, cwd, timeout=None): if "ps" in argv: return 0, f"{next(counter):015x}\n" return 0, "" result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() ) self.assertEqual(result.outcome, "unconfirmed") self.assertTrue( "teardown rounds" in result.detail or "deadline" in result.detail, result.detail )
def test_a_nested_runner_timeout_cannot_extend_the_shared_cleanup_deadline(self): # The teardown's docker calls run through the real default_runner. A # nested timeout there must spend what is left of the shared deadline, # not open a fresh 60s allowance of its own. clock = FakeClock() budget = run_ci.Deadline(45, clock=clock, label="partly spent cleanup") created: list[tuple[list[str], ClockedChild]] = []
def factory(argv, **kwargs): argv = list(argv) if "ps" in argv: child = ClockedChild(clock, output=b"abcdef012345\n", exit_code=0, wait_cost=20) elif "rm" in argv: child = ClockedChild(clock, output=b"", exit_code=None) # ignores SIGTERM else: # pragma: no cover - no other docker call is expected child = ClockedChild(clock, output=b"", exit_code=0) created.append((argv, child)) return child
with mock.patch.object(run_ci.subprocess, "Popen", factory), \ mock.patch.object(run_ci.os, "getpgid", side_effect=ProcessLookupError), \ contextlib.redirect_stdout(io.StringIO()): result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("tok123"), runner=run_ci.default_runner, env={}, budget=budget, settle=True, ) self.assertEqual(result.outcome, "unconfirmed") self.assertIn("not retried", result.detail) removal = [child for argv, child in created if "rm" in argv] self.assertEqual(len(removal), 1) child = removal[0] self.assertIn(signal.SIGTERM, child.signals) self.assertIn(signal.SIGKILL, child.signals, "a TERM-ignoring client needs the kill") # every nested wait was bounded, and the whole cleanup ended at the # deadline it was handed instead of overshooting it self.assertTrue(all(wait is not None for wait in child.waits)) self.assertLessEqual( clock.value, budget.total_seconds, "the nested stop must end at the shared cleanup deadline", )
def test_slow_docker_calls_are_bounded_by_the_one_cleanup_deadline(self): clock = FakeClock() # every successful ps/rm costs 25s of the 60s deadline daemon = FakeDaemon(clock=clock, advance=clock.advance, call_seconds=25) counter = itertools.count() budget = cleanup_budget(clock)
def runner(argv, *, env, cwd, timeout=None): # each listing returns a fresh container, so nothing is ever settled if "ps" in argv: return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)[0], f"{next(counter):016x}\n" return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)
result = run_ci.teardown_owned_containers( run_ci.ci_owned_label("x"), runner=runner, env={}, budget=budget, settle=True ) self.assertEqual(result.outcome, "unconfirmed") self.assertLessEqual(budget.elapsed(), run_ci.CLEANUP_RESERVE_SECONDS, "cleanup never outlives its deadline") # no docker call was ever given more than the per-call ceiling for _argv, timeout in daemon.calls: self.assertLessEqual(timeout, run_ci.T_CONTAINER_TEARDOWN)
def test_a_killed_client_still_loses_its_container(self): daemon = FakeDaemon(run_status=run_ci.EXIT_TIMEOUT, visible_after_seconds=0) with mock.patch.object(run_ci, "require_docker", return_value=None), \ mock.patch.object(run_ci, "TEARDOWN_POLL_SECONDS", 0), \ contextlib.redirect_stdout(io.StringIO()) as sink: with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.run_container("check", daemon.runner, {"PATH": "/bin"}) self.assertTrue(caught.exception.timed_out) self.assertEqual(daemon.containers, {}, "the daemon-owned container must be removed") self.assertEqual(len(daemon.removed), 1) self.assertIn("container teardown", sink.getvalue())
def test_unconfirmed_container_cleanup_is_reported_not_hidden(self): def runner(argv, *, env, cwd, timeout=None): if "build" in argv: return 0, "" if "run" in argv: return 0, "" return 1, "Cannot connect to the Docker daemon" with mock.patch.object(run_ci, "require_docker", return_value=None), \ contextlib.redirect_stdout(io.StringIO()) as sink: with self.assertRaises(run_ci.CleanupUnconfirmed) as caught: run_ci.run_container("smoke", runner, {"PATH": "/bin"}) self.assertIn("cleanup unconfirmed", str(caught.exception)) self.assertIn("not removed by a broad prune", str(caught.exception)) self.assertIn("unconfirmed", sink.getvalue())
def test_a_cooperative_timeout_with_no_container_listed_is_still_unconfirmed(self): # EXIT_TIMEOUT with a confirmed, unforced stop: the process fact is clean # but the resource fact is not, so an empty listing must not be called # absent. The settle stays bounded by the cleanup deadline. daemon = FakeDaemon()
def runner(argv, *, env, cwd, timeout=None): if any(part.endswith("verify_release_local.py") for part in argv): return run_ci.RunOutcome( run_ci.EXIT_TIMEOUT, "", False, "stopped on SIGTERM within 10s of the cleanup deadline", run_ci.Deadline(2, label="harness cleanup"), # already nearly spent ) return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)
with contextlib.redirect_stdout(io.StringIO()) as sink: with self.assertRaises(run_ci.CommandFailed) as caught: run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) self.assertTrue(caught.exception.timed_out) self.assertFalse(caught.exception.cleanup_unconfirmed) self.assertFalse(caught.exception.forced) self.assertIn("unconfirmed", sink.getvalue()) self.assertIn("a container may still appear", sink.getvalue())
def test_the_runtime_harness_container_is_removed_by_its_exact_label_after_a_kill(self): daemon = FakeDaemon() selectors = []
def runner(argv, *, env, cwd, timeout=None): if any(part.endswith("verify_release_local.py") for part in argv): # the harness started a container and was then killed at its bound self.assertIn("--external-container-cleanup", argv) token = argv[argv.index("--invocation-id") + 1] selector = f"{run_ci.RUNTIME_LABEL_KEY_PREFIX}.{token}" selectors.append(selector) daemon.start(selector, "verify-run") return run_ci.RunOutcome( run_ci.EXIT_TIMEOUT, "", False, "stopped on SIGTERM within 10s of the cleanup deadline", run_ci.Deadline(run_ci.CLEANUP_RESERVE_SECONDS, label="harness cleanup"), ) return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)
with contextlib.redirect_stdout(io.StringIO()) as sink: with self.assertRaises(run_ci.CommandFailed): run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) self.assertEqual(daemon.removed, ["verify-run"], "the harness container must be removed exactly") self.assertEqual(daemon.containers, {}) # the harness's containers are addressed by its own key-only label self.assertEqual( set(s for s in daemon.selectors if s.startswith(run_ci.RUNTIME_LABEL_KEY_PREFIX)), set(selectors), ) self.assertIn("runtime container teardown", sink.getvalue())
def test_a_reaped_harness_with_no_visible_container_is_not_called_clean(self): daemon = FakeDaemon()
def runner(argv, *, env, cwd, timeout=None): if any(part.endswith("verify_release_local.py") for part in argv): # the process was reaped, but it died before the daemon created # anything: that is not proof that nothing will appear return run_ci.RunOutcome( run_ci.EXIT_TIMEOUT, "", False, "stopped on SIGKILL within 5s", run_ci.Deadline(2, label="harness cleanup"), # already nearly spent False, True, # forced ) return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)
with contextlib.redirect_stdout(io.StringIO()) as sink: with self.assertRaises(run_ci.CommandFailed): run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) self.assertIn("unconfirmed", sink.getvalue()) self.assertIn("a container may still appear", sink.getvalue())
class BudgetTests(unittest.TestCase): """The action's budget is one monotonic allowance, never reset by a stage."""
def test_budget_stays_inside_the_workflow_step_ceiling(self): self.assertEqual( run_ci.CI_BUDGET_SECONDS + run_ci.WORKFLOW_RESERVE_SECONDS, run_ci.WORKFLOW_STEP_SECONDS, ) # the temporary qualification allowance: 90 minutes of microVM step time, # of which the helper spends at most 80 and 600s stay for the guest self.assertEqual(run_ci.WORKFLOW_STEP_SECONDS, 5400) self.assertEqual(run_ci.WORKFLOW_RESERVE_SECONDS, 600) self.assertEqual(run_ci.CI_BUDGET_SECONDS, 4800) self.assertLess(run_ci.CI_BUDGET_SECONDS, run_ci.WORKFLOW_STEP_SECONDS)
def test_every_declared_ceiling_fits_the_shared_budget(self): # the previously inconsistent ceilings all nested inside the step ceiling for command in all_commands(): self.assertLessEqual( command.timeout_seconds, run_ci.CI_BUDGET_SECONDS, f"{command.note or command.argv[0]} declares more than the budget", )
def test_remaining_time_caps_every_command_and_is_never_reset(self): ticks = [0.0]
def clock(): return ticks[0]
deadline = run_ci.Deadline(100, clock=clock) runner, calls = scripted_runner([0, 0, 0]) run_ci.run_checked(run_ci.cmd(["a"], 1000), runner, {}, deadline=deadline) ticks[0] = 60.0 run_ci.run_checked(run_ci.cmd(["b"], 1000), runner, {}, deadline=deadline) ticks[0] = 94.0 run_ci.run_checked(run_ci.cmd(["c"], 7), runner, {}, deadline=deadline) self.assertEqual([call["timeout"] for call in calls], [100, 40, 6])
def test_expired_budget_stops_before_the_next_stage_and_says_so(self): ticks = [0.0]
def clock(): return ticks[0]
deadline = run_ci.Deadline(100, clock=clock) runner, calls = scripted_runner([0]) run_ci.run_checked( run_ci.cmd(["a"], 100), runner, {}, deadline=deadline, stage="first" ) ticks[0] = 150.0 with self.assertRaises(run_ci.BudgetExpired) as caught: run_ci.run_checked( run_ci.cmd(["b"], 100), runner, {}, deadline=deadline, stage="second" ) message = str(caught.exception) self.assertIn("second", message) self.assertIn("bounded failure", message) self.assertIn("separately approved", message) self.assertEqual(len(calls), 1, "no stage starts after the budget expires")
def test_later_stages_draw_on_the_same_budget_as_earlier_ones(self): ticks = [0.0]
def clock(): return ticks[0]
deadline = run_ci.Deadline(900, clock=clock) timeouts = []
def runner(argv, *, env, cwd, timeout=None): # stage commands only; the container teardown also calls docker if "ps" in argv or "rm" in argv: return 0, "" timeouts.append(timeout) ticks[0] += 200 # every stage consumes budget return 0, ""
with mock.patch.object(run_ci, "require_docker", return_value=None), \ contextlib.redirect_stdout(io.StringIO()): run_ci.run_check(runner, {"PATH": "/bin"}, deadline=deadline) # docker build, docker run, then the runtime harness self.assertEqual(len(timeouts), 3, timeouts) self.assertGreater(timeouts[0], timeouts[1]) self.assertGreater(timeouts[1], timeouts[2])
def test_container_handoff_is_shorter_than_the_outer_timeout(self): seen = []
def runner(argv, *, env, cwd, timeout=None): seen.append((list(argv), timeout)) return 0, ""
with mock.patch.object(run_ci, "require_docker", return_value=None), \ contextlib.redirect_stdout(io.StringIO()): run_ci.run_container("smoke", runner, {"PATH": "/bin"}) run_argv, run_timeout = seen[1] inner = int(run_argv[run_argv.index("--budget-seconds") + 1]) self.assertEqual(inner, run_timeout - run_ci.CONTAINER_HANDOFF_SECONDS) self.assertLess(inner, run_timeout)
def test_a_direct_run_cannot_raise_the_budget(self): parser = run_ci.build_parser() args = parser.parse_args(["check", "--budget-seconds", "7200"]) self.assertEqual(run_ci.action_budget(args.budget_seconds), run_ci.CI_BUDGET_SECONDS) self.assertEqual(run_ci.action_budget(None), run_ci.CI_BUDGET_SECONDS) self.assertEqual(run_ci.action_budget(600), 600)
def test_the_gate_keeps_every_stage(self): check_flat = flatten(run_ci.container_steps("check")) for token in ("npm", "ci", "check-all", "demo", "tools/package_release.py", "tools/check_release.py"): self.assertIn(token, check_flat) seen = []
def runner(argv, *, env, cwd, timeout=None): seen.append(list(argv)) return 0, ""
with mock.patch.object(run_ci, "require_docker", return_value=None), \ contextlib.redirect_stdout(io.StringIO()): run_ci.run_check(runner, {"PATH": "/bin"}) joined = " ".join(" ".join(argv) for argv in seen) # the gate runs inside the container, and the runtime harness still runs self.assertIn("_container", joined) self.assertIn("check", seen[1]) self.assertIn("tools/verify_release_local.py", joined)
class CommandDefinitionTests(unittest.TestCase): def test_container_steps_are_argv_arrays(self): for action in run_ci.CONTAINER_ACTIONS: for command in run_ci.container_steps(action): self.assertIsInstance(command.argv, list) self.assertTrue(command.argv) self.assertTrue(all(isinstance(part, str) for part in command.argv))
def test_container_steps_never_invoke_docker(self): for action in run_ci.CONTAINER_ACTIONS: self.assertNotIn("docker", flatten(run_ci.container_steps(action)))
def test_smoke_builds_and_runs_the_first_run_fake_inside_the_image(self): flat = flatten(run_ci.container_steps("smoke")) self.assertIn("build", flat) self.assertIn("validate", flat) self.assertIn("examples/first-run/c64dev.toml", flat) self.assertIn("tools/run_demo.py", flat) self.assertNotIn("ultimate", flat)
def test_check_has_explicit_npm_ci_stage(self): flat = flatten(run_ci.container_steps("check")) self.assertIn("npm", flat) self.assertIn("ci", flat) self.assertIn("check-all", flat)
def test_check_runs_the_release_path_without_make_dist(self): commands = run_ci.container_steps("check") flat = flatten(commands) # direct tool invocations, not the Docker-backed `make dist` self.assertIn("tools/package_release.py", flat) self.assertIn("tools/check_release.py", flat) self.assertIn("--prepare-source", flat) self.assertIn("--pack", flat) for argv in (command.argv for command in commands): for index, token in enumerate(argv): if token == "make": # only the local gate targets, never the Docker-backed ones self.assertIn(argv[index + 1], ("check-all", "demo"))
def test_build_env_descriptor_is_outside_dist(self): self.assertTrue(run_ci.BUILD_ENV_DESCRIPTOR.startswith("/tmp/")) self.assertNotIn("dist", run_ci.BUILD_ENV_DESCRIPTOR)
def test_release_build_is_locked(self): flat = flatten(run_ci.release_steps()) self.assertIn("--locked", flat) self.assertIn("--release", flat) self.assertIn("--target-dir", flat) self.assertIn("--build-env", flat) self.assertIn("--binary-dir", flat)
def test_binary_dir_matches_the_cargo_target_dir(self): self.assertEqual( run_ci.RELEASE_BINARY_DIR, f"{run_ci.CARGO_TARGET_DIR}/release" ) flat = flatten(run_ci.release_steps()) self.assertIn(run_ci.CARGO_TARGET_DIR, flat) self.assertIn(run_ci.RELEASE_BINARY_DIR, flat)
def test_emulator_passes_the_executable_flag_not_env(self): commands = run_ci.emulator_steps("/usr/bin/x64sc") self.assertEqual(len(commands), 1) argv = commands[0].argv self.assertEqual(argv[:2], ["python3", "tools/vice_debugging_and_checkpoint_acceptance.py"]) self.assertIn("--vice-executable", argv) self.assertEqual(argv[argv.index("--vice-executable") + 1], "/usr/bin/x64sc")
def test_runtime_harness_is_wired(self): token = "run-123-abc" flat = flatten(run_ci.runtime_steps(token)) self.assertIn("tools/verify_release_local.py", flat) self.assertIn("runtime", flat) # the helper is the sole remover and names itself, so no docker rm race self.assertIn("--external-container-cleanup", flat) self.assertEqual(flat[flat.index("--invocation-id") + 1], token) self.assertIn( run_ci.RUNTIME_LABEL_KEY_PREFIX, run_ci.runtime_owned_label(token).selector(), )
def test_no_command_uses_release_preflight_mode(self): for command in all_commands(): argv = command.argv for index, token in enumerate(argv): if token == "--mode": self.assertNotEqual(argv[index + 1], "release")
def test_docker_run_has_no_host_socket_privileges_or_root(self): argv = run_ci.docker_run_argv("check") joined = " ".join(argv) self.assertIn("--rm", argv) self.assertEqual(argv[:4], ["docker", "--host", "unix:///var/run/docker.sock", "run"]) self.assertNotIn("--privileged", argv) self.assertEqual(argv[argv.index("--user") + 1], f"{os.getuid()}:{os.getgid()}") mounts = [argv[index + 1] for index, value in enumerate(argv[:-1]) if value in ("-v", "--volume", "--mount")] self.assertFalse(any("docker.sock" in mount for mount in mounts)) self.assertIn("/work", joined)
def test_docker_build_context_contains_exact_copy_inputs(self): argv = run_ci.docker_build_argv() self.assertEqual(argv[:4], ["docker", "--host", "unix:///var/run/docker.sock", "build"]) self.assertNotIn("BUILD_UID", " ".join(argv)) self.assertEqual(Path(argv[-1]), ROOT) self.assertEqual((ROOT / Path(argv[argv.index("-f") + 1])).resolve(), ROOT / "tools/release-env/Dockerfile") for relative in ("package.json", "package-lock.json", "tools/release-env/ultimate-tool-build-env", ".dockerignore"): self.assertTrue((Path(argv[-1]) / relative).is_file(), relative)
class RuntimeEnvironmentTests(unittest.TestCase): def test_runtime_uses_empty_owned_home_not_guest_docker_context(self): seen = [] def runner(argv, *, env, cwd, timeout=None): home = Path(env["HOME"]) self.assertTrue(home.is_dir()) self.assertEqual(list(home.iterdir()), []) seen.append(home) return 0, "" original = {"HOME": "/existing-guest-home"} run_ci.run_runtime_steps(runner, original) self.assertTrue(seen) self.assertFalse(seen[0].exists()) self.assertEqual(original["HOME"], "/existing-guest-home")
class ContainerToolchainEnvironmentTests(unittest.TestCase): def test_fixed_image_paths_and_owned_home_override_caller(self): homes = [] def runner(argv, *, env, cwd, timeout=None): home = Path(env["HOME"]) self.assertTrue(home.is_dir()) self.assertEqual(home.stat().st_uid, os.getuid()) self.assertEqual(env["CARGO_HOME"], "/usr/local/cargo") self.assertEqual(env["RUSTUP_HOME"], "/usr/local/rustup") self.assertEqual(env["PLAYWRIGHT_BROWSERS_PATH"], "/opt/ms-playwright") self.assertEqual(env["XDG_CACHE_HOME"], str(home / ".cache")) homes.append(home) return 0, "" original = {"HOME": "/home/builder", "RUSTUP_HOME": "/untrusted", "CARGO_HOME": "/untrusted"} run_ci.run_container_steps("smoke", runner, original) self.assertTrue(homes) self.assertEqual(len(set(homes)), 1) self.assertFalse(homes[0].exists()) self.assertEqual(original["HOME"], "/home/builder") self.assertEqual(original["RUSTUP_HOME"], "/untrusted")
class GuestPreconditionTests(unittest.TestCase): def test_parse_os_release(self): fields = run_ci.parse_os_release( 'NAME=NixOS\nID=nixos\nVERSION_ID="24.11"\n' ) self.assertEqual(fields["ID"], "nixos") self.assertEqual(fields["VERSION_ID"], "24.11")
def test_guest_problems_empty_when_expected(self): facts = {"os_id": "nixos", "machine": "x86_64", "uid": 0} self.assertEqual(run_ci.guest_problems(facts, True), [])
def test_guest_problems_reports_each_mismatch(self): facts = {"os_id": "ubuntu", "machine": "aarch64", "uid": 1000} problems = run_ci.guest_problems(facts, False) self.assertEqual(len(problems), 3)
def test_nonroot_workflow_user_is_supported(self): facts = {"os_id": "nixos", "machine": "x86_64", "uid": 970} self.assertEqual(run_ci.guest_problems(facts, True), []) runner = mock.Mock(return_value=(0, "29.6.2")) with mock.patch.object(run_ci, "read_guest_facts", return_value=facts), mock.patch.object(run_ci, "docker_present", return_value=True): run_ci.verify_guest({}, runner) args, kwargs = runner.call_args self.assertEqual(args[0], ["docker", "--host", "unix:///var/run/docker.sock", "info", "--format", "{{.ServerVersion}}"]) self.assertEqual(kwargs["timeout"], 15)
def test_guest_docker_access_failure_stops_before_build(self): facts = {"os_id": "nixos", "machine": "x86_64", "uid": 970} runner = mock.Mock(return_value=(1, "permission denied")) with mock.patch.object(run_ci, "read_guest_facts", return_value=facts), mock.patch.object(run_ci, "docker_present", return_value=True): with self.assertRaisesRegex(run_ci.PreconditionError, "unavailable to the workflow user"): run_ci.run_guest_check(runner, {}) self.assertEqual(runner.call_count, 1)
def test_wrong_guest_does_not_query_docker(self): runner = mock.Mock() with mock.patch.object(run_ci, "read_guest_facts", return_value={"os_id": "ubuntu", "machine": "x86_64", "uid": 970}), mock.patch.object(run_ci, "docker_present", return_value=True): with self.assertRaises(run_ci.PreconditionError): run_ci.verify_guest({}, runner) runner.assert_not_called()
class SourceManifestTests(unittest.TestCase): def test_archive_includes_exact_ci_test_and_context_inputs(self): sys.path.insert(0, str(TOOLS)) import package_release manifest = package_release.load_source_manifest(TOOLS / "release_source_manifest.json") files, _ = package_release.collect_source_files(ROOT, manifest) paths = {entry["path"] for entry in files} for name in WORKFLOW_FILES: self.assertIn(f".tangled/workflows/{name}", paths) self.assertIn(".dockerignore", paths) self.assertIn("tools/diagnose_ci_dns.py", paths) bad = {"source_paths": {"include": [{"path": ".env", "kind": "build-config", "hidden": True}], "deny": []}} with self.assertRaises(package_release.CheckFailure): package_release.collect_source_files(ROOT, bad)
def test_manifest_denies_c64dev_at_extracted_source(self): manifest = json.loads( (TOOLS / "release_source_manifest.json").read_text(encoding="utf-8") ) denied = { entry["path"] for entry in manifest.get("source_paths", {}).get("deny", []) } self.assertIn("c64dev.toml", denied)
class SourceHygieneTests(unittest.TestCase): def test_helper_never_uses_a_shell(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn("shell=True", source) self.assertNotIn("os.system", source) self.assertNotIn("eval(", source)
def test_helper_never_imports_the_publisher(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn("publish_tangled_artifacts", source)
def test_helper_does_not_capture_output_in_memory(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn("capture_output", source) self.assertNotIn("subprocess.run(", source)
def test_output_is_not_spooled_to_a_temp_file_until_exit(self): # A spool that is only read at exit loses every log when the workflow # kills the helper, so the runner streams instead. source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn("mkstemp", source) self.assertIn("select.select", source)
def test_child_processes_are_put_in_their_own_group_and_killed_by_group(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertIn("start_new_session=True", source) self.assertIn("os.killpg", source)
def test_helper_never_prints_the_environment(self): source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") self.assertNotIn("os.environ.items", source) self.assertNotIn("os.environ.copy", source)
if __name__ == "__main__": unittest.main()