move_detect/tests/test_deploy_artifacts.py

"""Guard: the systemd unit + env template must stay in sync with what the daemon
actually reads. If someone renames an env var in edge/config.py or edge/daemon.py
(or edits the unit), this fails instead of shipping a broken bundle.
"""
import os

DEPLOY = os.path.join(os.path.dirname(__file__), "..", "deploy", "systemd")
ENV_EXAMPLE = os.path.join(DEPLOY, "move-detect-edge.env.example")
UNIT = os.path.join(DEPLOY, "move-detect-edge.service")
SETUP_ONLINE = os.path.join(os.path.dirname(__file__), "..", "deploy", "setup_online.sh")
INSTALL_OFFLINE = os.path.join(os.path.dirname(__file__), "..", "deploy", "bundle",
                               "install_offline.sh")
BUILD_OFFLINE = os.path.join(os.path.dirname(__file__), "..", "deploy", "bundle",
                             "build_offline_bundle.sh")
PILOT_RUNBOOK = os.path.join(os.path.dirname(__file__), "..", "docs", "pilot-runbook.md")

# Keys the daemon consumes (edge/config.py:load_config + edge/daemon.py:main).
# HAILO_HEF, not MOVENET_ONNX: the delivered bundle carries only the HEF (the MoveNet ONNX has
# no reachable upstream), so the template must ship a backend whose model actually exists. An
# uncommented MOVENET_ONNX pointing at a file that was never bundled is a unit that cannot start.
REQUIRED_ENV = {
    "DEVICE_ID", "DEVICE_SECRET", "CAFE24_URL", "CAMERA_ID",
    "HAILO_HEF", "UVC_DEVICE", "TARGET_FPS", "QUEUE_DB",
}


def _env_keys(text):
    keys = set()
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        keys.add(line.split("=", 1)[0].strip())
    return keys


def test_env_example_lists_every_required_key():
    with open(ENV_EXAMPLE, encoding="utf-8") as f:
        keys = _env_keys(f.read())
    missing = REQUIRED_ENV - keys
    assert not missing, f"env template missing required keys: {sorted(missing)}"


def test_env_example_values_have_no_inline_comments():
    # systemd EnvironmentFile= does NOT strip inline comments — a trailing "# ..." on
    # a KEY=value line becomes part of the value and breaks int()/float()/path/secret.
    with open(ENV_EXAMPLE, encoding="utf-8") as f:
        offenders = []
        for line in f.read().splitlines():
            s = line.strip()
            if not s or s.startswith("#") or "=" not in s:
                continue
            value = s.split("=", 1)[1]
            if "#" in value:
                offenders.append(s)
    assert not offenders, f"inline comments in env values break systemd: {offenders}"


def _directives(text, section=None):
    """Active Key=value lines of a systemd unit, comments stripped. Restricted to one
    [Section] when given. Comments must be excluded: the unit deliberately *documents*
    directives it must not carry (e.g. ReadWritePaths=), and a naive substring check
    matches the warning against re-adding them."""
    out, cur = [], None
    for line in text.splitlines():
        s = line.strip()
        if not s or s.startswith(("#", ";")):   # systemd honors both as comment leaders
            continue
        if s.startswith("[") and s.endswith("]"):
            cur = s[1:-1]
            continue
        if section is None or cur == section:
            out.append(s)
    return "\n".join(out)


def test_systemd_unit_wiring():
    # Every assertion binds to the [Section] that gives it meaning. A section-less check
    # would pass on a directive moved into the wrong section, which is precisely the bug
    # class that made StartLimitIntervalSec= a silent no-op for this unit.
    with open(UNIT, encoding="utf-8") as f:
        raw = f.read()
    svc = _directives(raw, "Service")
    assert "EnvironmentFile=/etc/move-detect/edge.env" in svc
    assert "PYTHONPATH=/opt/move-detect/src" in svc
    assert "python -m edge" in svc            # runs the package entrypoint
    assert "Restart=always" in svc            # unattended resilience
    assert "WantedBy=multi-user.target" in _directives(raw, "Install")

    # Type=simple reports the start job done the instant fork() returns, so an exec-step
    # failure (226/NAMESPACE, 216/GROUP, 217/USER, 203/EXEC) never reaches
    # `systemctl enable --now` and setup_online.sh prints "Done." over a dead daemon.
    assert "Type=exec" in svc
    assert "Type=simple" not in svc

    # StartLimitIntervalSec= is a [Unit] key. systemd 252's parser table has
    # Unit.StartLimitIntervalSec but only Service.StartLimitInterval (no "Sec"), so in
    # [Service] it is an unknown key: silently ignored, leaving the default 5-starts/10s
    # limit to defeat Restart=always.
    assert "StartLimitIntervalSec=0" in _directives(raw, "Unit")
    assert "StartLimit" not in svc

    # ReadWritePaths=/var/lib/move-detect is dead config: StateDirectory= already emits a
    # recursive read-write bind mount for that same path, which outranks a READWRITE entry
    # and drops it before any path is resolved. Keep it out so it cannot be re-added as a
    # "fix" for a state-dir problem it is incapable of causing or solving.
    assert "StateDirectory=move-detect" in svc
    assert "ReadWritePaths=" not in svc


