model-platform/schedule/tests/test_layering.py

"""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.application.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


# ── execution/runners: stage-4 merge boundary smoke (F4) ──────────────────


def test_limited_log_under_limit_returns_encoded_unchanged() -> None:
    from schedule.execution.runners import notebook

    assert notebook._limited_log("hello") == b"hello"


def test_limited_log_empty_returns_sentinel() -> None:
    from schedule.execution.runners import notebook

    assert notebook._limited_log("") == b"execution produced no console output\n"


def test_limited_log_above_max_bytes_truncates_with_suffix() -> None:
    from schedule.execution.runners import notebook

    sentinel = b"\n[log truncated by scheduler worker]\n"
    overflowing = "x" * (notebook.MAX_LOG_BYTES + 1)
    truncated = notebook._limited_log(overflowing)
    assert len(truncated) == notebook.MAX_LOG_BYTES
    assert truncated.endswith(sentinel)


@pytest.mark.asyncio
async def test_execute_artifact_unsupported_script_type_raises() -> None:
    """``script_type`` outside ``{"notebook", "python"}`` must raise —
    otherwise the worker would silently drop a malformed run into the
    queue and the node would hang.
    """
    from schedule.execution.runners import notebook

    with pytest.raises(ValueError, match="unsupported script_type"):
        await notebook.execute_artifact(
            source=b"print('hi')\n",
            run_id="01RUN0000000000000000000A",
            node_run_id="01NODE000000000000000000A",
            script_type="bogus",
            artifact_path="x.py",
            arguments=[],
            timeout_seconds=30,
        )