From 3678661c1a267a8a653e5290b875bf186e16b8e3 Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 16:24:49 +0200 Subject: [PATCH 1/6] refactor(cf): initialize the ChannelFinder client in the constructor Create the adapter when the processor is constructed and always register required properties when the service starts, including with an injected test client. --- server/recceiver/cf/processor.py | 42 +++++++++++++++----------------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/server/recceiver/cf/processor.py b/server/recceiver/cf/processor.py index 4ae97c75..7df2a6ad 100755 --- a/server/recceiver/cf/processor.py +++ b/server/recceiver/cf/processor.py @@ -42,7 +42,15 @@ def __init__(self, name: Optional[str], conf: ConfigAdapter): self.name = name # Override name from service.Service self.channel_ioc_ids: Dict[str, List[str]] = defaultdict(list) self.iocs: Dict[str, IOCInfo] = {} - self.client: Optional[ChannelFinderAdapter] = None + self.client: Optional[ChannelFinderAdapter] = PyCFClientAdapter( + ChannelFinderClient( + BaseURL=self.cf_config.base_url, + username=self.cf_config.cf_username, + password=self.cf_config.cf_password, + verify_ssl=self.cf_config.verify_ssl, + ), + size_limit=int(self.cf_config.cf_query_limit), + ) self.current_time: Callable[[Optional[str]], str] = get_current_time self.lock: DeferredLock = DeferredLock() self._ioc_channels: Dict[str, Set[str]] = defaultdict(set) # iocid → set of channel names @@ -78,28 +86,18 @@ def _logStatus(self): def _start_service_with_lock(self): log.info("CF_START with configuration: %s", self.cf_config) - if self.client is None: # For setting up mock test client - self.client = PyCFClientAdapter( - ChannelFinderClient( - BaseURL=self.cf_config.base_url, - username=self.cf_config.cf_username, - password=self.cf_config.cf_password, - verify_ssl=self.cf_config.verify_ssl, - ), - size_limit=int(self.cf_config.cf_query_limit), - ) - try: - cf_properties = set(self.client.get_property_names()) - self._setup_cf_properties(cf_properties) - except ConnectionError: - log.exception("Cannot connect to Channelfinder service") - raise - else: - if self.cf_config.clean_on_start: - log.info("CF Clean: scheduling background startup sweep") - from twisted.internet import reactor + try: + cf_properties = set(self.client.get_property_names()) + self._setup_cf_properties(cf_properties) + except ConnectionError: + log.exception("Cannot connect to Channelfinder service") + raise + else: + if self.cf_config.clean_on_start: + log.info("CF Clean: scheduling background startup sweep") + from twisted.internet import reactor - reactor.callLater(0, self._start_background_clean) + reactor.callLater(0, self._start_background_clean) def _setup_cf_properties(self, cf_properties: Set[str]) -> None: """Compute required CF properties, register any missing ones, and cache state. From 00ce45147586635c1585ce18e1e26b45561df8c4 Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 16:40:03 +0200 Subject: [PATCH 2/6] fix(cf): serialize startup cleanup, commits, and shutdown Gate transactions on startup cleanup while keeping cancellation local to each transaction. Track startup commit waiters so shutdown can cancel them without cancelling the cleanup worker. Wait for startup and active commit workers before stop-time cleanup, reject commits after stopping, and prevent restart until shutdown completes. Include deterministic regression tests for ordering, cancellation, failures, and repeated service starts. --- server/recceiver/cf/processor.py | 84 +++- server/tests/unit/cf/test_processor.py | 86 ++++ .../tests/unit/cf/test_processor_lifecycle.py | 451 ++++++++++++++++++ 3 files changed, 608 insertions(+), 13 deletions(-) create mode 100644 server/tests/unit/cf/test_processor_lifecycle.py diff --git a/server/recceiver/cf/processor.py b/server/recceiver/cf/processor.py index 7df2a6ad..329a4bb6 100755 --- a/server/recceiver/cf/processor.py +++ b/server/recceiver/cf/processor.py @@ -10,6 +10,7 @@ from twisted.internet import defer, task from twisted.internet.defer import DeferredLock from twisted.internet.threads import deferToThread +from twisted.python.failure import Failure from zope.interface import implementer from recceiver import interfaces, metrics @@ -53,10 +54,18 @@ def __init__(self, name: Optional[str], conf: ConfigAdapter): ) self.current_time: Callable[[Optional[str]], str] = get_current_time self.lock: DeferredLock = DeferredLock() + self._startup_clean: Optional[defer.Deferred] = None + self._startup_commit_waiters: Set[defer.Deferred] = set() + self._stopping = False + self._shutdown: Optional[defer.Deferred] = None self._ioc_channels: Dict[str, Set[str]] = defaultdict(set) # iocid → set of channel names self._statusLoop = None def startService(self): + if self.running or self._stopping: + raise RuntimeError("Cannot start CF Processor while running or stopping") + # Deferreds are one-shot; create a fresh gate for every service start. + self._startup_clean = defer.Deferred() service.Service.startService(self) # Returning a Deferred is not supported by startService(), # so instead attempt to acquire the lock synchonously! @@ -68,8 +77,10 @@ def startService(self): try: self._start_service_with_lock() + deferToThread(self.clean_on_start).chainDeferred(self._startup_clean) except: service.Service.stopService(self) + self._startup_clean.callback(None) raise finally: self.lock.release() @@ -92,12 +103,6 @@ def _start_service_with_lock(self): except ConnectionError: log.exception("Cannot connect to Channelfinder service") raise - else: - if self.cf_config.clean_on_start: - log.info("CF Clean: scheduling background startup sweep") - from twisted.internet import reactor - - reactor.callLater(0, self._start_background_clean) def _setup_cf_properties(self, cf_properties: Set[str]) -> None: """Compute required CF properties, register any missing ones, and cache state. @@ -147,11 +152,41 @@ def _setup_cf_properties(self, cf_properties: Set[str]) -> None: log.debug("record_property_names_list = %s", self.record_property_names_list) def stopService(self): + """Wait for startup and commit workers before performing stop cleanup. + + Startup-waiting transactions are cancelled, not drained. A restart is + allowed only after shutdown finishes, so old workers cannot cross into + the next startup cycle. + """ + if self._stopping: + return self._shutdown + if not self.running: + return defer.succeed(None) log.info("CF_STOP") if self._statusLoop is not None and self._statusLoop.running: self._statusLoop.stop() service.Service.stopService(self) - return self.lock.run(self._stop_service_with_lock) + self._stopping = True + shutdown = defer.Deferred() + self._shutdown = shutdown + shutdown.addCallback(lambda _: self.lock.run(self._stop_service_with_lock)) + + def finished(result): + self._stopping = False + return result + + shutdown.addBoth(finished) + for waiter in list(self._startup_commit_waiters): + waiter.cancel() + + def startup_finished(_result): + shutdown.callback(None) + # Startup failures have already reached commit callers. Consume the + # shared gate on shutdown, when no further commits can observe it. + return None + + self._startup_clean.addBoth(startup_finished) + return shutdown def _stop_service_with_lock(self): """Stop the CFProcessor service with lock held. @@ -164,13 +199,29 @@ def _stop_service_with_lock(self): if self.cf_config.clean_on_stop: return deferToThread(self.clean_service) - def _start_background_clean(self): - log.info("CF Clean: background startup sweep beginning") - deferToThread(self.clean_service).addErrback(lambda err: log.error("CF Clean background sweep failed: %s", err)) - def commit(self, transaction_record: interfaces.ITransaction) -> defer.Deferred: - """Commit a transaction to Channelfinder.""" - return self.lock.run(self._commit_with_lock, transaction_record) + """Commit a transaction to Channelfinder, waiting for startup cleanup if enabled.""" + if not self.running: + return defer.fail(defer.CancelledError("CF Processor is not running")) + result: defer.Deferred = defer.Deferred(lambda d: self._startup_commit_waiters.discard(d)) + self._startup_commit_waiters.add(result) + # Returning the lock Deferred from a callback preserves cancellation, + # including waiting for a cancelled commit's worker to finish. + result.addCallback(lambda _: self.lock.run(self._commit_with_lock, transaction_record)) + + def start_commit(clean_result): + self._startup_commit_waiters.discard(result) + # Cancellation while waiting must not cancel the shared startup gate + # or allow this transaction to run once cleanup completes. + if not result.called: + if isinstance(clean_result, Failure): + result.errback(clean_result) + else: + result.callback(None) + return clean_result + + self._startup_clean.addBoth(start_commit) + return result def _commit_with_lock(self, transaction: interfaces.ITransaction) -> defer.Deferred: """Bridge the blocking commit thread to a cancellable Deferred. @@ -179,6 +230,8 @@ def _commit_with_lock(self, transaction: interfaces.ITransaction) -> defer.Defer Deferred (d) around it so that cancelling d (e.g. on service stop) sets self.cancelled=True, which _assert_not_cancelled picks up mid-push. """ + if not self.running: + return defer.fail(defer.CancelledError("CF Processor stopped while waiting for commit lock")) self.cancelled = False t = deferToThread(self._commit_with_thread, transaction) @@ -433,6 +486,11 @@ def remove_channel(self, record_name: str, iocid: str) -> None: if len(self.channel_ioc_ids[record_name]) == 0: del self.channel_ioc_ids[record_name] + def clean_on_start(self) -> None: + if self.cf_config.clean_on_start: + log.info("CF Clean: starting background clean") + self.clean_service() + def clean_service(self) -> None: """Mark all channels belonging to this recceiver as 'Inactive'.""" sleep = 1 diff --git a/server/tests/unit/cf/test_processor.py b/server/tests/unit/cf/test_processor.py index 2d29f362..620251c4 100644 --- a/server/tests/unit/cf/test_processor.py +++ b/server/tests/unit/cf/test_processor.py @@ -74,6 +74,92 @@ def test_keeps_ioc_when_channelcount_still_positive(self): assert proc.iocs[iocid].channelcount == 1 +class TestStartupCleanGate: + def test_commit_waits_for_startup_clean(self, monkeypatch): + proc = make_processor() + proc.running = True + transaction = make_transaction(_HOST_A, 5064) + startup_clean = defer.Deferred() + proc._startup_clean = startup_clean + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + + result = proc.commit(transaction) + completed = [] + result.addCallback(completed.append) + assert committed == [] + assert completed == [] + + startup_clean.callback(None) + + assert committed == [transaction] + assert completed == [None] + + def test_cancel_while_waiting_skips_commit_without_cancelling_startup(self, monkeypatch): + proc = make_processor() + proc.running = True + startup_clean = defer.Deferred() + proc._startup_clean = startup_clean + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + transaction = make_transaction(_HOST_A, 5064) + + cancelled = proc.commit(transaction) + errors = [] + cancelled.addErrback(errors.append) + cancelled.cancel() + assert errors[0].check(defer.CancelledError) + assert not startup_clean.called + + following = proc.commit(transaction) + completed = [] + following.addCallback(completed.append) + startup_clean.callback(None) + + assert committed == [transaction] + assert completed == [None] + + def test_cancel_while_waiting_for_lock_removes_queued_commit(self, monkeypatch): + proc = make_processor() + proc.running = True + proc._startup_clean = defer.succeed(None) + proc.lock.acquire() + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + + result = proc.commit(make_transaction(_HOST_A, 5064)) + errors = [] + result.addErrback(errors.append) + result.cancel() + proc.lock.release() + + assert errors[0].check(defer.CancelledError) + assert committed == [] + assert not proc.lock.locked + + def test_cancel_active_commit_waits_for_worker_and_releases_lock(self, monkeypatch): + proc = make_processor() + proc.running = True + proc._startup_clean = defer.succeed(None) + worker = defer.Deferred() + monkeypatch.setattr("recceiver.cf.processor.deferToThread", lambda *args: worker) + + result = proc.commit(make_transaction(_HOST_A, 5064)) + errors = [] + result.addErrback(errors.append) + result.cancel() + + assert proc.cancelled + assert proc.lock.locked + assert not worker.called + assert errors == [] + + worker.callback(None) + + assert errors[0].check(defer.CancelledError) + assert not proc.lock.locked + + class TestCleanService: def test_marks_active_channels_inactive(self): proc, adapter = make_processor_with_mock() diff --git a/server/tests/unit/cf/test_processor_lifecycle.py b/server/tests/unit/cf/test_processor_lifecycle.py new file mode 100644 index 00000000..51618117 --- /dev/null +++ b/server/tests/unit/cf/test_processor_lifecycle.py @@ -0,0 +1,451 @@ +"""Deterministic service tests: run real workers at explicitly chosen boundaries.""" + +from collections import deque +from dataclasses import dataclass, field + +import pytest +from twisted.internet import defer +from twisted.internet.address import IPv4Address + +from recceiver.cf.model import PVStatus +from recceiver.cf.processor import CFProcessor +from recceiver.recast import Transaction +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel +from tests.unit.cf.mock_adapter import MockCFAdapter +from tests.unit.conftest import make_adapter + + +@dataclass +class ThreadJob: + function: object + args: tuple + kwargs: dict + result: defer.Deferred = field(default_factory=defer.Deferred) + execution: object = field(default=None, init=False) + delivered: bool = field(default=False, init=False) + + def execute(self): + """Run worker code, but hold its completion until the test delivers it.""" + assert self.execution is None + self.execution = defer.maybeDeferred(self.function, *self.args, **self.kwargs) + + def finish(self): + assert self.execution is not None + assert not self.delivered + self.delivered = True + self.execution.chainDeferred(self.result) + + def run(self): + self.execute() + self.finish() + + +class ControlledThreads: + """Queue thread jobs without executing them or starting a reactor.""" + + def __init__(self): + self.pending = deque() + self.submitted = [] + + def submit(self, function, *args, **kwargs): + job = ThreadJob(function, args, kwargs) + self.pending.append(job) + self.submitted.append(job) + return job.result + + def take_next(self, expected_name): + job = self.pending.popleft() + assert job.function.__name__ == expected_name + return job + + def run_next(self, expected_name): + job = self.take_next(expected_name) + job.run() + return job + + +class Outcome: + """Observe completion, including called-but-paused Deferreds; consume failures.""" + + def __init__(self, deferred): + self.values = [] + self.failures = [] + deferred.addCallbacks(self.values.append, self.failures.append) + + @property + def completed(self): + return bool(self.values or self.failures) + + def assert_success(self): + assert self.values == [None] + assert self.failures == [] + + def assert_failure(self, exception): + assert self.values == [] + assert len(self.failures) == 1 + assert self.failures[0].check(exception) + + +def transaction(name="LIVE:PV", ioc_name="IOC1"): + tx = Transaction(IPv4Address("TCP", "1.2.3.4", 5064), id=0) + tx.initial = True + tx.client_infos.update({"IOCNAME": ioc_name, "HOSTNAME": "ioc.example.com"}) + tx.records_to_add[1] = (name, "ai") + return tx + + +def status(adapter, name): + return next(p.value for p in adapter._channels[name].properties if p.name == "pvStatus") + + +@pytest.fixture +def lifecycle(monkeypatch): + threads = ControlledThreads() + monkeypatch.setattr("recceiver.cf.processor.deferToThread", threads.submit) + proc = CFProcessor( + "test", + make_adapter(values={"recceiverId": DEFAULT_RECCEIVER_ID, "statusInterval": 0}), + ) + adapter = MockCFAdapter() + proc.client = adapter + yield proc, adapter, threads + # Tests must account for every submitted worker and release every lock. + assert not threads.pending + assert all(job.delivered for job in threads.submitted) + assert not proc.lock.locked + if proc.running: + stopped = Outcome(proc.stopService()) + if threads.pending: + threads.run_next("clean_service") + stopped.assert_success() + assert not threads.pending + assert all(job.delivered for job in threads.submitted) + assert not proc.lock.locked + assert not proc._startup_commit_waiters + assert not proc._stopping + + +def test_start_commit_stop_uses_real_workers(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + + committed = Outcome(proc.commit(transaction())) + assert not committed.completed + threads.run_next("_commit_with_thread") + committed.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + stopped = Outcome(proc.stopService()) + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.INACTIVE.value + + +def test_executor_propagates_worker_exceptions(): + threads = ControlledThreads() + + def failing_worker(): + raise ValueError("worker failed") + + outcome = Outcome(threads.submit(failing_worker)) + assert not outcome.completed + threads.run_next("failing_worker") + outcome.assert_failure(ValueError) + assert not threads.pending + + +def test_cleanup_precedes_live_commit_and_leaves_stale_channels_inactive(lifecycle): + proc, adapter, threads = lifecycle + adapter.set_channels([make_channel("LIVE:PV"), make_channel("STALE:PV")]) + proc.startService() + committed = Outcome(proc.commit(transaction())) + + assert [job.function.__name__ for job in threads.submitted] == ["clean_on_start"] + assert not committed.completed + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.ACTIVE.value + + threads.run_next("clean_on_start") + assert not committed.completed + assert status(adapter, "LIVE:PV") == PVStatus.INACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + + threads.run_next("_commit_with_thread") + committed.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + + +def test_startup_waiters_commit_serially(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + first = Outcome(proc.commit(transaction("FIRST:PV", "FIRST"))) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + assert len(threads.submitted) == 1 + + threads.run_next("clean_on_start") + assert len(threads.pending) == 1 + assert not first.completed + assert not second.completed + + threads.run_next("_commit_with_thread") + first.assert_success() + assert not second.completed + assert len(threads.pending) == 1 + assert "SECOND:PV" not in adapter._channels + + threads.run_next("_commit_with_thread") + second.assert_success() + assert status(adapter, "FIRST:PV") == PVStatus.ACTIVE.value + assert status(adapter, "SECOND:PV") == PVStatus.ACTIVE.value + + +@pytest.mark.parametrize("clean_on_start", [True, False]) +def test_commit_after_startup_does_not_repeat_cleanup(lifecycle, clean_on_start): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_start = clean_on_start + adapter.set_channels([make_channel("STALE:PV")]) + proc.startService() + threads.run_next("clean_on_start") + expected = PVStatus.INACTIVE if clean_on_start else PVStatus.ACTIVE + assert status(adapter, "STALE:PV") == expected.value + + committed = Outcome(proc.commit(transaction())) + threads.run_next("_commit_with_thread") + committed.assert_success() + assert [job.function.__name__ for job in threads.submitted] == ["clean_on_start", "_commit_with_thread"] + assert status(adapter, "STALE:PV") == expected.value + + +def test_cancelled_startup_waiter_does_not_affect_other_commits(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + cancelled = proc.commit(transaction("CANCELLED:PV", "CANCELLED")) + outcome = Outcome(cancelled) + following = Outcome(proc.commit(transaction())) + cancelled.cancel() + outcome.assert_failure(defer.CancelledError) + assert len(threads.submitted) == 1 + + threads.run_next("clean_on_start") + threads.run_next("_commit_with_thread") + following.assert_success() + assert "CANCELLED:PV" not in adapter._channels + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + +def test_startup_failure_reaches_waiting_and_later_commits(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + + def fail_find(_recceiver_id): + raise ValueError("unexpected cleanup failure") + + monkeypatch.setattr(adapter, "find_active_for_recceiver", fail_find) + proc.startService() + first = Outcome(proc.commit(transaction())) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + threads.run_next("clean_on_start") + first.assert_failure(ValueError) + second.assert_failure(ValueError) + later = Outcome(proc.commit(transaction("LATER:PV", "LATER"))) + later.assert_failure(ValueError) + assert len(threads.submitted) == 1 + assert adapter._channels == {} + # Shutdown consumes the shared gate, after all callers observed its failure. + # Avoid running shutdown's worker with the patched, failing adapter method. + proc.cf_config.clean_on_stop = False + + +@pytest.mark.parametrize("clean_on_stop", [True, False]) +def test_shutdown_waits_for_active_commit_before_stop_cleanup(lifecycle, clean_on_stop): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = clean_on_stop + proc.startService() + threads.run_next("clean_on_start") + committed = Outcome(proc.commit(transaction())) + worker = threads.take_next("_commit_with_thread") + worker.execute() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert not committed.completed + + stopped = Outcome(proc.stopService()) + assert not stopped.completed + assert not threads.pending + assert proc.lock.locked + + worker.finish() + committed.assert_success() + if clean_on_stop: + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + expected = PVStatus.INACTIVE if clean_on_stop else PVStatus.ACTIVE + assert status(adapter, "LIVE:PV") == expected.value + + +@pytest.mark.parametrize("clean_on_stop", [True, False]) +def test_shutdown_during_startup_cancels_waiters_and_waits_for_cleanup(lifecycle, clean_on_stop): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = clean_on_stop + adapter.set_channels([make_channel("STALE:PV")]) + proc.startService() + first = Outcome(proc.commit(transaction())) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + + shutdown = proc.stopService() + assert proc.stopService() is shutdown + stopped = Outcome(shutdown) + first.assert_failure(defer.CancelledError) + second.assert_failure(defer.CancelledError) + assert not stopped.completed + assert len(threads.submitted) == 1 + with pytest.raises(RuntimeError, match="stopping"): + proc.startService() + + threads.run_next("clean_on_start") + if clean_on_stop: + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + assert "LIVE:PV" not in adapter._channels + assert "SECOND:PV" not in adapter._channels + assert all(job.function.__name__ != "_commit_with_thread" for job in threads.submitted) + + +def test_restart_uses_fresh_cleanup_gate_without_old_transactions(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + old_gate = proc._startup_clean + old_commit = Outcome(proc.commit(transaction("OLD:PV", "OLD"))) + stopped = Outcome(proc.stopService()) + threads.run_next("clean_on_start") + threads.run_next("clean_service") + stopped.assert_success() + old_commit.assert_failure(defer.CancelledError) + + proc.startService() + assert proc._startup_clean is not old_gate + restarted = Outcome(proc.commit(transaction())) + assert not restarted.completed + assert len(threads.pending) == 1 + threads.run_next("clean_on_start") + assert not restarted.completed + threads.run_next("_commit_with_thread") + restarted.assert_success() + assert "OLD:PV" not in adapter._channels + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + +def test_initialization_failure_stops_service_without_launching_workers(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + + def fail_properties(): + raise ValueError("invalid property setup") + + monkeypatch.setattr(adapter, "get_property_names", fail_properties) + with pytest.raises(ValueError, match="invalid property setup"): + proc.startService() + assert not proc.running + assert not proc.lock.locked + assert threads.submitted == [] + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + Outcome(proc.stopService()).assert_success() + + +def test_commit_before_start_and_after_stop_is_cancelled(lifecycle): + proc, adapter, threads = lifecycle + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + proc.startService() + threads.run_next("clean_on_start") + stopped = Outcome(proc.stopService()) + threads.run_next("clean_service") + stopped.assert_success() + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + assert len(threads.submitted) == 2 + assert adapter._channels == {} + + +def test_active_cancellation_waits_for_completion_before_next_commit(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + cancelled = proc.commit(transaction("FIRST:PV", "FIRST")) + first = Outcome(cancelled) + worker = threads.take_next("_commit_with_thread") + worker.execute() + following = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + + cancelled.cancel() + assert proc.cancelled + assert not first.completed + assert not following.completed + assert not threads.pending + assert proc.lock.locked + + worker.finish() + first.assert_failure(defer.CancelledError) + assert not following.completed + threads.run_next("_commit_with_thread") + following.assert_success() + assert not proc.lock.locked + assert status(adapter, "SECOND:PV") == PVStatus.ACTIVE.value + + +def test_shutdown_waits_for_startup_failure_without_stranding_waiters(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = False + + def fail_find(_recceiver_id): + raise ValueError("unexpected cleanup failure") + + monkeypatch.setattr(adapter, "find_active_for_recceiver", fail_find) + proc.startService() + committed = Outcome(proc.commit(transaction())) + stopped = Outcome(proc.stopService()) + committed.assert_failure(defer.CancelledError) + assert not stopped.completed + threads.run_next("clean_on_start") + stopped.assert_success() + assert not proc._stopping + + +def test_reentrant_stop_from_cancelled_waiter_returns_same_shutdown(lifecycle): + proc, _adapter, threads = lifecycle + proc.startService() + reentrant_shutdowns = [] + committed = proc.commit(transaction()) + + def cancelled(failure): + assert failure.check(defer.CancelledError) + reentrant_shutdowns.append(proc.stopService()) + + committed.addErrback(cancelled) + shutdown = proc.stopService() + assert reentrant_shutdowns == [shutdown] + stopped = Outcome(shutdown) + threads.run_next("clean_on_start") + threads.run_next("clean_service") + stopped.assert_success() + + +def test_shutdown_rejects_lock_queued_commit_without_starting_worker(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + first = Outcome(proc.commit(transaction("FIRST:PV", "FIRST"))) + worker = threads.take_next("_commit_with_thread") + worker.execute() + queued = Outcome(proc.commit(transaction("QUEUED:PV", "QUEUED"))) + stopped = Outcome(proc.stopService()) + + worker.finish() + first.assert_success() + queued.assert_failure(defer.CancelledError) + assert [job.function.__name__ for job in threads.pending] == ["clean_service"] + assert "QUEUED:PV" not in adapter._channels + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "FIRST:PV") == PVStatus.INACTIVE.value From 4df2b1239259781e8219cc2301525129b2fa484d Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 16:24:51 +0200 Subject: [PATCH 3/6] test(cf): cover cleanup scope, batching, and retries Move cleanup coverage into a dedicated module and exercise receiver isolation, empty results, query batches, transient failures, capped backoff, and bounded shutdown retries. --- server/tests/unit/cf/test_cleanup.py | 136 +++++++++++++++++++++++++ server/tests/unit/cf/test_processor.py | 16 +-- 2 files changed, 137 insertions(+), 15 deletions(-) create mode 100644 server/tests/unit/cf/test_cleanup.py diff --git a/server/tests/unit/cf/test_cleanup.py b/server/tests/unit/cf/test_cleanup.py new file mode 100644 index 00000000..bb36e907 --- /dev/null +++ b/server/tests/unit/cf/test_cleanup.py @@ -0,0 +1,136 @@ +"""Cleanup scope and retry tests, independent of Deferred lifecycle ordering.""" + +import pytest +from requests import RequestException + +from recceiver.cf.model import PVStatus +from recceiver.cf.processor import CFProcessor +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel +from tests.unit.cf.mock_adapter import MockCFAdapter +from tests.unit.conftest import make_adapter + + +class CleanupAdapter(MockCFAdapter): + def __init__(self, batch_size=None): + super().__init__() + self.batch_size = batch_size + self.query_failures = 0 + self.update_failures = 0 + self.queries = 0 + self.update_attempts = 0 + self.updates = [] + + def find_active_for_recceiver(self, recceiverid): + self.queries += 1 + if self.query_failures: + self.query_failures -= 1 + raise RequestException("temporary query failure") + channels = super().find_active_for_recceiver(recceiverid) + return channels if self.batch_size is None else channels[: self.batch_size] + + def update_property(self, prop, channel_names): + self.update_attempts += 1 + if self.update_failures: + self.update_failures -= 1 + raise RequestException("temporary update failure") + super().update_property(prop, channel_names) + self.updates.append((prop, list(channel_names))) + + +@pytest.fixture +def cleanup(monkeypatch): + proc = CFProcessor("test", make_adapter(values={"recceiverId": DEFAULT_RECCEIVER_ID})) + adapter = CleanupAdapter() + proc.client = adapter + sleeps = [] + monkeypatch.setattr("recceiver.cf.processor.time.sleep", sleeps.append) + return proc, adapter, sleeps + + +def channel_status(adapter, name): + return next(p.value for p in adapter._channels[name].properties if p.name == "pvStatus") + + +def test_cleanup_only_updates_this_receivers_active_channels(cleanup): + proc, adapter, sleeps = cleanup + adapter.set_channels( + [ + make_channel("ACTIVE:PV"), + make_channel("INACTIVE:PV", active=False), + make_channel("FOREIGN:PV", recceiver_id="another-receiver"), + ] + ) + proc.clean_service() + + assert channel_status(adapter, "ACTIVE:PV") == PVStatus.INACTIVE.value + assert channel_status(adapter, "INACTIVE:PV") == PVStatus.INACTIVE.value + assert channel_status(adapter, "FOREIGN:PV") == PVStatus.ACTIVE.value + assert len(adapter.updates) == 1 + prop, names = adapter.updates[0] + assert names == ["ACTIVE:PV"] + assert prop.name == "pvStatus" + assert prop.value == PVStatus.INACTIVE.value + assert prop.owner == proc.cf_config.username + assert sleeps == [] + + +def test_empty_cleanup_makes_no_updates(cleanup): + proc, adapter, sleeps = cleanup + proc.clean_service() + assert adapter.queries == 1 + assert adapter.update_attempts == 0 + assert sleeps == [] + + +def test_cleanup_exhausts_all_query_batches(cleanup): + proc, adapter, sleeps = cleanup + adapter.batch_size = 2 + adapter.set_channels([make_channel(f"PV:{index}") for index in range(5)]) + proc.clean_service() + + assert [names for _, names in adapter.updates] == [["PV:0", "PV:1"], ["PV:2", "PV:3"], ["PV:4"]] + assert adapter.queries == 4 + assert all(channel_status(adapter, name) == PVStatus.INACTIVE.value for name in adapter._channels) + assert sleeps == [] + + +@pytest.mark.parametrize("failure_kind", ["query_failures", "update_failures"]) +def test_cleanup_recovers_from_transient_failure(cleanup, failure_kind): + proc, adapter, sleeps = cleanup + proc.running = True + setattr(adapter, failure_kind, 1) + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert sleeps == [1] + assert channel_status(adapter, "PV:1") == PVStatus.INACTIVE.value + assert len(adapter.updates) == 1 + assert adapter.queries == 3 + assert adapter.update_attempts == (2 if failure_kind == "update_failures" else 1) + + +def test_cleanup_retry_backoff_is_capped_and_recovers_while_running(cleanup): + proc, adapter, sleeps = cleanup + proc.running = True + adapter.query_failures = 14 + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert len(sleeps) == 14 + assert sleeps[:3] == [1, 1.5, 2.25] + assert max(sleeps) == 60 + assert sleeps[-1] == 60 + assert channel_status(adapter, "PV:1") == PVStatus.INACTIVE.value + + +def test_stopped_cleanup_abandons_persistent_failure_with_bounded_retries(cleanup): + proc, adapter, sleeps = cleanup + proc.running = False + adapter.query_failures = 100 + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert sleeps == [1, 1.5, 2.25, 3.375] + assert adapter.queries == 4 + assert adapter.update_attempts == 0 + assert channel_status(adapter, "PV:1") == PVStatus.ACTIVE.value diff --git a/server/tests/unit/cf/test_processor.py b/server/tests/unit/cf/test_processor.py index 620251c4..9cdd7739 100644 --- a/server/tests/unit/cf/test_processor.py +++ b/server/tests/unit/cf/test_processor.py @@ -8,7 +8,7 @@ from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, PVStatus, RecordInfo from recceiver.cf.processor import CFProcessor from recceiver.recast import Transaction -from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel, make_ioc +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_ioc from tests.unit.cf.mock_adapter import MockCFAdapter from tests.unit.conftest import make_adapter @@ -160,20 +160,6 @@ def test_cancel_active_commit_waits_for_worker_and_releases_lock(self, monkeypat assert not proc.lock.locked -class TestCleanService: - def test_marks_active_channels_inactive(self): - proc, adapter = make_processor_with_mock() - adapter.set_channels([make_channel("PV:1"), make_channel("PV:2")]) - proc.clean_service() - for name in ("PV:1", "PV:2"): - status = next(p for p in adapter._channels[name].properties if p.name == CFPropertyName.PV_STATUS.value) - assert status.value == PVStatus.INACTIVE.value - - def test_is_no_op_when_no_active_channels(self): - proc, _ = make_processor_with_mock() - proc.clean_service() - - class TestUpdateChannelFinder: def _make_proc(self): proc, adapter = make_processor_with_mock() From 81a3703b5c792d6e379dcdbfa9a7851a88042e80 Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 17:18:16 +0200 Subject: [PATCH 4/6] feat(cf): add channel property lookup and matching helpers Add property_value() for canonical and custom property names, and has_property() for name-only checks or exact CFProperty matches. Cover enum and string names, missing properties, empty and None values, exact matches, and first-match lookup semantics. --- server/recceiver/cf/model.py | 21 +++++++++- server/tests/unit/cf/test_model.py | 66 ++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/server/recceiver/cf/model.py b/server/recceiver/cf/model.py index 8dff6d26..4f817299 100644 --- a/server/recceiver/cf/model.py +++ b/server/recceiver/cf/model.py @@ -1,6 +1,6 @@ import enum from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union class PVStatus(enum.Enum): @@ -57,6 +57,25 @@ class CFChannel: owner: str properties: List[CFProperty] + def has_property(self, prop: Union[CFProperty, CFPropertyName, str]) -> bool: + """Check for a property name or an exact CFProperty match. + + Name-only checks ignore owner and value, including None or empty values. + A CFProperty checks name, owner, and value using dataclass equality. + """ + if isinstance(prop, CFProperty): + return prop in self.properties + property_name = prop.value if isinstance(prop, CFPropertyName) else prop + return any(existing.name == property_name for existing in self.properties) + + def property_value(self, name: Union[CFPropertyName, str]) -> Optional[str]: + """Return the first matching property's value, or None if absent. + + Accepts a canonical CFPropertyName or a string for custom properties. + """ + property_name = name.value if isinstance(name, CFPropertyName) else name + return next((prop.value for prop in self.properties if prop.name == property_name), None) + def as_dict(self) -> Dict[str, Any]: """Serialise to the dict shape expected by pyCFClient.""" return { diff --git a/server/tests/unit/cf/test_model.py b/server/tests/unit/cf/test_model.py index 5a917107..6c114b58 100644 --- a/server/tests/unit/cf/test_model.py +++ b/server/tests/unit/cf/test_model.py @@ -1,3 +1,5 @@ +import pytest + from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, IOCInfo, PVStatus @@ -39,6 +41,70 @@ def test_from_dict_roundtrip(self): class TestCFChannel: + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + @pytest.mark.parametrize("value", ["timestamp", None, ""]) + def test_has_property_accepts_enum_or_string_regardless_of_value(self, name, value): + channel = CFChannel("PV:1", "admin", [CFProperty("time", "admin", value)]) + assert channel.has_property(name) + + def test_has_property_accepts_custom_property_name(self): + channel = CFChannel("PV:1", "admin", [CFProperty("custom", "admin", "value")]) + assert channel.has_property("custom") + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_has_property_returns_false_when_name_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert not channel.has_property(CFPropertyName.TIME) + + def test_has_property_matches_equal_property(self): + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Inactive")]) + assert channel.has_property(CFProperty("pvStatus", "admin", "Inactive")) + + @pytest.mark.parametrize( + "prop", + [ + CFProperty("other", "admin", "Inactive"), + CFProperty("pvStatus", "other", "Inactive"), + CFProperty("pvStatus", "admin", "Active"), + ], + ) + def test_has_property_requires_matching_name_owner_and_value(self, prop): + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Inactive")]) + assert not channel.has_property(prop) + + def test_has_property_returns_false_for_exact_match_on_empty_channel(self): + channel = CFChannel("PV:1", "admin", []) + assert not channel.has_property(CFProperty("pvStatus", "admin", "Inactive")) + + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + def test_property_value_accepts_enum_or_string(self, name): + channel = CFChannel( + "PV:1", + "admin", + [CFProperty("pvStatus", "admin", "Active"), CFProperty("time", "admin", "timestamp")], + ) + assert channel.property_value(name) == "timestamp" + + def test_property_value_accepts_custom_property(self): + channel = CFChannel("PV:1", "admin", [CFProperty("custom", "admin", "custom value")]) + assert channel.property_value("custom") == "custom value" + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_property_value_returns_none_when_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert channel.property_value(CFPropertyName.TIME) is None + + @pytest.mark.parametrize("value", [None, ""]) + def test_property_value_preserves_none_or_empty_value(self, value): + channel = CFChannel("PV:1", "admin", [CFProperty("time", "admin", value)]) + assert channel.property_value(CFPropertyName.TIME) == value + + def test_property_value_returns_first_match(self): + channel = CFChannel( + "PV:1", "admin", [CFProperty("time", "admin", "first"), CFProperty("time", "admin", "second")] + ) + assert channel.property_value(CFPropertyName.TIME) == "first" + def test_from_dict_roundtrip(self): ch = CFChannel( name="PV:1", From c450578de5e8fecdfe705046778f13ceff0220cb Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 17:38:23 +0200 Subject: [PATCH 5/6] feat(cf): add channel property object lookup Add CFChannel.property() for canonical and custom names. Reuse it for value and name-existence lookups while preserving exact-property matching. --- server/recceiver/cf/model.py | 12 ++++++++---- server/tests/unit/cf/test_model.py | 21 +++++++++++++++++++++ 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/server/recceiver/cf/model.py b/server/recceiver/cf/model.py index 4f817299..12ae3416 100644 --- a/server/recceiver/cf/model.py +++ b/server/recceiver/cf/model.py @@ -65,16 +65,20 @@ def has_property(self, prop: Union[CFProperty, CFPropertyName, str]) -> bool: """ if isinstance(prop, CFProperty): return prop in self.properties - property_name = prop.value if isinstance(prop, CFPropertyName) else prop - return any(existing.name == property_name for existing in self.properties) + return self.property(prop) is not None + + def property(self, name: Union[CFPropertyName, str]) -> Optional[CFProperty]: + """Return the first property matching an enum or string name, or None.""" + property_name = name.value if isinstance(name, CFPropertyName) else name + return next((prop for prop in self.properties if prop.name == property_name), None) def property_value(self, name: Union[CFPropertyName, str]) -> Optional[str]: """Return the first matching property's value, or None if absent. Accepts a canonical CFPropertyName or a string for custom properties. """ - property_name = name.value if isinstance(name, CFPropertyName) else name - return next((prop.value for prop in self.properties if prop.name == property_name), None) + prop = self.property(name) + return prop.value if prop is not None else None def as_dict(self) -> Dict[str, Any]: """Serialise to the dict shape expected by pyCFClient.""" diff --git a/server/tests/unit/cf/test_model.py b/server/tests/unit/cf/test_model.py index 6c114b58..5cc298bc 100644 --- a/server/tests/unit/cf/test_model.py +++ b/server/tests/unit/cf/test_model.py @@ -41,6 +41,27 @@ def test_from_dict_roundtrip(self): class TestCFChannel: + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + def test_property_returns_matching_object(self, name): + expected = CFProperty("time", "user", "timestamp") + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Active"), expected]) + assert channel.property(name) is expected + + def test_property_accepts_custom_name_and_none_value(self): + expected = CFProperty("custom", "user", None) + channel = CFChannel("PV:1", "admin", [expected]) + assert channel.property("custom") is expected + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_property_returns_none_when_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert channel.property(CFPropertyName.TIME) is None + + def test_property_returns_first_match(self): + first = CFProperty("time", "admin", "first") + channel = CFChannel("PV:1", "admin", [first, CFProperty("time", "admin", "second")]) + assert channel.property(CFPropertyName.TIME) is first + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) @pytest.mark.parametrize("value", ["timestamp", None, ""]) def test_has_property_accepts_enum_or_string_regardless_of_value(self, name, value): From 394641b45f28ae07860d7d578436934de7ebe9f9 Mon Sep 17 00:00:00 2001 From: Sky Brewer Date: Fri, 9 Oct 2026 17:45:11 +0200 Subject: [PATCH 6/6] test(integration): verify live IOC recovery after receiver restart Keep the IOC running across a hard receiver restart. Require fresh timestamps on every live channel and verify that a seeded stale channel becomes Inactive. Use the shared adapter fixture and typed status properties, and restore the receiver if the test leaves it stopped. --- server/tests/integration/test_single_ioc.py | 69 +++++++++++++++++++ .../unit/test_integration_live_restart.py | 37 ++++++++++ 2 files changed, 106 insertions(+) create mode 100644 server/tests/unit/test_integration_live_restart.py diff --git a/server/tests/integration/test_single_ioc.py b/server/tests/integration/test_single_ioc.py index 5dacf812..0b42aeb5 100644 --- a/server/tests/integration/test_single_ioc.py +++ b/server/tests/integration/test_single_ioc.py @@ -1,11 +1,15 @@ import logging import time from pathlib import Path +from uuid import uuid4 import pytest from channelfinder import ChannelFinderClient from testcontainers.compose import DockerCompose +from recceiver.cf.adapter import PyCFClientAdapter +from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, PVStatus + from .cf_client import ( BASE_IOC_CHANNEL_COUNT, DEFAULT_CHANNEL_NAME, @@ -39,6 +43,11 @@ def cf_client(setup_compose: DockerCompose) -> ChannelFinderClient: # noqa: F81 return create_client_and_wait(setup_compose, expected_channel_count=BASE_IOC_CHANNEL_COUNT) +@pytest.fixture(scope="class") +def cf_adapter(setup_compose: DockerCompose) -> PyCFClientAdapter: # noqa: F811 + return PyCFClientAdapter(create_client_and_wait(setup_compose, expected_channel_count=BASE_IOC_CHANNEL_COUNT)) + + class TestRestartIOC: def test_channels_same_after_restart(self, setup_compose: DockerCompose, cf_client: ChannelFinderClient) -> None: # noqa: F811 channels_begin = cf_client.find(name="*") @@ -154,6 +163,66 @@ def test_startup_sweep_marks_stale_channels_inactive( assert all(INACTIVE_PROPERTY in ch["properties"] for ch in channels_inactive) +class TestLiveIOCRestartRecceiver: + def test_restart_restores_live_channels_and_cleans_stale_channels( + self, setup_compose: DockerCompose, cf_adapter: PyCFClientAdapter + ) -> None: # noqa: F811 + baseline = cf_adapter.find_by_names(["*"]) + assert len(baseline) == BASE_IOC_CHANNEL_COUNT + expected_names = {channel.name for channel in baseline} + live_names = sorted(expected_names) + receiver = baseline[0].property(CFPropertyName.RECCEIVER_ID) + assert receiver is not None and receiver.value is not None + active = CFProperty(CFPropertyName.PV_STATUS.value, "admin", PVStatus.ACTIVE.value) + inactive = CFProperty(CFPropertyName.PV_STATUS.value, "admin", PVStatus.INACTIVE.value) + stale_name = f"RECSYNC:STALE:{uuid4().hex}" + receiver_id = setup_compose.get_container("recc1").ID + + def recovered(adapter: PyCFClientAdapter) -> bool: + channels = adapter.find_active_for_recceiver(receiver.value) + stale = adapter.find_by_names([stale_name]) + return ( + {channel.name for channel in channels} == expected_names + and all( + channel.property_value(CFPropertyName.TIME) not in (None, previous_times[channel.name]) + for channel in channels + ) + and len(stale) == 1 + and stale[0].has_property(inactive) + ) + + assert wait_for_sync( + cf_adapter, + lambda adapter: {ch.name for ch in adapter.find_active_for_recceiver(receiver.value)} == expected_names, + ), "IOC channels did not become Active before restart" + receiver_stopped = False + try: + # The class-scoped Compose fixture removes the seeded channel along + # with the rest of the ChannelFinder data during teardown. + cf_adapter.set_channels([CFChannel(stale_name, "admin", [receiver, active])]) + assert wait_for_sync( + cf_adapter, + lambda adapter: any(ch.has_property(active) for ch in adapter.find_by_names([stale_name])), + ) + # Leave the IOC running so its upload competes with startup cleanup. + kill_container(setup_compose, "recc1") + receiver_stopped = True + # Startup cleanup changes only pvStatus. A changed timestamp proves + # each live channel was uploaded again, rather than left Active. + previous_times = { + channel.name: channel.property_value(CFPropertyName.TIME) + for channel in cf_adapter.find_by_names(live_names) + } + assert set(previous_times) == expected_names + assert all(value is not None for value in previous_times.values()), "Live channels are missing timestamps" + start_container(setup_compose, container_id=receiver_id) + receiver_stopped = False + assert wait_for_sync(cf_adapter, recovered), "Fresh IOC upload and stale-channel cleanup did not complete" + finally: + if receiver_stopped: + start_container(setup_compose, container_id=receiver_id) + + class TestMoveIocHost: def test_move_ioc_host( self, diff --git a/server/tests/unit/test_integration_live_restart.py b/server/tests/unit/test_integration_live_restart.py new file mode 100644 index 00000000..429a6e03 --- /dev/null +++ b/server/tests/unit/test_integration_live_restart.py @@ -0,0 +1,37 @@ +from unittest.mock import Mock + +import pytest + +from recceiver.cf.adapter import PyCFClientAdapter +from recceiver.cf.model import CFChannel, CFProperty +from tests.integration import test_single_ioc + + +def test_live_restart_seeds_typed_properties_without_the_adapter_helper_migration(monkeypatch): + receiver = CFProperty("recceiverID", "admin", "receiver") + baseline = [CFChannel(name, "admin", [receiver]) for name in ["PV:1", "PV:1:alias"]] + adapter = Mock(spec=PyCFClientAdapter) + adapter.find_by_names.return_value = baseline + adapter.find_active_for_recceiver.return_value = baseline + compose = Mock() + compose.get_container.return_value.ID = "container-id" + monkeypatch.setattr(test_single_ioc, "wait_for_sync", lambda adapter, check: check(adapter)) + + class SeedValidated(Exception): + pass + + def validate_seed(channels): + assert len(channels) == 1 + assert all(isinstance(prop, CFProperty) for prop in channels[0].properties) + assert channels[0].as_dict()["properties"] == [ + receiver.as_dict(), + {"name": "pvStatus", "owner": "admin", "value": "Active"}, + ] + raise SeedValidated + + adapter.set_channels.side_effect = validate_seed + with pytest.raises(SeedValidated): + test_single_ioc.TestLiveIOCRestartRecceiver().test_restart_restores_live_channels_and_cleans_stale_channels( + compose, adapter + ) + adapter.set_channels.assert_called_once()