10Backward/tests/test_cron_render.py

from __future__ import annotations

import shutil
import subprocess
from pathlib import Path

import pytest

from tenbackward.config import _is_valid_hhmm


REPO_ROOT = Path(__file__).resolve().parents[1]
ENTRYPOINT = REPO_ROOT / "entrypoint.sh"


def test_valid_hhmm() -> None:
    assert _is_valid_hhmm("00:00")
    assert _is_valid_hhmm("09:30")
    assert _is_valid_hhmm("23:59")


@pytest.mark.parametrize("value", ["24:00", "9:00", "09:60", "9", "9:00:00", "ab:cd", "", None])
def test_invalid_hhmm(value) -> None:
    assert not _is_valid_hhmm(value)  # type: ignore[arg-type]


def _entrypoint_env(extra: dict[str, str]) -> dict[str, str]:
    env = {
        "PATH": "/usr/bin:/bin:/usr/sbin:/sbin",
        "DEBIAN_FRONTEND": "noninteractive",
        "HOME": "/tmp",
        "LANG": "C.UTF-8",
    }
    env.update(extra)
    return env


def _run_entrypoint_cron_render(extra: dict[str, str]) -> str:
    raise NotImplementedError


def _extract_rendered_cron(env: dict[str, str]) -> tuple[int, str, str]:
    """Run the entrypoint with a fake `install` that records the cron file,
    and a stub `cron` that exits immediately so the entrypoint doesn't hang.
    Returns (returncode, stderr, installed-cron-contents).
    """
    import tempfile

    with tempfile.TemporaryDirectory() as fake_bin_dir, tempfile.TemporaryDirectory() as tmp:
        fake_cron = Path(tmp) / "tenbackward.cron"

        # Fake `install`: real syntax is `install [-m mode] [-o owner] [-g group] SRC DST`.
        # We extract the last positional arg as DST and the last existing source before it.
        fake_install = Path(fake_bin_dir) / "install"
        fake_install.write_text(
            "#!/bin/bash\n"
            "src=\"\"\n"
            "dst=\"\"\n"
            "while [ $# -gt 0 ]; do\n"
            "  case \"$1\" in\n"
            "    -m|-o|-g) shift; shift;;\n"
            "    *) src=\"${src:-$1}\"; dst=\"$1\"; shift;;\n"
            "  esac\n"
            "done\n"
            "cp \"$src\" \"" + str(fake_cron) + "\"\n"
        )
        fake_install.chmod(0o755)

        # Fake `cron` exits immediately so we never hang waiting on cron.
        fake_cron_bin = Path(fake_bin_dir) / "cron"
        fake_cron_bin.write_text("#!/bin/bash\nexit 0\n")
        fake_cron_bin.chmod(0o755)

        env = dict(env)
        env["PATH"] = f"{fake_bin_dir}:/usr/bin:/bin"

        proc = subprocess.run(
            ["/bin/bash", str(ENTRYPOINT)],
            capture_output=True,
            text=True,
            env=env,
            timeout=10,
            check=False,
        )

        rendered = fake_cron.read_text() if fake_cron.exists() else ""
        return proc.returncode, proc.stderr, rendered


@pytest.mark.skipif(not shutil.which("bash"), reason="bash not available")
def test_entrypoint_renders_run_at_into_cron() -> None:
    """Drive the entrypoint with a hermetic PATH so we can observe the rendered cron line.

    The entrypoint hard-codes `/app/crontab/install-cron.sh` (correct in the
    Docker image). For the host-side test we override PATH so `install` and
    `cron` are stubbed, but we still need `/app/crontab/install-cron.sh` — we
    bind-mount it via a bind-mount emulation by overriding the env var
    `PATH` and having the install-cron.sh check fall back: instead, we patch
    the entrypoint with an in-place substitution for the test.
    """
    tmp_cron_dest = _write_substituted_entrypoint()
    try:
        env = _entrypoint_env({
            "MASTODON_BASE_URL": "https://mastodon.example",
            "MASTODON_ACCESS_TOKEN": "x",
            "SITE_URL": "https://blog.example.com",
            "HASHTAGS": "#throwback",
            "RUN_AT": "09:00",
            "TZ": "Europe/Berlin",
        })

        returncode, stderr, rendered = _extract_rendered_cron(env, tmp_cron_dest)

        assert returncode == 0, (returncode, stderr)
        assert "00 09 * * *" in rendered, rendered
        assert "python -m tenbackward" in rendered
        assert "TZ=Europe/Berlin" in rendered
    finally:
        tmp_cron_dest.unlink(missing_ok=True)


