diff --git a/src/chimera/controllers/scheduler/controller.py b/src/chimera/controllers/scheduler/controller.py index 647cd992..9bdf99b9 100644 --- a/src/chimera/controllers/scheduler/controller.py +++ b/src/chimera/controllers/scheduler/controller.py @@ -62,6 +62,8 @@ class Scheduler(ChimeraObject): "point_verify": "/PointVerify/0", "operator": "/Operator/0", "algorithm": SchedulingAlgorithm.SEQUENTIAL, + # left tracking, an unattended mount walks into a limit + "stop_tracking_on_program_end": True, } def __init__(self): diff --git a/src/chimera/controllers/scheduler/executor.py b/src/chimera/controllers/scheduler/executor.py index 7f7b36aa..5b2837b1 100644 --- a/src/chimera/controllers/scheduler/executor.py +++ b/src/chimera/controllers/scheduler/executor.py @@ -53,43 +53,54 @@ def __start__(self): def execute(self, program): self.must_stop.clear() - for action in program.actions: - # aborted? - if self.must_stop.is_set(): - raise ProgramExecutionAborted() - - t0 = time.time() - - try: - self.current_action = action - self.current_handler = self.action_handlers[type(action)] - - log_msg = str(self.current_handler.log(action)) - log.debug(f"[start] {log_msg} ") - self.controller.action_begin(action.id, log_msg) - - self.current_handler.process(action) - - # instruments just returns in case of abort, so we need to check handler - # returned 'cause of abort or not + try: + for action in program.actions: + # aborted? if self.must_stop.is_set(): - self.controller.action_complete(action.id, SchedulerStatus.ABORTED) raise ProgramExecutionAborted() - else: - self.controller.action_complete(action.id, SchedulerStatus.OK) - except ProgramExecutionException: - self.controller.action_complete(action.id, SchedulerStatus.ERROR) - raise - except KeyError: - log.debug(f"No handler to {action} action. Skipping it") - finally: - log.debug("[finish] took: %f s" % (time.time() - t0)) + t0 = time.time() + + try: + self.current_action = action + self.current_handler = self.action_handlers[type(action)] + + log_msg = str(self.current_handler.log(action)) + log.debug(f"[start] {log_msg} ") + self.controller.action_begin(action.id, log_msg) + + self.current_handler.process(action) + + # instruments just returns in case of abort, so we need to check handler + # returned 'cause of abort or not + if self.must_stop.is_set(): + self.controller.action_complete( + action.id, SchedulerStatus.ABORTED + ) + raise ProgramExecutionAborted() + else: + self.controller.action_complete(action.id, SchedulerStatus.OK) + + except ProgramExecutionException: + self.controller.action_complete(action.id, SchedulerStatus.ERROR) + raise + except KeyError: + log.debug(f"No handler to {action} action. Skipping it") + finally: + log.debug("[finish] took: %f s" % (time.time() - t0)) + finally: + # a later stop() must not abort an action of a finished program + self.current_action = None + self.current_handler = None def stop(self): - if self.current_handler: - self.must_stop.set() - self.current_handler.abort(self.current_action) + # flag first: a worker between actions must see the stop even when + # there is no action in flight to abort + self.must_stop.set() + + handler, action = self.current_handler, self.current_action + if handler: + handler.abort(action) def _inject_instrument(self, handler): if not issubclass(handler, ActionHandler): diff --git a/src/chimera/controllers/scheduler/handlers.py b/src/chimera/controllers/scheduler/handlers.py index dbd7dd2e..f6e9a822 100644 --- a/src/chimera/controllers/scheduler/handlers.py +++ b/src/chimera/controllers/scheduler/handlers.py @@ -51,9 +51,13 @@ def process(action): float(ra_dec.ra.to_h()), float(ra_dec.dec.to_d()), 2000.0 ) # epoch is always 2000.0 for pointing elif action.target_alt_az is not None: + # Position.alt/az are plain float degrees, unlike .ra/.dec + # which are Coords; accept either + alt = action.target_alt_az.alt + az = action.target_alt_az.az telescope.slew_to_alt_az( - float(action.target_alt_az.alt.to_d()), - float(action.target_alt_az.az.to_d()), + float(alt.to_d() if hasattr(alt, "to_d") else alt), + float(az.to_d() if hasattr(az, "to_d") else az), ) elif action.target_name is not None: telescope.slew_to_object(action.target_name) diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 176d9534..26483090 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -1,5 +1,6 @@ import logging import threading +import time from chimera.controllers.scheduler.model import Program, Session from chimera.controllers.scheduler.states import State @@ -8,6 +9,9 @@ log = logging.getLogger(__name__) +#: seconds between heartbeat lines while a program runs +HEARTBEAT_SECONDS = 60.0 + class Machine(threading.Thread): __state = None @@ -22,6 +26,11 @@ def __init__(self, scheduler, executor, controller): self.controller = controller self.current_program = None + # thread running the current program; the IDLE branch waits on it + self._worker = None + # heartbeat bookkeeping (monotonic seconds) + self._last_heartbeat = 0.0 + self._program_started_at = 0.0 self.daemon = False @@ -32,13 +41,37 @@ def state(self, state=None): return self.__state if state == self.__state: return - self.controller.state_changed(state, self.__state) - log.debug(f"Changing state, from {self.__state} to {state}.") + old_state = self.__state + log.debug(f"Changing state, from {old_state} to {state}.") self.__state = state self.wake_up() finally: self.__state_lock.release() + # publish outside the lock: a slow subscriber must not block state() + self.controller.state_changed(state, old_state) + + def transition(self, expected, state): + """Set the state only if the machine is currently in `expected`. + + Terminal transitions coming from the worker use this so a program's + own completion or error can never re-arm a machine that was told to + stop: on opd-40 a program error flipped a stopped machine OFF -> IDLE. + """ + self.__state_lock.acquire() + try: + if self.__state != expected or state == self.__state: + return False + old_state = self.__state + log.debug(f"Changing state, from {old_state} to {state}.") + self.__state = state + self.wake_up() + finally: + self.__state_lock.release() + + self.controller.state_changed(state, old_state) + return True + def run(self): log.info("Starting scheduler machine") self.state(State.OFF) @@ -57,6 +90,16 @@ def run(self): self.state(State.IDLE) if self.state() == State.IDLE: + # START overwrites BUSY, so a start() during a run lands here + # with the same unfinished program: wait for the worker + # instead of forking a duplicate execution + if self._worker is not None and self._worker.is_alive(): + # poll, don't sleep(): the worker signals IDLE just + # before exiting, so the wakeup can be lost + self._worker.join(1.0) + self._heartbeat() + continue + log.debug("[idle] looking for something to do...") # find something to do @@ -67,6 +110,8 @@ def run(self): log.debug("[idle] program slew start %s", program.start_at) self.state(State.BUSY) self.current_program = program + self._program_started_at = time.monotonic() + self._last_heartbeat = self._program_started_at self._process(program) continue @@ -77,25 +122,94 @@ def run(self): elif self.state() == State.BUSY: log.debug("[busy] waiting tasks to finish..") - self.sleep() + # bounded, so the heartbeat below gets a turn; the Condition + # is notified on every state change, so a STOP still lands + # immediately + self.sleep(timeout=HEARTBEAT_SECONDS) + self._heartbeat() elif self.state() == State.STOP: log.debug("[stop] trying to stop current program") - self.executor.stop() - self.state(State.OFF) - - elif self.state() == State.SHUTDOWN: - log.debug("[shutdown] trying to stop current program") - self.executor.stop() - log.debug("[shutdown] should die soon.") - break - + # STOPPING first: a program counting down to its slew time + # is released by the state change itself, and the abort + # below may request a new state (START) that must not be + # overwritten afterwards + self.state(State.STOPPING) + # abort off this thread: executor.stop() blocks until the + # running action yields (a full camera readout), and inline + # it froze the whole state machine for that long + threading.Thread( + target=self.executor.stop, name="scheduler-stop", daemon=True + ).start() + # OFF only once the worker is dead: declaring it earlier let + # a surviving exposure loop run 52 min behind a closed dome + self._join_worker("stop") + # honour a START requested while the abort ran + self.transition(State.STOPPING, State.OFF) + + # out of the loop: SHUTDOWN. Abort whatever is running and wait for + # the worker so nothing outlives the machine's word. + log.debug("[shutdown] trying to stop current program") + threading.Thread( + target=self.executor.stop, name="scheduler-stop", daemon=True + ).start() + self._join_worker("shutdown") log.debug("[shutdown] thread ending...") - def sleep(self): + def _heartbeat(self): + """Say what the machine is doing, at most once a minute. + + Both waits are silent: BUSY sleeps on the wake-up Condition while + the worker runs, and a worker counting down to its slew time (or + running a 10 min autofocus) prints nothing either. Twice that + silence was diagnosed as "the scheduler is wedged" from the log + alone. + """ + now = time.monotonic() + if now - self._last_heartbeat < HEARTBEAT_SECONDS: + return + self._last_heartbeat = now + + program = self.current_program + waiting_for = "" + start_at = getattr(program, "start_at", None) + if start_at: + waiting_for = f", waiting until MJD {start_at:.5f}" + try: + waiting_for += f" (now {self.controller.get_site().mjd():.5f})" + except Exception: + pass + log.info( + "[%s] running %s for %.0f s%s", + self.state(), + program, + now - self._program_started_at, + waiting_for, + ) + + def _join_worker(self, reason): + """Wait until the worker thread is actually dead, heartbeating: a + camera abort legitimately takes minutes, and a silent wait reads as + a wedged machine.""" + worker = self._worker + if worker is None: + return + + waited = 0 + while worker.is_alive(): + worker.join(1.0) + waited += 1 + if worker.is_alive() and waited % 10 == 0: + log.info( + "[%s] program still aborting after ~%d s, waiting...", + reason, + waited, + ) + + def sleep(self, timeout=None): self.__wake_up_call.acquire() log.debug("Sleeping") - self.__wake_up_call.wait() + self.__wake_up_call.wait(timeout) self.__wake_up_call.release() def wake_up(self): @@ -113,6 +227,30 @@ def restart_all_programs(self): session.commit() + def _stop_tracking(self): + """Leave the mount idle at the end of a program. + + Must run inline on the program thread, before program_complete: + from a detached thread the stop queues behind the next program's + slew and untracks the target it just acquired. + """ + if not self.controller["stop_tracking_on_program_end"]: + return + + location = self.controller["telescope"] + # a string-typed chimera config key coerces None to "None" + if not location or str(location).lower() in ("none", ""): + return + + try: + telescope = self.controller.get_proxy(location) + if telescope.is_tracking(): + telescope.stop_tracking() + log.info("Tracking stopped at program end.") + except Exception: + # never let this fail the program that just ran + log.exception("Could not stop telescope tracking at program end.") + def _process(self, program): def process(): # session to be used by executor and handlers @@ -120,7 +258,14 @@ def process(): task = session.merge(program) - log.debug(f"[start] {str(task)}") + # Describe the program ONCE, while its row is certainly there: + # every commit below expires the instance, and the queue can be + # rewritten under a running program (robobs cleans it, a replan + # rebuilds it), so formatting it afterwards reloads a row that + # may be gone. + label = str(task) + + log.debug(f"[start] {label}") # the manager-injected site, not a private Site(): one clock # system-wide @@ -141,27 +286,43 @@ def process(): # state change -- so a STOP/SHUTDOWN breaks the wait at once # instead of after a fixed sleep. while site.mjd() < program.start_at: - if self.state() in (State.STOP, State.SHUTDOWN): - log.debug("[start] Aborted while waiting for slew start") - return - with self.__wake_up_call: - self.__wake_up_call.wait(1.0) - else: - if program.valid_for >= 0.0: - if -wait_time > program.valid_for: + if self.state() in ( + State.STOP, + State.STOPPING, + State.SHUTDOWN, + ): log.debug( - "[start] Program is not valid anymore {program.start_at}, {program.valid_for}" + "[start] wait cancelled; abandoning %s", str(task) ) self.controller.program_complete( program.id, - SchedulerStatus.OK, - "Program not valid anymore.", + SchedulerStatus.ABORTED, + "Aborted while waiting for its slew time.", ) - else: + return + with self.__wake_up_call: + self.__wake_up_call.wait(1.0) + else: + if program.valid_for >= 0.0 and -wait_time > program.valid_for: + # too late to run: finish it without executing log.debug( - "[start] Specified slew start MJD %s has already passed; proceeding without waiting", + "[start] Program is not valid anymore (start_at %f, valid_for %f)", program.start_at, + program.valid_for, + ) + self.scheduler.done(task) + session.commit() + self.controller.program_complete( + program.id, + SchedulerStatus.OK, + "Program not valid anymore.", ) + self.transition(State.BUSY, State.IDLE) + return + log.debug( + "[start] Specified slew start MJD %s has already passed; proceeding without waiting", + program.start_at, + ) else: log.debug("[start] No slew time specified, so no waiting") log.debug("[start] Current MJD is %f", site.mjd()) @@ -170,29 +331,54 @@ def process(): ) self.controller.program_begin(program.id) + def finish(status, message=None, error=None, next_state=State.IDLE): + """Close the program out. Nothing here may kill this thread. + + The program is over by the time we get here, so a failure to + record it must not cost the completion event (robobs waits + for it to pick the next program) nor the state transition + (the machine would sit BUSY until chimera is restarted). + """ + try: + self.scheduler.done(task, error=error) + # commit the finished flag before releasing the machine: + # the idle picker re-checks it against the database + session.commit() + except Exception: + log.exception("[finish] could not record the end of %s", label) + self._stop_tracking() + try: + self.controller.program_complete(program.id, status, message) + except Exception: + log.exception("[finish] could not publish the end of %s", label) + # conditional: a completion event must not re-arm a machine + # that went STOPPING/SHUTDOWN while this program ran + self.transition(State.BUSY, next_state) + try: self.executor.execute(task) - log.debug(f"[finish] {str(task)}") - self.scheduler.done(task) - self.controller.program_complete(program.id, SchedulerStatus.OK) - self.state(State.IDLE) + log.debug(f"[finish] {label}") + finish(SchedulerStatus.OK) except ProgramExecutionException as e: - self.scheduler.done(task, error=e) - self.controller.program_complete( - program.id, SchedulerStatus.ERROR, str(e) - ) - self.state(State.IDLE) - log.debug(f"[error] {str(task)} ({str(e)})") + finish(SchedulerStatus.ERROR, str(e), error=e) + log.debug(f"[error] {label} ({str(e)})") except ProgramExecutionAborted as e: - self.scheduler.done(task, error=e) - self.controller.program_complete( - program.id, SchedulerStatus.ABORTED, "Aborted by user." + # normally the machine itself goes STOPPING -> OFF once this + # worker dies; cover an out-of-band abort arriving while BUSY + finish( + SchedulerStatus.ABORTED, + "Aborted by user.", + error=e, + next_state=State.OFF, ) - self.state(State.OFF) - log.debug(f"[aborted by user] {str(task)}") + log.debug(f"[aborted by user] {label}") - session.commit() + try: + session.commit() + except Exception: + log.exception("[finish] final commit failed for %s", label) - t = threading.Thread(target=process) + t = threading.Thread(target=process, name="scheduler-program") t.daemon = False + self._worker = t t.start() diff --git a/src/chimera/controllers/scheduler/model.py b/src/chimera/controllers/scheduler/model.py index 53814f9d..e81b9b95 100644 --- a/src/chimera/controllers/scheduler/model.py +++ b/src/chimera/controllers/scheduler/model.py @@ -11,6 +11,7 @@ PickleType, String, create_engine, + inspect, ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import backref, relation, sessionmaker @@ -71,7 +72,20 @@ class Program(Base): ) def __str__(self): - return f"#{self.id} {self.name} pi:{self.pi} #actions: {len(self.actions)}" + # Never depend on live database state: every session.commit() + # expires the instance, so this triggers a lazy reload - and if the + # queue was rewritten while the program ran (robobs cleans it on + # start AND on stop, plan_robobs rebuilds it), the row is gone and + # the reload raises ObjectDeletedError. From a log line inside an + # exception handler that killed the scheduler-program thread + # outright (opd-40 2026-07-28, on every operator lock during a + # program). A __str__ that can raise is a trap wherever it is used. + try: + return f"#{self.id} {self.name} pi:{self.pi} #actions: {len(self.actions)}" + except Exception: + identity = inspect(self).identity + program_id = identity[0] if identity else "?" + return f"#{program_id} " class Action(Base): diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index d157e462..f0044f57 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -16,14 +16,28 @@ def __init__(self): def reschedule(self, machine): self.machine = machine + # a plain thread-safe FIFO, rebuilt on every reschedule; we never + # join() it, so no task_done() accounting (which drifts across the + # rebuild and raised "task_done() called too many times") self.run_queue = Queue(-1) session = Session() # FIXME: remove noqa + # start_at orders execution, priority breaks ties: priority alone + # let a future-timed program hold the machine with the whole night + # queued behind it. Programs without start_at sort first and keep + # the old priority order among themselves; equal priorities then run + # in insertion order, which for a hand-written `chimera-sched --new` + # queue is the order the targets appear in the YAML (they came out + # reversed, which is not what anyone writing a file expects). programs = ( session.query(Program) - .order_by(desc(Program.priority)) + .order_by( + Program.start_at.asc().nullsfirst(), + desc(Program.priority), + Program.id.asc(), + ) .filter(Program.finished == False) # noqa .all() ) @@ -39,8 +53,16 @@ def reschedule(self, machine): machine.wake_up() def __next__(self): - if not self.run_queue.empty(): - return self.run_queue.get() + session = Session() + while not self.run_queue.empty(): + program = self.run_queue.get() + # reschedule() also enqueues the currently RUNNING program (its + # finished flag is only written at completion), so entries can + # be stale by the time they are popped: re-check the database + current = session.get(Program, program.id) + if current is None or current.finished: + continue + return program return None @@ -51,5 +73,4 @@ def done(self, task, error=None): else: task.finished = True - self.run_queue.task_done() self.machine.wake_up() diff --git a/src/chimera/controllers/scheduler/states.py b/src/chimera/controllers/scheduler/states.py index 1fb223db..871661bd 100644 --- a/src/chimera/controllers/scheduler/states.py +++ b/src/chimera/controllers/scheduler/states.py @@ -9,4 +9,5 @@ class State(Enum): IDLE = "IDLE" BUSY = "BUSY" STOP = "STOP" + STOPPING = "STOPPING" SHUTDOWN = "SHUTDOWN" diff --git a/tests/chimera/conftest.py b/tests/chimera/conftest.py index f45a6cdb..f1876450 100644 --- a/tests/chimera/conftest.py +++ b/tests/chimera/conftest.py @@ -1,12 +1,22 @@ import random +import tempfile import threading import time import pytest -from chimera.core.bus import Bus -from chimera.core.manager import Manager -from chimera.core.site import Site +import chimera.core.constants as constants + +# Redirect the scheduler database BEFORE any test module is imported: +# scheduler.model binds its engine to DEFAULT_PROGRAM_DATABASE at import +# time, and several test modules import it during collection — patching +# from inside a test module is too late once a sibling imported first, and +# the suite would then read AND WRITE the user's real ~/.chimera database. +constants.DEFAULT_PROGRAM_DATABASE = tempfile.mkstemp(suffix="-chimera-tests.db")[1] + +from chimera.core.bus import Bus # noqa: E402 +from chimera.core.manager import Manager # noqa: E402 +from chimera.core.site import Site # noqa: E402 @pytest.fixture diff --git a/tests/chimera/controllers/test_scheduler_deleted_program.py b/tests/chimera/controllers/test_scheduler_deleted_program.py new file mode 100644 index 00000000..3829ba3d --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_deleted_program.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A program whose row was deleted while it ran must not kill the machine. + +The queue is rewritten under running programs as a matter of routine - +robobs cleans it on start AND on stop, plan_robobs rebuilds it - so by the +time the abort lands, the row may be gone. The abort branch logged +`str(task)` after a commit had expired the instance, which reloads the row +and raises ObjectDeletedError from inside an exception handler: the +scheduler-program thread died mid-cleanup, taking the completion event +with it. Seen on opd-40 2026-07-28 on every operator lock during a +program, the abort itself having completed correctly 6 min earlier. +""" + +import threading + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler import model +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus +from chimera.core.exceptions import ProgramExecutionAborted + +NOW_MJD = 60000.0 + + +def test_str_of_a_deleted_program_does_not_raise(tmp_path): + """Program.__str__ is used in every scheduler log line, including ones + inside exception handlers: it must never depend on live database + state.""" + engine = create_engine(f"sqlite:///{tmp_path / 'scheduler.db'}") + model.metadata.create_all(engine) + session_factory = sessionmaker(bind=engine) + + session = session_factory() + program = model.Program(name="WASP-145A", pi="", priority=1) + session.add(program) + session.commit() + + # somebody rebuilt the queue while this program was executing + other = session_factory() + other.query(model.Program).delete() + other.commit() + + session.commit() # expires every instance in the session + assert str(program).startswith("#") # must not raise ObjectDeletedError + + +class FakeSite: + def mjd(self): + return NOW_MJD + + +class Controller: + def __init__(self): + self.events = [] + self._config = {"stop_tracking_on_program_end": False, "telescope": None} + + def __getitem__(self, key): + return self._config[key] + + def get_site(self): + return FakeSite() + + def program_begin(self, program_id): + pass + + def program_complete(self, program_id, status, message=None): + self.events.append((status, message)) + + def state_changed(self, new, old): + pass + + +class DeletedRowProgram: + """A program whose row disappears mid-run: formatting it then reloads + the row, exactly as an expired SQLAlchemy instance does.""" + + id = 1 + start_at = 0.0 + valid_for = -1 + + def __init__(self): + self.deleted = False + + def __str__(self): + if self.deleted: + raise RuntimeError("Instance has been deleted") + return "#1 WASP-145A pi: #actions: 3" + + +class Scheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + # SequentialScheduler logs the task it is closing out + str(task) + + +class AbortingExecutor: + """The queue is rewritten, then the abort the operator asked for lands.""" + + def __init__(self, program): + self.program = program + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + self.program.deleted = True + raise ProgramExecutionAborted() + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def test_aborting_a_deleted_program_keeps_the_machine_alive(): + program = DeletedRowProgram() + controller = Controller() + machine = Machine(Scheduler(), AbortingExecutor(program), controller) + + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + + assert not machine._worker.is_alive() + assert controller.events == [(SchedulerStatus.ABORTED, "Aborted by user.")], ( + "the abort's completion event was lost with the thread" + ) + assert machine.state() == State.OFF + assert threading.active_count() >= 1 diff --git a/tests/chimera/controllers/test_scheduler_expired_program.py b/tests/chimera/controllers/test_scheduler_expired_program.py new file mode 100644 index 00000000..7f10fdc1 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_expired_program.py @@ -0,0 +1,134 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A program past its validity window must be finished without running. + +The expired branch reported program_complete(OK, "not valid anymore") and +then fell through to execute the program anyway - a second, contradictory +program_complete at the end, and anything listening to the first one (robobs +advances its queue on it) raced a program that was still running. +""" + +import pytest + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus + +NOW_MJD = 60000.0 +ONE_HOUR = 1.0 / 24.0 + + +class FakeSite: + def mjd(self): + return NOW_MJD + + +class Controller: + def __init__(self): + self.begun = [] + self.events = [] + self._config = {"site": "/Site/0", "stop_tracking_on_program_end": False} + + def __getitem__(self, key): + return self._config[key] + + def get_proxy(self, location): + return FakeSite() + + def get_site(self): + return FakeSite() + + def program_begin(self, program_id): + self.begun.append(program_id) + + def program_complete(self, program_id, status, message=None): + self.events.append((status, message)) + + def state_changed(self, new, old): + pass + + +class Scheduler: + def __init__(self): + self.done_calls = [] + + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + self.done_calls.append((task, error)) + + +class Executor: + def __init__(self): + self.executed = [] + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + self.executed.append(program) + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + """_process only uses the session to merge/commit; keep the DB out.""" + + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def _run(start_at, valid_for): + controller = Controller() + scheduler = Scheduler() + executor = Executor() + machine = Machine(scheduler, executor, controller) + program = type("P", (), {"id": 1, "start_at": start_at, "valid_for": valid_for})() + + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + assert not machine._worker.is_alive(), "the program thread never finished" + return controller, scheduler, executor, machine + + +def test_expired_program_is_finished_without_running(): + # start_at one hour ago, valid for one minute + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, 60.0) + + assert executor.executed == [], "an expired program was executed" + assert controller.begun == [] + assert controller.events == [(SchedulerStatus.OK, "Program not valid anymore.")], ( + f"expected a single completion, got {controller.events}" + ) + assert len(scheduler.done_calls) == 1, "the expired program was not marked done" + assert machine.state() == State.IDLE, "the machine was not released" + + +def test_late_program_without_window_still_runs(): + # valid_for < 0 means no expiry: a late start must still execute, once + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, -1.0) + + assert len(executor.executed) == 1 + assert controller.begun == [1] + assert controller.events == [(SchedulerStatus.OK, None)] + + +def test_late_program_inside_its_window_still_runs(): + # one hour late, valid for two hours + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, 7200.0) + + assert len(executor.executed) == 1 + assert controller.events == [(SchedulerStatus.OK, None)] diff --git a/tests/chimera/controllers/test_scheduler_machine_stop.py b/tests/chimera/controllers/test_scheduler_machine_stop.py new file mode 100644 index 00000000..cc001d37 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_machine_stop.py @@ -0,0 +1,147 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# SPDX-FileCopyrightText: 2006-present Paulo Henrique Silva + +"""A START requested while executor.stop() runs must not be discarded. + +executor.stop() blocks until the running action gives up; a START in that +window only sets the state variable, and dropping unconditionally to OFF +afterwards threw the request away. +""" + +import threading + +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State + + +class Controller: + def state_changed(self, new, old): + pass + + +class Scheduler: + """Records that the machine acted on a START.""" + + def __init__(self): + self.rescheduled = threading.Event() + + def reschedule(self, machine): + self.rescheduled.set() + + def __next__(self): + return None + + +class Executor: + """stop() takes time, and an operator asks for a START while it runs.""" + + def __init__(self): + self.machine = None + self.stopped = threading.Event() + + def __start__(self): + pass + + def stop(self): + self.stopped.set() + self.machine.state(State.START) + + +def test_start_requested_during_stop_is_honoured(): + executor = Executor() + scheduler = Scheduler() + machine = Machine(scheduler, executor, Controller()) + executor.machine = machine + machine.daemon = True + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + # let it settle into OFF, then ask it to stop the running program + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + machine.state(State.STOP) + + assert executor.stopped.wait(10), "executor.stop() was never called" + assert scheduler.rescheduled.wait(10), ( + "the START requested during executor.stop() was discarded" + ) + + machine.state(State.SHUTDOWN) + thread.join(timeout=10) + + +class PlainScheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + +class PlainExecutor: + def __start__(self): + pass + + def stop(self): + pass + + +def test_off_waits_for_the_worker_to_die(): + """STOP must not report OFF while the program thread is alive. + + On opd-40 the machine declared OFF while an exposure loop survived for + 52 minutes behind a closed dome: OFF is only honest once the worker is + dead, and the machine says STOPPING until then. + """ + release = threading.Event() + machine = Machine(PlainScheduler(), PlainExecutor(), Controller()) + machine.daemon = True + + worker = threading.Thread(target=lambda: release.wait(30), daemon=True) + machine._worker = worker + worker.start() + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + + # while the worker lives the machine must hold STOPPING, never OFF + threading.Event().wait(1.0) + assert worker.is_alive() + assert machine.state() == State.STOPPING + + release.set() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + assert machine.state() == State.OFF + + machine.state(State.SHUTDOWN) + thread.join(timeout=10) + + +def test_worker_completion_cannot_rearm_a_stopped_machine(): + """A program's own completion or error must not re-arm a stopped machine. + + On opd-40 the surviving program eventually died with an error and its + completion path flipped the stopped machine OFF -> IDLE, re-arming the + night. Terminal transitions are conditional on BUSY. + """ + machine = Machine(PlainScheduler(), PlainExecutor(), Controller()) + + machine.state(State.STOPPING) + assert machine.transition(State.BUSY, State.IDLE) is False + assert machine.state() == State.STOPPING + + machine.state(State.BUSY) + assert machine.transition(State.BUSY, State.IDLE) is True + assert machine.state() == State.IDLE diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py new file mode 100644 index 00000000..7fe22d27 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -0,0 +1,362 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A start() arriving while a program runs must not fork a second execution. + +START overwrites BUSY, so the machine re-entered IDLE, next(scheduler) handed +back the SAME still-unfinished program and _process forked another thread. +""" + +import threading + +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State + + +class Controller: + def state_changed(self, new, old): + pass + + +class Scheduler: + """Always offers the same program, as a real queue does until it finishes.""" + + def __init__(self, program): + self.program = program + + def reschedule(self, machine): + pass + + def __next__(self): + return self.program + + +class Executor: + def __start__(self): + pass + + def stop(self): + pass + + +def test_start_while_busy_does_not_duplicate(monkeypatch): + started = [] + release = threading.Event() + + program = type("P", (), {"id": 1, "start_at": None})() + machine = Machine(Scheduler(program), Executor(), Controller()) + machine.daemon = True + + def fake_process(prog): + started.append(prog) + t = threading.Thread(target=lambda: release.wait(10), daemon=True) + machine._worker = t + t.start() + + monkeypatch.setattr(machine, "_process", fake_process) + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.START) # first ingestion + for _ in range(100): + if started: + break + threading.Event().wait(0.05) + assert len(started) == 1 + + for _ in range(5): # robobs queues more + machine.state(State.START) + threading.Event().wait(0.1) + + assert len(started) == 1, f"program forked {len(started)} concurrent executions" + + release.set() + machine.state(State.SHUTDOWN) + + +def test_machine_recovers_when_the_worker_exits(): + """The worker signals IDLE from inside itself, just before exiting. + + A guard that sleeps on the condition variable misses that wakeup (it is + delivered before the sleep) and the machine parks forever. + """ + picked = [] + program = type("P", (), {"id": 1, "start_at": None})() + + class OneShotScheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return program if len(picked) < 2 else None + + machine = Machine(OneShotScheduler(), Executor(), Controller()) + machine.daemon = True + + def fake_process(prog): + picked.append(prog) + + def body(): + # signal IDLE from inside the worker, as the real one does, and + # stay alive a moment longer - that window is the race: the + # machine sees IDLE while this thread is still running + machine.state(State.IDLE) + threading.Event().wait(0.4) + + t = threading.Thread(target=body, daemon=True) + machine._worker = t + t.start() + + machine._process = fake_process + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + machine.state(State.START) + + # the machine must go on to pick the SECOND program, not park + for _ in range(120): + if len(picked) >= 2: + break + threading.Event().wait(0.05) + + assert len(picked) >= 2, "machine parked after the worker exited" + machine.state(State.SHUTDOWN) + + +def test_start_survives_a_slow_abort(): + """A --start must be acted on even while executor.stop() is still running. + + stop() blocks until the running action yields (a full camera readout); + run inline it froze the machine for that long. The machine waits on the + WORKER, never on the abort call, so a hung abort cannot park it either. + """ + rescheduled = threading.Event() + releasing = threading.Event() + + class SlowExecutor: + def __start__(self): + pass + + def stop(self): + releasing.wait(30) # the camera readout + + class Scheduler: + def reschedule(self, machine): + rescheduled.set() + + def __next__(self): + return None + + machine = Machine(Scheduler(), SlowExecutor(), Controller()) + machine.daemon = True + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + threading.Event().wait(0.1) + machine.state(State.START) # operator asks while the abort blocks + + assert rescheduled.wait(5), "the START was not acted on during the abort" + + releasing.set() + machine.state(State.SHUTDOWN) + + +def test_stop_cancels_a_program_waiting_for_its_slew_time(): + """--stop must release a program that is only counting down. + + executor.stop() aborts the CURRENT action, but a program waiting for a + future start_at has not started any: it sat in an uninterruptible + time.sleep() and held the machine until its slew time. + """ + machine = Machine( + type( + "S", (), {"reschedule": lambda self, m: None, "__next__": lambda self: None} + )(), + Executor(), + Controller(), + ) + + finished = threading.Event() + + def waiter(): + # what _process does while counting down to slew_at: poll state and + # block on the machine's wake-up Condition, which state() notifies on + # every change, so STOP/SHUTDOWN breaks the wait at once + wake = machine._Machine__wake_up_call + while machine.state() not in (State.STOP, State.STOPPING, State.SHUTDOWN): + with wake: + wake.wait(1.0) + finished.set() + + machine._worker = threading.Thread(target=waiter, daemon=True) + machine._worker.start() + threading.Event().wait(0.1) + assert machine._worker.is_alive() + + # run() sets OFF as its very first action, so the machine has to be up + # before STOP is requested or it is simply overwritten + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + + assert finished.wait(10), "the waiting program was not released by STOP" + machine.state(State.SHUTDOWN) + + +def test_sequential_runs_in_time_order(): + """A queue with baked start times must execute chronologically. + + Priority-only ordering let a future-timed program hold the machine with + the whole night queued behind it: start_at orders execution, priority + only breaks ties. + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="morning-flat", priority=0, start_at=61244.35)) + session.add(Program(name="science", priority=-21, start_at=61243.93)) + session.add(Program(name="evening-flat", priority=0, start_at=61243.86)) + session.commit() + + scheduler = SequentialScheduler() + scheduler.reschedule(type("M", (), {"wake_up": staticmethod(lambda: None)})()) + + order = [] + while True: + program = next(scheduler) + if program is None: + break + order.append(program.name) + assert order == ["evening-flat", "science", "morning-flat"], order + + session.query(Program).delete() + session.commit() + + +def test_sequential_keeps_file_order_for_untimed_programs(): + """A hand-written queue must run in the order it was written. + + `chimera-sched --new` bakes no start_at and usually one priority, so + the ordering fell through to whatever the database returned - the + targets came out in the reverse of the YAML (2026-07-20). + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + for name in ("first", "second", "third"): + session.add(Program(name=name, priority=0)) + session.commit() + + scheduler = SequentialScheduler() + scheduler.reschedule(type("M", (), {"wake_up": staticmethod(lambda: None)})()) + + order = [] + while True: + program = next(scheduler) + if program is None: + break + order.append(program.name) + assert order == ["first", "second", "third"], order + + session.query(Program).delete() + session.commit() + + +def test_finished_program_is_not_rerun_after_reschedule(): + """A reschedule while a program runs re-enqueues that same program (its + finished flag is only written at completion); the pop must skip the + stale entry or the night replays it.""" + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="focus", priority=-2, start_at=61244.19)) + session.add(Program(name="science", priority=-19, start_at=61244.20)) + session.commit() + + machine = type("M", (), {"wake_up": staticmethod(lambda: None)})() + scheduler = SequentialScheduler() + scheduler.reschedule(machine) + running = next(scheduler) + assert running.name == "focus" + + # robobs hands over another program mid-run: the rebuilt queue holds + # the still-unfinished focus again + scheduler.reschedule(machine) + + # the focus completes, exactly as machine._process does it + worker_session = Session() + task = worker_session.merge(running) + scheduler.done(task) + worker_session.commit() + + picked = next(scheduler) + assert picked is not None, "the science program was lost with the stale entry" + assert picked.name == "science", f"finished program re-ran: {picked.name}" + + session = Session() + session.query(Program).delete() + session.commit() + + +def test_reschedule_during_run_does_not_unbalance_the_queue(): + """Popping the stale re-enqueued program must not raise on the queue. + + reschedule() rebuilds run_queue and re-enqueues the still-running program; + done() and the stale-entry pop then both accounted for it with + task_done(), and the second call raised 'task_done() called too many + times', killing the machine thread mid-night. A single-program night is + the minimal case that drove the counter negative. + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="only", priority=0, start_at=61244.0)) + session.commit() + + machine = type("M", (), {"wake_up": staticmethod(lambda: None)})() + scheduler = SequentialScheduler() + scheduler.reschedule(machine) + running = next(scheduler) + assert running.name == "only" + + # a re-plan mid-run re-enqueues the still-unfinished program + scheduler.reschedule(machine) + + worker_session = Session() + scheduler.done(worker_session.merge(running)) + worker_session.commit() + + # the stale entry is skipped and the night ends cleanly, no ValueError + assert next(scheduler) is None + + session = Session() + session.query(Program).delete() + session.commit() diff --git a/tests/chimera/controllers/test_scheduler_point_altaz.py b/tests/chimera/controllers/test_scheduler_point_altaz.py new file mode 100644 index 00000000..7de9cd45 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_point_altaz.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# SPDX-FileCopyrightText: 2006-present Paulo Henrique Silva + +"""A point action given alt/az must reach the telescope as plain degrees. + +Position.alt/az return floats (Coord.deg) while .ra/.dec return Coords, so +the handler calling .to_d() on all four raised +"'float' object has no attribute 'to_d'" for every alt/az point. +""" + +from types import SimpleNamespace + +import pytest + +from chimera.controllers.scheduler.handlers import PointHandler +from chimera.util.position import Position + + +class FakeTelescope: + def __init__(self): + self.alt_az = None + + def slew_to_alt_az(self, alt, az): + self.alt_az = (alt, az) + + +def _action(position): + return SimpleNamespace( + target_ra_dec=None, + target_alt_az=position, + target_name=None, + offset_ns=None, + offset_ew=None, + dome_tracking=None, + dome_az=None, + pa=None, + ) + + +def test_position_alt_az_are_floats(): + # the premise of the bug: these are NOT Coords + position = Position.from_alt_az(88.0, 89.0) + assert isinstance(position.alt, float) + assert isinstance(position.az, float) + + +@pytest.fixture +def telescope(monkeypatch): + fake = FakeTelescope() + monkeypatch.setattr(PointHandler, "telescope", fake, raising=False) + monkeypatch.setattr(PointHandler, "dome", None, raising=False) + monkeypatch.setattr(PointHandler, "rotator", None, raising=False) + return fake + + +def test_point_alt_az_slews(telescope): + PointHandler.process(_action(Position.from_alt_az(88.0, 89.0))) + assert telescope.alt_az == (88.0, 89.0) diff --git a/tests/chimera/controllers/test_scheduler_stop.py b/tests/chimera/controllers/test_scheduler_stop.py new file mode 100644 index 00000000..2ae8f49b --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_stop.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# End-to-end regression tests for scheduler stop semantics (lna40 incident, +# opd-40 2026-07-25): a sched.stop() must actually stop the running program, +# including after a start() during BUSY re-queued the still-unfinished +# program — on master that spawned a second worker that outlived every stop +# and kept the camera exposing for 52 min behind a closed dome. + +import time + +import pytest + +# the scheduler DB is redirected to a temp file in tests/chimera/conftest.py, +# before any module import binds the model engine +from chimera.controllers.imageserver.imageserver import ImageServer +from chimera.controllers.scheduler.controller import Scheduler +from chimera.controllers.scheduler.model import Expose, Program, Session +from chimera.controllers.scheduler.states import State +from chimera.core.site import Site +from chimera.instruments.fakecamera import FakeCamera +from chimera.instruments.fakefilterwheel import FakeFilterWheel + + +@pytest.fixture +def system(manager, wait_for, tmp_path): + manager.add_class(Site, "fake") + manager.add_class( + ImageServer, "fake", {"images_dir": str(tmp_path), "httpd": False} + ) + manager.add_class(FakeCamera, "fake") + manager.add_class(FakeFilterWheel, "fake") + manager.add_class(Scheduler, "fake") + + sched = manager.get_proxy("/Scheduler/0") + camera = manager.get_proxy("/Camera/0") + + # wait for the controller's control() tick to start the machine thread + assert wait_for(lambda: sched.state() is not None, 10) + + # a clean program table for each test + session = Session() + for program in session.query(Program).all(): + session.delete(program) + session.commit() + + yield sched, camera, tmp_path + + +def add_program(tmp_path, frames=8, exptime=1): + program = Program(name="stop-test", pi="tester") + program.actions = [ + Expose( + frames=frames, + exptime=exptime, + shutter="OPEN", + image_type="object", + object_name="stop-test", + # absolute path: no ImageServer lookup surprises + filename=str(tmp_path / "frame"), + ) + ] + session = Session() + session.add(program) + session.commit() + + +class TestSchedulerStop: + def test_stop_single_worker(self, system, wait_for): + """One worker, stop mid-multi-frame-expose stops for real.""" + sched, camera, tmp_path = system + add_program(tmp_path) + + sched.start() + assert wait_for(camera.is_exposing, 15), "exposure never started" + + sched.stop() + + assert wait_for(lambda: sched.state() == State.OFF, 10) + assert not camera.is_exposing(), "camera still exposing after OFF" + + # and it must STAY stopped + time.sleep(3) + assert not camera.is_exposing(), "camera exposing again after stop" + assert sched.state() == State.OFF + + def test_stop_after_restart_during_busy(self, system, wait_for): + """The incident case: a start() while BUSY re-queues the unfinished + program; the following stop must still kill everything and OFF must + mean the camera is no longer being driven.""" + sched, camera, tmp_path = system + # long enough that a surviving execution could not finish naturally + # inside the assertion windows below (the incident program had 100s + # of frames) + add_program(tmp_path, frames=60, exptime=1) + + sched.start() + assert wait_for(camera.is_exposing, 15), "exposure never started" + + # what robobs does when (re)submitting programs mid-night + sched.start() + time.sleep(1) + + sched.stop() + + assert wait_for(lambda: sched.state() == State.OFF, 10), ( + "machine never reached OFF after stop" + ) + + # OFF must be honest: the camera stops and STAYS stopped + assert not camera.is_exposing(), ( + "camera still exposing after the scheduler reported OFF" + ) + assert not wait_for(camera.is_exposing, 5), ( + "camera started exposing again after the stop (a second worker survived)" + ) + assert sched.state() == State.OFF, "scheduler re-armed itself after stop" diff --git a/tests/chimera/controllers/test_scheduler_tracking_stop.py b/tests/chimera/controllers/test_scheduler_tracking_stop.py new file mode 100644 index 00000000..01d1655a --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_tracking_stop.py @@ -0,0 +1,235 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""The scheduler leaves the mount idle when a program ends, however it ended. + +This used to live in robobs, on a detached thread fired at program_complete: +the stop blocked on the telescope lock behind the *next* program's slew and +landed after it, untracking the freshly acquired target (TheSkyX re-enables +tracking at slew end). The fix is ordering, not timing: the stop runs inline +on the program thread, before program_complete, while the machine is still +BUSY. These tests give stop_tracking a delay standing in for that lock wait, +so a stop moved back onto its own thread loses the race and fails them. +""" + +import threading +import time + +import pytest + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus +from chimera.core.exceptions import ProgramExecutionAborted, ProgramExecutionException + +#: stands in for the telescope lock held by the next program's slew; long +#: enough that a detached stop lands after program_complete +LOCK_WAIT = 0.2 + + +class FakeTelescope: + """Records into the shared timeline so ordering can be asserted.""" + + def __init__(self, timeline, tracking=True, delay=LOCK_WAIT): + self.timeline = timeline + self.tracking = tracking + self.delay = delay + self.calls = [] + + def is_tracking(self): + self.calls.append("is_tracking") + return self.tracking + + def stop_tracking(self): + time.sleep(self.delay) # the lock wait + self.calls.append("stop_tracking") + self.timeline.append("stop_tracking") + self.tracking = False + + +class FakeSite: + def mjd(self): + return 60000.0 + + +class Controller: + """Stands in for the Scheduler ChimeraObject.""" + + def __init__(self, timeline, telescope, stop_tracking=True): + self.timeline = timeline + self.telescope = telescope + self.events = [] + self._config = { + "site": "/Site/0", + "telescope": "/Telescope/0", + "stop_tracking_on_program_end": stop_tracking, + } + + def __getitem__(self, key): + return self._config[key] + + def get_proxy(self, location): + return FakeSite() if location == "/Site/0" else self.telescope + + def get_site(self): + return FakeSite() + + def program_begin(self, program_id): + pass + + def program_complete(self, program_id, status, message=None): + self.events.append(status) + self.timeline.append("program_complete") + + def state_changed(self, new, old): + pass + + +class Scheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + pass + + +class Executor: + """Ends the program the way the test asks for.""" + + def __init__(self, raises=None): + self.raises = raises + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + if self.raises: + raise self.raises + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + """_process only uses the session to merge/commit; keep the DB out.""" + + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def _build(telescope_factory, stop_tracking=True, raises=None): + timeline = [] + telescope = telescope_factory(timeline) + controller = Controller(timeline, telescope, stop_tracking=stop_tracking) + machine = Machine(Scheduler(), Executor(raises=raises), controller) + return timeline, telescope, controller, machine + + +def _run(machine): + program = type("P", (), {"id": 1, "start_at": 0.0, "valid_for": -1.0})() + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + assert not machine._worker.is_alive(), "the program thread never finished" + + +@pytest.mark.parametrize( + "raises,expected_status", + [ + (None, SchedulerStatus.OK), + (ProgramExecutionException("boom"), SchedulerStatus.ERROR), + (ProgramExecutionAborted(), SchedulerStatus.ABORTED), + ], + ids=["completed", "errored", "aborted"], +) +def test_tracking_stops_before_the_program_is_reported_done(raises, expected_status): + """However the program ended, the stop must precede program_complete, + which is what releases the next program.""" + timeline, telescope, controller, machine = _build(FakeTelescope, raises=raises) + + _run(machine) + + assert telescope.calls == ["is_tracking", "stop_tracking"] + assert controller.events == [expected_status] + assert timeline == ["stop_tracking", "program_complete"], ( + f"tracking was stopped out of order: {timeline}" + ) + + +def test_the_stop_happens_while_the_machine_is_still_busy(): + """While the machine is BUSY no other program can be picked, so no slew + can be in flight competing for the telescope lock.""" + seen_state = [] + holder = [] + + class WatchingTelescope(FakeTelescope): + def stop_tracking(self): + super().stop_tracking() + seen_state.append(holder[0].state()) + + timeline, telescope, controller, machine = _build(WatchingTelescope) + holder.append(machine) + + _run(machine) + + assert seen_state == [State.BUSY], ( + f"tracking was stopped in state {seen_state}, not BUSY: the next " + f"program could already have been picked" + ) + + +def test_the_stop_is_inline_not_detached(): + """A detached stop is what made it land late; it must be done on return.""" + timeline, telescope, controller, machine = _build(FakeTelescope) + before = threading.active_count() + + _run(machine) + + # commanded by the time the program thread has finished, with no helper + # thread left behind to deliver it later + assert telescope.calls == ["is_tracking", "stop_tracking"] + assert threading.active_count() <= before + + +def test_option_off_leaves_tracking_alone(): + timeline, telescope, controller, machine = _build( + FakeTelescope, stop_tracking=False + ) + + _run(machine) + + assert telescope.calls == [] + assert timeline == ["program_complete"] + + +def test_a_mount_that_is_not_tracking_is_not_commanded(): + timeline, telescope, controller, machine = _build( + lambda tl: FakeTelescope(tl, tracking=False) + ) + + _run(machine) + + assert telescope.calls == ["is_tracking"] + + +def test_a_failing_telescope_does_not_break_the_program(): + """Best effort: the program still has to be reported complete.""" + + class BrokenTelescope(FakeTelescope): + def is_tracking(self): + raise RuntimeError("no link to the mount") + + timeline, telescope, controller, machine = _build(BrokenTelescope) + + _run(machine) + + assert controller.events == [SchedulerStatus.OK]