From 7476c27d814285294bfbb98d3ea61b5beb7ba911 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:52:31 +0800 Subject: [PATCH] test(schedule): add layer-boundary smoke tests (stage 7) Pin the new five-layer contract so a later refactor can't silently break an import surface or lifecycle: - domain: ExecutionResult defaults, terminal/failed state-set invariants, naive_utc normalization (import-side-effect-free) - infrastructure.storage: SchedulerStorageClient base64 upload via httpx.MockTransport (no live server; base_url required for relative URL) - scheduling: CronScheduler start/close lifecycle, global trigger cleared - application: SchedulerService wires cron + orchestrator + worker, handler dispatch table points at the wired NodeExecutor - execution: schedule.notebook_runner shim re-exports the real main 25 schedule tests green; schedule/pyproject.toml untouched. Co-Authored-By: Claude --- schedule/tests/test_layering.py | 177 ++++++++++++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 schedule/tests/test_layering.py diff --git a/schedule/tests/test_layering.py b/schedule/tests/test_layering.py new file mode 100644 index 0000000..d44ef40 --- /dev/null +++ b/schedule/tests/test_layering.py @@ -0,0 +1,177 @@ +"""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