fuzzyworld

fuzzyworld-tools / files / tests / test_runner.py

test_runner.py

Blame: the commit that last changed each line · file · history

0330e724 2026-11-01 Pavel Safronov1"""Tests for the CI runner (runner/fw-runner.py) and its server side (server/runner-api).
2
3Run with: python3 -m unittest discover tests
4"""
5
6import importlib.util
7import json
8import os
9import subprocess
10import tempfile
11import time
12import unittest
13from unittest import mock
14
15ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
16spec = importlib.util.spec_from_file_location("fw_runner", os.path.join(ROOT, "runner", "fw-runner.py"))
17fw_runner = importlib.util.module_from_spec(spec)
18spec.loader.exec_module(fw_runner)
19
20API = os.path.join(ROOT, "server", "runner-api")
21JOB = "1700000000-demo-0123456789ab"
22SHA = "a" * 40
23
24
25class MissingTags(unittest.TestCase):
26 def need(self, tags, ref="refs/heads/main", protected=True, has_secrets=False, have=("docker", "chrome")):
27 jobs = [{"tags": tags}]
28 with mock.patch.object(fw_runner, "TAGS", set(have)), mock.patch.object(fw_runner, "shot_plan", return_value=None):
29 return fw_runner.missing_tags({"ref": ref}, jobs, "/nowhere", protected, has_secrets)
30
31 def test_nothing_missing(self):
32 self.assertEqual(self.need(["docker"]), [])
33
34 def test_lists_missing_tags_sorted(self):
35 self.assertEqual(self.need(["zeta", "alpha", "docker"]), ["alpha", "zeta"])
36
37 def test_screenshots_need_chrome_only_on_a_protected_branch(self):
38 jobs = [{"tags": []}]
39 with mock.patch.object(fw_runner, "TAGS", {"docker"}), mock.patch.object(fw_runner, "shot_plan", return_value=({}, "app", [])):
40 self.assertEqual(fw_runner.missing_tags({"ref": "refs/heads/main"}, jobs, "/x", True, False), ["chrome"])
41 self.assertEqual(fw_runner.missing_tags({"ref": "refs/tags/v1"}, jobs, "/x", True, False), [])
42 self.assertEqual(fw_runner.missing_tags({"ref": "refs/heads/main"}, jobs, "/x", False, False), [])
43
44 def test_secrets_is_not_a_runner_tag(self):
45 self.assertEqual(self.need(["secrets"], has_secrets=True), [])
46 self.assertEqual(self.need(["secrets"], has_secrets=False), ["secrets"])
47 self.assertEqual(self.need(["secrets"], protected=False, has_secrets=False), []) # off the default branch: never has them
48
49
50class TopParsing(unittest.TestCase):
51 def test_last_sample_wins(self):
52 out = ("CPU usage: 10.0% user, 5.0% sys, 85.0% idle\n"
53 + "CPU usage: 30.5% user, 9.5% sys, 60.0% idle\n")
54 self.assertEqual(fw_runner.parse_top_idle(out), 60.0)
55
56 def test_no_line(self):
57 self.assertIsNone(fw_runner.parse_top_idle("nothing useful"))
58
59
60class BusyReason(unittest.TestCase):
61 def busy(self, run):
62 with mock.patch.object(fw_runner, "on_ac_power", return_value=True), \
63 mock.patch.object(fw_runner, "cpu_idle_percent", return_value=99.0), \
64 mock.patch.object(fw_runner.subprocess, "run", run):
65 return fw_runner.busy_reason()
66
67 def test_docker_ok(self):
68 self.assertEqual(self.busy(lambda *a, **k: subprocess.CompletedProcess(a, 0)), "")
69
70 def test_docker_down(self):
71 self.assertEqual(self.busy(lambda *a, **k: subprocess.CompletedProcess(a, 1)), "Docker is not running")
72
73 def test_docker_hangs(self):
74 def hang(*a, **k):
75 self.assertEqual(k.get("timeout"), fw_runner.DOCKER_INFO_TIMEOUT)
76 raise subprocess.TimeoutExpired(a, k["timeout"])
77 self.assertEqual(self.busy(hang), "Docker is not answering")
78
79 def test_docker_missing(self):
80 def missing(*a, **k):
81 raise FileNotFoundError("docker")
82 self.assertEqual(self.busy(missing), "Docker is not installed")
83
84
85class Heartbeat(unittest.TestCase):
86 def test_beat_has_a_short_timeout(self):
87 seen = {}
88 def fake_run(cmd, **k):
89 seen["timeout"] = k["timeout"]
90 return subprocess.CompletedProcess(cmd, 0, b"", b"")
91 with mock.patch.object(fw_runner.subprocess, "run", fake_run):
92 fw_runner.api("beat", JOB, "r1", timeout=fw_runner.BEAT_TIMEOUT)
93 self.assertEqual(seen["timeout"], fw_runner.BEAT_TIMEOUT)
94 self.assertLess(fw_runner.BEAT_TIMEOUT, 60)
95
96
97 def test_heartbeat_thread_passes_the_timeout(self):
98 calls = []
99 hb = fw_runner.Heartbeat(JOB)
100 def fake_api(*args, **kw):
101 calls.append((args, kw))
102 hb.stop()
103 with mock.patch.object(fw_runner, "BEAT_EVERY", 0.01), mock.patch.object(fw_runner, "api", fake_api):
104 hb.start()
105 hb.join(5)
106 self.assertEqual(calls[0][1], {"timeout": fw_runner.BEAT_TIMEOUT})
107 self.assertEqual(calls[0][0][0], "beat")
108
109
110class RunnerApi(unittest.TestCase):
111 """server/runner-api run for real against a throwaway HOME."""
112
113 def setUp(self):
114 self.tmp = tempfile.TemporaryDirectory()
115 self.home = self.tmp.name
116 self.addCleanup(self.tmp.cleanup)
117 pending = os.path.join(self.home, "fw", "queue", "pending")
118 os.makedirs(pending)
119 with open(os.path.join(pending, JOB + ".json"), "w") as f:
120 json.dump({"id": JOB}, f)
121
122 def api(self, cmd):
123 env = {"PATH": os.environ["PATH"], "HOME": self.home, "SSH_ORIGINAL_COMMAND": cmd, "SSH_CLIENT": "127.0.0.1 1 2"}
124 return subprocess.run(["bash", API], env=env, capture_output=True, text=True)
125
126 def path(self, *p):
127 return os.path.join(self.home, "fw", "queue", *p)
128
129 def log(self):
130 try:
131 with open(os.path.join(self.home, "fw", "logs", "runner-api.log")) as f:
132 return f.read()
133 except OSError:
134 return ""
135
136 def test_claim_beat_done(self):
137 r = self.api("claim r1 docker")
138 self.assertEqual(json.loads(r.stdout), {"id": JOB})
139 self.assertTrue(os.path.exists(self.path("running", JOB + ".json")))
140 self.assertEqual(self.api(f"beat {JOB} r1").returncode, 0)
141 self.assertNotEqual(self.api(f"beat {JOB} other").returncode, 0) # not the holder
142 self.assertNotEqual(self.api(f"done demo {SHA} {JOB} other").returncode, 0)
143 self.assertTrue(os.path.exists(self.path("running", JOB + ".json"))) # the stale caller changed nothing
144
145 def test_release_declines_for_that_runner_only(self):
146 self.api("claim r1 docker")
147 self.assertEqual(self.api(f"release {JOB} r1").returncode, 0)
148 self.assertTrue(os.path.exists(self.path("pending", JOB + ".json")))
149 self.assertEqual(self.api("claim r1 docker").stdout, "") # r1 is not offered it again
150 self.assertEqual(json.loads(self.api("claim r2 docker").stdout), {"id": JOB})
151
152 def test_quiet_runner_loses_the_lease(self):
153 self.api("claim r1 docker")
154 old = time.time() - 10 * 60
155 os.utime(self.path("running", JOB + ".beat"), (old, old))
156 r = self.api("claim r2 docker") # the job goes back, and r2 takes it
157 self.assertEqual(json.loads(r.stdout), {"id": JOB})
158 self.assertNotEqual(self.api(f"beat {JOB} r1").returncode, 0) # r1 finds out on its next beat
159 with open(self.path("running", JOB + ".beat")) as f:
160 self.assertEqual(f.read().strip(), "r2")
161
162 def test_refused_beat_is_logged(self):
163 self.api("claim r1 docker")
164 self.assertNotEqual(self.api(f"beat {JOB} other").returncode, 0)
165 self.assertIn("refused", self.log())
166
167 def test_beats_and_empty_claims_are_not_logged(self):
168 self.api("claim r1 docker")
169 self.api(f"beat {JOB} r1")
170 self.api("claim r1 docker")
171 log = self.log()
172 self.assertNotIn("beat", log)
173 self.assertIn("claim r1 docker: " + JOB + ".json", log) # a successful claim is
174
175
176if __name__ == "__main__":
177 unittest.main()