def test_online_installer_guards_apt_and_surfaces_start_failures():
    with open(SETUP_ONLINE, encoding="utf-8") as f:
        sh = f.read()

    # [1/8] runs under `set -euo pipefail`. On a freshly-booted Pi, apt-daily and
    # unattended-upgrades hold /var/lib/apt/lists/lock and /var/lib/dpkg/lock-frontend;
    # a bare apt-get then exits 100 with "Unable to lock directory /var/lib/apt/lists/"
    # and kills the install. -o DPkg::Lock::Timeout does not cover the lists lock, so the
    # wait has to live in the shell.
    assert "apt_retry" in sh
    for bare in ("\napt-get update", "\napt-get install"):
        assert bare not in sh, "apt-get must go through apt_retry()"

    # The lock-detection grep matches apt's C-locale strings, but apt gettext-translates its
    # errors and sudo passes LANG/LC_* through. Without the locale pin, a ko_KR Pi misreads
    # lock contention as a real error and fail-fasts — the exact bug apt_retry exists to fix.
    assert "LC_ALL=C apt-get" in sh

    # [8/8] must print WHY the daemon did not come up rather than dying under set -e on an
    # opaque "Job for move-detect-edge.service failed". Assert the real invocation including
    # flags: a bare "journalctl -xeu move-detect-edge" is also present in a human-facing echo
    # hint, so the loose form passes even if the actual diagnostic call is deleted.
    assert "journalctl -xeu move-detect-edge -n 50 --no-pager" in sh
    assert "systemctl is-active --quiet move-detect-edge" in sh
    # is-active is phase-dependent (a crash at T<=~0.2s is back in "active" by T+RestartSec);
    # NRestarts is not, so it must stay as the AND alongside it.
    assert "NRestarts" in sh


def test_onnxruntime_is_optional_but_a_pose_backend_is_mandatory():
    with open(SETUP_ONLINE, encoding="utf-8") as f:
        sh = f.read()

    # onnxruntime is imported inside MoveNetPoseAdapter.__init__ (edge/adapters/
    # movenet_pose.py) and the Hailo path (edge/daemon.py) never imports the module at all —
    # so it is the CPU-MoveNet-only dep. It publishes no sdist and no 32-bit-ARM wheel, so
    # pip can legitimately find zero candidates and exit non-zero. On the same hard-fail line
    # as requests/opencv under `set -e`, that killed the whole install on a Hailo-8 Pi that
    # never needed it.
    assert "pip\" install requests onnxruntime opencv-python-headless" not in sh, \
        "onnxruntime must not share a hard-fail pip line with the mandatory deps"

    # cv2 and requests stay mandatory: edge/snapshot.py imports cv2 at module level on the
    # unconditional startup chain, and the daemon builds its server clients with no injected
    # HTTP fn, so `import requests` fires at startup for every backend.
    assert "install requests opencv-python-headless" in sh

    # ...but the installer must not then hand over a Pi with NO working backend. build_pose()
    # runs outside the daemon's try/except, so a missing backend is an uncaught traceback on
    # every Restart=always cycle, forever. Fail at install time, naming the fix.
    assert "no usable pose backend" in sh
    # Probe capability, not intent: a HEF on disk does not mean HailoRT is installed, and a
    # pip exit code does not mean the .so loads.
    assert "import hailo_platform" in sh
    assert "import onnxruntime" in sh


def test_online_installer_stages_a_delivered_hef_and_never_downloads_one():
    with open(SETUP_ONLINE, encoding="utf-8") as f:
        sh = f.read()
    # On-device model fetch was removed: it failed two ways at once. A partial repo copy onto the
    # Pi made `python -m edge.tools.fetch_*` raise ModuleNotFoundError, which the `|| echo WARN`
    # fallbacks mislabeled as "download failed" (sending the operator after a phantom network/board
    # problem); and the MoveNet URL is dead upstream. Models now ship IN the delivery, staged on the
    # BUILD machine — air-gap rule "build pulls, target doesn't". The installer must NOT fetch.
    assert "edge.tools.fetch_hailo_hef" not in sh
    assert "edge.tools.fetch_movenet" not in sh
    # It PLACES the HEF the delivery carries (models/yolov8s_pose.hef), or an explicit HAILO_HEF path.
    assert "$REPO_ROOT/models/yolov8s_pose.hef" in sh
    # An explicit HAILO_HEF that does not exist is a typo, not a request to fall back.
    assert "does not exist" in sh
    # The failure table must still show whether the BOARD answered: that field separates "no HEF"
    # from "driver installed, reboot pending" from "no Hailo at all" — three different fixes.
    assert "hailortcli fw-control identify" in sh
    assert "board=" in sh


