afb46fdab4
Follow-up to the salvaged registration contract: - share one _raise_if_cron_registration_error() helper for the two byte-identical dashboard 424 except-blocks (web_server + cron router, via the existing late() seam) - add endpoint-level 424 coverage for /api/cron/blueprints/instantiate (previously only the sync worker was tested) - give chat/CLI surfaces a human-facing user_message() (job name, no exception class name) and add a recovery hint (pause/resume or update re-registers via provider reconcile) to the model/REST message - consolidate five inline provider test doubles into one ABC-subclassing make_cron_provider conftest factory; the web_server test double now subclasses CronScheduler so an ABC rename fails loudly - narrow the wrapper facade to keyword-only (**kwargs) and route the tool's partial-failure return through tool_error()
73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""Cron-test fixtures.
|
|
|
|
Provides a default ``HERMES_MODEL`` for cron run_job tests so each one
|
|
doesn't have to spell out a model. The global conftest blanks
|
|
HERMES_MODEL hermetically; without this autouse fixture every cron test
|
|
that exercises ``run_job`` would hit the fail-fast guard added in
|
|
``cron/scheduler.py`` (see issue #23979) and have to be rewritten.
|
|
|
|
Tests that specifically need ``HERMES_MODEL`` unset — model-resolution
|
|
edge cases — call ``monkeypatch.delenv("HERMES_MODEL", raising=False)``
|
|
inside the test, which overrides this fixture's value for that scope.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_cron_provider():
|
|
"""Factory for minimal CronScheduler test doubles.
|
|
|
|
``make_cron_provider(register_job=...)`` returns a real ``CronScheduler``
|
|
subclass instance whose ``register_job`` is the given callable — so tests
|
|
exercising the creation-registration contract share one stub instead of
|
|
redefining inline spy/failing classes, and an ABC rename breaks them
|
|
loudly instead of silently passing a duck-type.
|
|
"""
|
|
from cron.scheduler_provider import CronScheduler
|
|
|
|
def _make(register_job=None, name="stub"):
|
|
class _StubProvider(CronScheduler):
|
|
@property
|
|
def name(self): # pragma: no cover - trivial
|
|
return name
|
|
|
|
def start(self, stop_event, **kw): # pragma: no cover - unused
|
|
pass
|
|
|
|
def register_job(self, job):
|
|
if register_job is not None:
|
|
return register_job(job)
|
|
return None
|
|
|
|
return _StubProvider()
|
|
|
|
return _make
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _default_cron_test_model(monkeypatch):
|
|
"""Pin a default HERMES_MODEL so cron run_job tests have a resolvable model."""
|
|
monkeypatch.setenv("HERMES_MODEL", "test-cron-default-model")
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_session_context_vars():
|
|
"""Restore session ContextVars around cron tests that call run_job directly.
|
|
|
|
Production confines each cron run to a copied context, but direct unit tests
|
|
share the pytest context. ``run_job`` intentionally clears ordinary session
|
|
variables to explicit empty values, which would otherwise shadow legacy env
|
|
fallbacks used by later approval tests in the same process.
|
|
"""
|
|
from gateway.session_context import _UNSET, _VAR_MAP
|
|
|
|
def _reset_all():
|
|
for var in _VAR_MAP.values():
|
|
var.set(_UNSET)
|
|
|
|
_reset_all()
|
|
yield
|
|
_reset_all()
|