From bb91c13926fc738dd13f385f1039f5e831db5b81 Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 13 Oct 2026 20:59:25 +0300 Subject: [PATCH] test(operations): workers and admissions across processes --- operations/tests/proc_kinds.py | 40 ++++++++ operations/tests/test_processes.py | 157 +++++++++++++++++++++++++++++ 2 files changed, 197 insertions(+) create mode 100644 operations/tests/proc_kinds.py create mode 100644 operations/tests/test_processes.py diff --git a/operations/tests/proc_kinds.py b/operations/tests/proc_kinds.py new file mode 100644 index 00000000..021c0aa2 --- /dev/null +++ b/operations/tests/proc_kinds.py @@ -0,0 +1,40 @@ +"""A kind for process tests: run waits for files, so the parent can kill the +worker at a chosen point. Directory from params["dir"].""" + +import os +import time + +from operations.jobs import Kind, Outcome, Plan, register +from operations.models import Claim, Operation + + +def _wait(path, seconds=30): + end = time.monotonic() + seconds + while not os.path.exists(path) and time.monotonic() < end: + time.sleep(0.05) + + +@register +class FileStepsKind(Kind): + name = "test.filesteps" + label = "File steps" + + def normalize(self, raw): + return {"dir": str(raw["dir"]), "uuid": str(raw.get("uuid", "p1"))} + + def plan(self, user, params): + key = "instance:" + params["uuid"] + return Plan(claims={key: Claim.EXCLUSIVE}, resource=key, label="c/vm") + + def run(self, op, ctx): + d = op.params["dir"] + open(os.path.join(d, "started"), "w").close() + _wait(os.path.join(d, "go-intent")) + ctx.record_intent(step="effect") + open(os.path.join(d, "effect"), "w").close() # stands for the libvirt effect + _wait(os.path.join(d, "go-finish")) + return {"ok": True} + + def reconcile(self, op): + effect = os.path.exists(os.path.join(op.params["dir"], "effect")) + return Outcome(Operation.SUCCEEDED if effect else Operation.FAILED) diff --git a/operations/tests/test_processes.py b/operations/tests/test_processes.py new file mode 100644 index 00000000..58652d65 --- /dev/null +++ b/operations/tests/test_processes.py @@ -0,0 +1,157 @@ +"""Real processes: a worker killed at each point is recovered by the next; +one worker per host; concurrent admissions and submissions never double up.""" + +import multiprocessing +import os +import shutil +import signal +import tempfile +import time + +from django.contrib.auth import get_user_model +from django.db import connections +from django.test import TransactionTestCase + +from operations import claims, jobs +from operations.models import Claim, Operation, WorkerLease +from operations.tests import fakes, proc_kinds # noqa: F401 + + +def _worker(data): + connections.close_all() + from operations.worker import Worker + os._exit(Worker(tick=0.05, heartbeat_every=0.2, grace=1, data=data).run()) + + +def _admit(key, results): + connections.close_all() + try: + claims.admit({key: Claim.EXCLUSIVE}, holder=f"p{os.getpid()}") + results.put("admitted") + except claims.Busy: + results.put("busy") + connections.close_all() + results.close() + results.join_thread() # the feeder thread must deliver before os._exit + os._exit(0) + + +def _submit(user_pk, results): + connections.close_all() + user = get_user_model().objects.get(pk=user_pk) + try: + results.put(str(jobs.submit("test.noop", user, {"uuid": "race"}, "key-race0001").pk)) + except Exception as err: + results.put(f"error:{type(err).__name__}") + connections.close_all() + results.close() + results.join_thread() + os._exit(0) + + +class ProcessTestCase(TransactionTestCase): + def setUp(self): + self.ctx = multiprocessing.get_context("fork") + self.user = get_user_model().objects.create_user("proc_user", "p@example.com", "pw") + self.dir = tempfile.mkdtemp() + self.children = [] + + def tearDown(self): + for p in self.children: + if p.is_alive(): + p.kill() + p.join(10) + shutil.rmtree(self.dir, ignore_errors=True) + + def spawn(self, target, *args): + connections.close_all() + p = self.ctx.Process(target=target, args=args) + p.start() + self.children.append(p) + return p + + def wait(self, predicate, seconds=30): + end = time.monotonic() + seconds + while time.monotonic() < end: + if predicate(): + return True + time.sleep(0.1) + return False + + def touch(self, name): + open(os.path.join(self.dir, name), "w").close() + + def exists(self, name): + return os.path.exists(os.path.join(self.dir, name)) + + def test_concurrent_admissions_admit_once(self): + results = self.ctx.Queue() + for _ in range(8): + self.spawn(_admit, "instance:race", results) + outcomes = sorted(results.get(timeout=30) for _ in range(8)) + self.assertEqual(outcomes.count("admitted"), 1, outcomes) + for p in self.children: + p.join(10) + self.assertEqual(p.exitcode, 0) + + def test_concurrent_duplicate_submissions_return_one_job(self): + results = self.ctx.Queue() + for _ in range(4): + self.spawn(_submit, self.user.pk, results) + outcomes = {results.get(timeout=30) for _ in range(4)} + self.assertEqual(len(outcomes), 1, outcomes) + self.assertFalse(any(o.startswith("error:") for o in outcomes), outcomes) + self.assertEqual(Operation.objects.count(), 1) + + def test_a_second_worker_on_the_host_exits_with_1(self): + first = self.spawn(_worker, self.dir) + time.sleep(1) + second = self.spawn(_worker, self.dir) + second.join(15) + self.assertEqual(second.exitcode, 1) + os.kill(first.pid, signal.SIGTERM) + first.join(15) + self.assertEqual(first.exitcode, 0) + + def test_a_worker_of_another_installation_does_not_take_a_live_lease(self): + first = self.spawn(_worker, self.dir) + self.assertTrue(self.wait(lambda: WorkerLease.objects.exists())) + holder = WorkerLease.objects.get().holder + other = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, other, True) + second = self.spawn(_worker, other) # its own flock: same hostname, another data directory + time.sleep(2) + self.assertEqual(WorkerLease.objects.get().holder, holder) + for p in (first, second): + os.kill(p.pid, signal.SIGTERM) + p.join(15) + self.assertEqual(p.exitcode, 0) + + def kill_at(self, point): + op = jobs.submit("test.filesteps", self.user, {"dir": self.dir}, "key-proc0001") + worker = self.spawn(_worker, self.dir) + self.assertTrue(self.wait(lambda: self.exists("started"))) + if point == "after_effect": + self.touch("go-intent") + self.assertTrue(self.wait(lambda: self.exists("effect"))) + os.kill(worker.pid, signal.SIGKILL) + worker.join(10) + self.touch("go-intent") + self.touch("go-finish") + nxt = self.spawn(_worker, self.dir) + try: + self.assertTrue(self.wait(lambda: Operation.objects.get(pk=op.pk).state in Operation.TERMINAL)) + finally: + os.kill(nxt.pid, signal.SIGTERM) + nxt.join(15) + return Operation.objects.get(pk=op.pk) + + def test_killed_before_intent_the_job_fails(self): + op = self.kill_at("before_intent") + self.assertEqual(op.state, Operation.FAILED) + self.assertFalse(self.exists("effect")) + + def test_killed_after_the_effect_the_job_is_recovered_as_succeeded(self): + op = self.kill_at("after_effect") + self.assertEqual(op.state, Operation.SUCCEEDED) + self.assertFalse(Claim.objects.filter(operation=op).exists())