"""Layer-boundary smoke tests for the schedule service refactor. The 11 flat files under ``schedule/src/schedule/`` were reorganized into five subpackages (domain / scheduling / application / execution / infrastructure) with **zero behavior change**. These tests pin the new *boundaries* so a later refactor can't silently break an import surface or a lifecycle contract: - ``domain`` — pure constants / dataclass, import-side-effect-free - ``infrastructure.storage`` — SchedulerStorageClient uploads via the backend storage API (httpx MockTransport, no live server) - ``scheduling`` — CronScheduler start/close lifecycle - ``application`` — SchedulerService wires cron + orchestrator + worker - ``execution`` — the ``schedule.notebook_runner`` shim still points at the real runner (the worker's ``-m`` subprocess string depends on it) All mocks / tmp only — no MySQL, no real HTTP. """ from __future__ import annotations import json from datetime import UTC, datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import MagicMock import httpx import pytest from schedule.application.service import SchedulerService from schedule.domain.context import ( FAILED_NODE_STATES, TERMINAL_NODE_STATES, TERMINAL_RUN_STATES, naive_utc, ) from schedule.domain.execution import ExecutionResult from schedule.execution import NodeExecutor from schedule.infrastructure.storage import SchedulerStorageClient from schedule.scheduling.orchestrator import DispatchOrchestrator from schedule.scheduling.scheduler import CronScheduler # ── domain: pure types / constants ──────────────────────────────────────── def test_execution_result_has_sane_defaults() -> None: result = ExecutionResult( status="succeeded", exit_code=0, logs=b"ok\n", result=b"{}", result_file_name="out.json", result_content_type="application/json", ) assert result.status == "succeeded" assert result.exit_code == 0 # Optional error fields default to None, not to a sentinel. assert result.error_code is None assert result.error_message is None def test_terminal_and_failed_node_state_sets() -> None: assert isinstance(TERMINAL_NODE_STATES, frozenset) assert isinstance(FAILED_NODE_STATES, frozenset) assert "succeeded" in TERMINAL_NODE_STATES assert "failed" in TERMINAL_NODE_STATES # Every failed state must also be terminal — otherwise the DAG would # treat a dead node as still running. assert FAILED_NODE_STATES.issubset(TERMINAL_NODE_STATES) assert TERMINAL_RUN_STATES.issubset(TERMINAL_NODE_STATES) def test_naive_utc_normalizes_to_utc_naive() -> None: # 20:00 +08:00 == 12:00 UTC; the tz must be stripped afterwards. aware = datetime(2026, 8, 21, 20, 0, tzinfo=timezone(timedelta(hours=8))) assert naive_utc(aware) == datetime(2026, 8, 21, 12, 0) assert naive_utc(aware).tzinfo is None assert naive_utc(None) is None # ── infrastructure: storage client (no live server) ─────────────────────── def test_storage_client_create_object_uploads_base64() -> None: import base64 def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/internal/v1/objects" payload = json.loads(request.content) assert payload["usage_type"] == "run_log" assert payload["file_name"] == "run.log" assert payload["is_immutable"] is True # The raw bytes must round-trip through base64 in the JSON body. assert base64.b64decode(payload["content_base64"]) == b"hello\n" return httpx.Response( 200, json={"data": {"storage_object_id": "obj-1", "storage_uri": "s3://x"}}, ) transport = httpx.MockTransport(handler) async def upload() -> dict[str, object]: # base_url is required: a relative path + MockTransport fails to # normalize the response URL in this httpx version. async with httpx.AsyncClient( transport=transport, base_url="http://testserver" ) as http: storage = SchedulerStorageClient(http) return await storage.create_object( workspace_id="01WS0000000000000000000001", user_id="01USR0000000000000000000A", usage_type="run_log", file_name="run.log", content_type="text/plain", content=b"hello\n", idempotency_key="run:node:1", ) data = __import__("asyncio").run(upload()) assert data["storage_object_id"] == "obj-1" # ── scheduling: cron lifecycle (no MySQL touched) ────────────────────────── @pytest.mark.asyncio async def test_cron_scheduler_start_close_lifecycle() -> None: """start()/close() must not raise and must release the global trigger.""" scheduler = CronScheduler( session_factory=MagicMock(), database_url="sqlite://", on_trigger=MagicMock(), ) scheduler.start() await scheduler.close() # close() clears the module-level callback so a stopped service can't # dispatch persisted cron ticks. from schedule.scheduling import scheduler as scheduler_module assert scheduler_module._ACTIVE_TRIGGER is None # ── application: service assembly wires all three components ─────────────── @pytest.mark.asyncio @pytest.mark.filterwarnings( "ignore:coroutine .* was never awaited:RuntimeWarning" ) async def test_scheduler_service_wires_components_and_lifecycle() -> None: service = SchedulerService( session_factory=MagicMock(), storage_http_client=MagicMock(), object_store=MagicMock(), storage_client=MagicMock(), database_url="sqlite://", ) assert isinstance(service.cron, CronScheduler) assert isinstance(service.worker, NodeExecutor) assert isinstance(service.orchestrator, DispatchOrchestrator) # orchestrator's dispatch table must reference the wired worker handler. # (Bound methods create a fresh object per access, so compare __self__.) assert service.orchestrator._node_execute_handler.__self__ is service.worker await service.start() await service.close() assert service.orchestrator._loop_task is None # ── execution: the notebook_runner shim is the real runner ───────────────── def test_notebook_runner_shim_reexports_real_main() -> None: from schedule.execution.runners import notebook as real from schedule.notebook_runner import main as shim_main assert shim_main is real.main