def _write_substituted_entrypoint():
    """Copy entrypoint.sh to a temp file with `/app/crontab/install-cron.sh`
    rewritten to the repo-relative path so the host can execute it."""
    import tempfile
    src = ENTRYPOINT.read_text()
    sub = src.replace("/app/crontab/install-cron.sh", str(REPO_ROOT / "crontab" / "install-cron.sh"))
    tmp = Path(tempfile.mkstemp(prefix="entrypoint-", suffix=".sh")[1])
    tmp.write_text(sub)
    tmp.chmod(0o755)
    return tmp


def _extract_rendered_cron(env: dict[str, str], entrypoint_path: Path) -> tuple[int, str, str]:
    """Run the (substituted) entrypoint with stubs for `install` and `cron`."""
    import tempfile

    with tempfile.TemporaryDirectory() as fake_bin_dir, tempfile.TemporaryDirectory() as tmp:
        sidecar = Path(tmp) / "tenbackward.cron"

        fake_install = Path(fake_bin_dir) / "install"
        fake_install.write_text(
            "#!/bin/bash\n"
            "src=\"\"; dst=\"\"\n"
            "while [ $# -gt 0 ]; do\n"
            "  case \"$1\" in -m|-o|-g) shift; shift;; *) src=\"${src:-$1}\"; dst=\"$1\"; shift;; esac\n"
            "done\n"
            "cp \"$src\" \"" + str(sidecar) + "\"\n"
            "mkdir -p \"$(dirname \"$dst\")\" 2>/dev/null\n"
            "cp \"$src\" \"$dst\" 2>/dev/null || true\n"
        )
        fake_install.chmod(0o755)

        fake_cron_bin = Path(fake_bin_dir) / "cron"
        fake_cron_bin.write_text("#!/bin/bash\nexit 0\n")
        fake_cron_bin.chmod(0o755)

        env = dict(env)
        env["PATH"] = f"{fake_bin_dir}:/usr/bin:/bin"

        proc = subprocess.run(
            ["/bin/bash", str(entrypoint_path)],
            capture_output=True,
            text=True,
            env=env,
            timeout=10,
            check=False,
        )

        rendered = sidecar.read_text() if sidecar.exists() else ""
        return proc.returncode, proc.stderr, rendered


@pytest.mark.skipif(not shutil.which("bash"), reason="bash not available")
def test_install_cron_rejects_unresolved_placeholder(tmp_path: Path) -> None:
    script = REPO_ROOT / "crontab" / "install-cron.sh"

    bad = tmp_path / "bad.cron"
    bad.write_text("__RUN_AT__ * * * * /bin/false\n")
    proc = subprocess.run(
        ["/bin/bash", str(script), str(bad)],
        capture_output=True, text=True, timeout=10,
    )
    assert proc.returncode != 0
    assert "__RUN_AT__" in proc.stderr

    good = tmp_path / "good.cron"
    good.write_text("SHELL=/bin/bash\n5 9 * * * /bin/true\n")
    proc2 = subprocess.run(
        ["/bin/bash", str(script), str(good)],
        capture_output=True, text=True, timeout=10,
    )
    assert proc2.returncode == 0, proc2.stderr