def test_reinstall_reconciles_pose_backend_in_the_env_file():
    with open(SETUP_ONLINE, encoding="utf-8") as f:
        sh = f.read()
    # Re-running IS the Hailo flow (install -> reboot for the driver -> re-run), so [6/8] can
    # reach a different verdict than last time. Leaving edge.env wholly unchanged left run #1's
    # POSE_BACKEND=movenet in place while [6/8] printed "hailo" — a crash loop the installer
    # called a success.
    assert "re-run reconciled" in sh
    assert "s|^POSE_BACKEND=.*|POSE_BACKEND=$POSE_BACKEND|" in sh
    # sed -i / >> can leave the file root:root; the unit reads it as movedetect via
    # EnvironmentFile=, so the group must be restored or the start fails on permissions.
    assert 'chown root:"$SVC_USER" "$ETC/edge.env"' in sh


def test_offline_installer_does_not_require_onnxruntime_to_verify_the_install():
    with open(INSTALL_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # Same brittleness offline: a bundle built for a Hailo Pi need not carry onnxruntime, and
    # this import check would fail the install for a package the daemon never loads.
    assert "import edge.daemon, onnxruntime, cv2, numpy, requests" not in sh


def test_offline_installer_gates_the_pose_backend():
    with open(INSTALL_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # The offline installer must not hand over a Pi whose configured POSE_BACKEND cannot load
    # (a Hailo-only bundle installed with the template's POSE_BACKEND=movenet -> crash loop).
    # Probe capability and set the working backend, or hard-fail — like setup_online's [6/8].
    assert "import hailo_platform" in sh
    assert "import onnxruntime" in sh
    assert "no usable pose backend" in sh
    # It must WRITE the reconciled backend into the installed env, not leave the template's.
    assert "POSE_BACKEND=$POSE_BACKEND" in sh or "POSE_BACKEND=${POSE_BACKEND}" in sh


def test_bundled_runbook_covers_the_airgap_pi_path_not_just_the_laptop_pilot():
    # This runbook is staged into the offline bundle and is the systemd unit's Documentation=
    # target, so a Pi field operator opens it. It must direct them to the air-gapped install path,
    # and must NOT still claim the air-gap bundle does not exist (which used to send the operator to
    # `pip install` on a closed network at commissioning).
    with open(PILOT_RUNBOOK, encoding="utf-8") as f:
        doc = f.read()
    assert ("install_offline" in doc) or ("deploy/README.md" in doc)
    assert "폐쇄망 전체 번들 아님" not in doc


def test_bundled_runbook_is_what_the_unit_documents_and_the_bundle_ships():
    # Guard the wiring: the unit points at pilot-runbook.md and the bundle stages it. If either the
    # filename or the staging changes, this fails instead of shipping a dangling Documentation=.
    with open(UNIT, encoding="utf-8") as f:
        unit = f.read()
    with open(BUILD_OFFLINE, encoding="utf-8") as f:
        build = f.read()
    assert "pilot-runbook.md" in unit
    assert "pilot-runbook.md" in build


def test_offline_installer_tells_the_operator_to_probe_cameras_before_enabling():
    with open(INSTALL_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # The offline path cannot auto-probe at install time (CAMERAS is only set when the operator
    # edits edge.env afterwards), so it must at least INSTRUCT the camera check before enable —
    # otherwise a mis-indexed camera produces a green unit monitoring one camera less, the exact
    # failure setup_online.sh's [8/9] probe was added to prevent. The probe command must precede
    # the enable instruction.
    assert "edge.probe" in sh
    probe_at = sh.index("edge.probe")
    enable_at = sh.index("systemctl enable --now move-detect-edge", probe_at)
    assert probe_at < enable_at  # probe instruction comes before the enable instruction


def test_offline_bundle_stages_the_probe_module_for_the_camera_check():
    with open(BUILD_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # The bundle stages the whole edge/ package, so `python -m edge.probe` is available on the
    # air-gapped Pi. (A stale hand-built dist/ that predates edge/probe.py must be regenerated.)
    assert 'cp -r "$REPO_ROOT/edge" "$STAGE/src/edge"' in sh
    probe = os.path.join(os.path.dirname(__file__), "..", "edge", "probe.py")
    assert os.path.isfile(probe)  # the module the instruction invokes actually exists in source


def test_offline_bundle_fetches_and_stages_the_hailo_hef():
    with open(BUILD_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # A Hailo-8 delivery runs off the HEF; the offline bundle must carry it, fetched on the build
    # machine (the air-gapped Pi cannot). Without this the installer's backend gate finds no HEF.
    assert "edge.tools.fetch_hailo_hef" in sh
    assert "yolov8s_pose.hef" in sh  # staged into the bundle's models/


def test_offline_bundle_treats_movenet_onnx_as_optional():
    with open(BUILD_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # The MoveNet source can be unreachable and a Hailo bundle is complete without the CPU
    # fallback, so the fetch must be GUARDED (inside an `if`) — never an unconditional line that
    # aborts the whole build under `set -e`.
    assert "fetch_movenet ); then" in sh
    assert "Hailo-only bundle" in sh


def test_offline_installer_copies_the_hef_into_place():
    with open(INSTALL_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # install_offline used to copy only *.onnx, so a HEF in the bundle never reached
    # $PREFIX/models and the Hailo backend gate ([5/6]) could never see it.
    assert 'cp "$HERE/models/"*.hef' in sh


def test_offline_bundle_emits_pinned_integrity_manifest():
    with open(BUILD_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # Air-gap supply-chain: the target can't re-resolve from PyPI, so the build must pin exact
    # versions, hash EVERY artifact (not just wheels/models — the daemon source, unit and installer
    # too), and ship an SBOM. GPG signing is optional (SIGN_KEY).
    assert "find . -type f" in sh and "sha256sum > SHA256SUMS" in sh   # hashes every staged file
    assert "! -name SHA256SUMS" in sh                                  # except the manifest itself
    assert "requirements.lock" in sh          # exact name==version pins for a reproducible rebuild
    assert "sbom.json" in sh
    assert "CycloneDX" in sh
    assert 'for mdl in "$STAGE"/models/*' in sh   # SBOM includes the ML model weights, not just wheels
    assert 'if [ -n "${SIGN_KEY:-}" ]; then' in sh  # optional detached signature


def test_offline_installer_verifies_integrity_before_install():
    with open(INSTALL_OFFLINE, encoding="utf-8") as f:
        sh = f.read()
    # A truncated/tampered bundle must fail loudly BEFORE any system mutation — not after rm -rf
    # has destroyed the previous install and the pip step has already run.
    assert "sha256sum -c SHA256SUMS" in sh
    assert "integrity check failed" in sh
    # Ordering (the substance of the fix): the verify must come BEFORE the first $PREFIX mutation
    # and before the offline wheel install.
    verify_at = sh.index("sha256sum -c SHA256SUMS")
    assert verify_at < sh.index('rm -rf "$PREFIX/src/edge')   # before destroying the old tree
    assert verify_at < sh.index("install --no-index")          # before installing wheels
    # A missing SHA256SUMS must be FATAL (refuse), not a warn-and-proceed that installs blind.
    assert "refusing to install unverified code" in sh


# ---------------------------------------------------------------------------
# 2026-08-18: 셸 스크립트 줄바꿈/문법 가드.
# Windows 빌드머신에서 편집하다 CRLF 가 섞이면 bash 가 `{\r` 에서 죽는다 — 그러면 납품
# 산출물을 아예 만들 수 없다. cafe24 빌드 스크립트에는 이미 같은 가드가 있었고, Pi 쪽에는
# 없어서 실제로 이 사고가 났다.
# ---------------------------------------------------------------------------

import subprocess
SHELL_SCRIPTS = [SETUP_ONLINE, INSTALL_OFFLINE, BUILD_OFFLINE]
REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
# Git Bash 는 Windows 백슬래시 경로를 받으면 구분자를 먹어버린다. 저장소 상대 POSIX 경로를
# cwd=REPO_ROOT 로 넘긴다 (test_cafe24_package.py 가 이미 쓰는 방식).
SHELL_SCRIPTS_REL = [
    "deploy/setup_online.sh",
    "deploy/bundle/install_offline.sh",
    "deploy/bundle/build_offline_bundle.sh",
]


def test_shell_scripts_are_lf_only():
    for path in SHELL_SCRIPTS:
        raw = open(path, "rb").read()
        assert bytes([13, 10]) not in raw, (
            f"{os.path.basename(path)} 에 CRLF 가 있습니다. bash 가 여는 중괄호에서 죽습니다 — "
            "LF 로 저장하세요."
        )


def test_shell_scripts_are_syntactically_valid_bash():
    for rel in SHELL_SCRIPTS_REL:
        r = subprocess.run(["bash", "-n", rel], cwd=REPO_ROOT, capture_output=True, text=True)
        assert r.returncode == 0, "bash -n " + rel + " 실패:" + chr(10) + r.stderr