Jyotisha/tests/test_beta_safety_net.py

"""Contracts for the closed-beta safety net. Fixtures are fictional."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = ROOT / ".gitea" / "workflows" / "backup-production-daily.yml"
RECOVERY_WORKFLOW = ROOT / ".gitea" / "workflows" / "create-production-recovery.yml"
RECOVERY_SCRIPT = ROOT / "deploy" / "run-production-recovery.sh"
PRUNE_SCRIPT = ROOT / "deploy" / "prune-production-daily-backups.sh"
HEALTH_SCRIPT = ROOT / "deploy" / "monitor" / "healthcheck-alert.sh"
DIGEST_SCRIPT = ROOT / "deploy" / "monitor" / "error-digest.sh"
FIND_SCRIPT = ROOT / "deploy" / "find-request-log.sh"
NOTIFY_SCRIPT = ROOT / "deploy" / "monitor" / "notify-feishu.sh"
SERVER_COMPOSE = ROOT / "deploy" / "docker-compose.server.yml"
POSTGRES_COMPOSE = ROOT / "deploy" / "docker-compose.postgres.yml"


def resolve_bash() -> str:
    found = shutil.which("bash")
    git_bash = Path(r"D:\Program Files\Git\bin\bash.exe")
    if os.name == "nt" and git_bash.exists():
        lowered = (found or "").lower()
        if not found or "system32" in lowered or "windowsapps" in lowered:
            return str(git_bash)
    if found:
        return found
    if git_bash.exists():
        return str(git_bash)
    return "bash"


BASH = resolve_bash()


def bash_path(path: Path) -> str:
    if os.name != "nt":
        return str(path)
    completed = subprocess.run(
        [BASH, "-lc", 'cygpath -u "$1"', "cygpath", str(path)],
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return completed.stdout.strip()


def run_bash(script: Path, extra: dict[str, str], args: list[str] | None = None) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env.pop("FEISHU_WEBHOOK_URL", None)
    env.pop("FEISHU_SIGN_SECRET", None)
    env.update(extra)
    env["JYOTISHA_PYTHON"] = bash_path(Path(sys.executable))
    return subprocess.run(
        [BASH, bash_path(script), *(args or [])],
        cwd=ROOT,
        env=env,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


def write_text(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8", newline="\n")


class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path != "/health":
            self.send_response(404)
            self.end_headers()
            return
        kind = getattr(self.server, "kind", "ok")
        if kind == "timeout":
            time.sleep(3)
        commit = getattr(self.server, "commit", "a" * 40)
        if kind == "500":
            status, body = 500, b'{"status":"ok"}'
        elif kind == "missing":
            status, body = 200, b"{}"
        elif kind == "degraded":
            status, body = 200, json.dumps({"status": "degraded", "deployment": {"gitCommit": commit}}).encode()
        else:
            status, body = 200, json.dumps({"status": "ok", "deployment": {"gitCommit": commit}}).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt: str, *args: object) -> None:
        return


class HookHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length)
        self.server.bodies.append(body)
        self.close_connection = True
        self.send_response(200)
        self.send_header("Content-Length", "2")
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(b"ok")

    def log_message(self, fmt: str, *args: object) -> None:
        return


def serve(handler: type[BaseHTTPRequestHandler]) -> tuple[ThreadingHTTPServer, threading.Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


def alert_lines(state: Path) -> list[str]:
    log = state / "alerts.log"
    if not log.exists():
        return []
    return [line for line in log.read_text(encoding="utf-8").splitlines() if line]


def test_daily_workflow_is_scheduled_from_main_and_reuses_recovery() -> None:
    text = WORKFLOW.read_text(encoding="utf-8")
    assert "\tif:" not in text
    assert "if: gitea.ref == 'refs/heads/main'" in text
    assert 'cron: "0 20 * * *"' in text
    assert "workflow_dispatch:" in text
    assert 'git show "$GITEA_SHA:deploy/run-production-recovery.sh"' in text
    assert "RECOVERY_BASENAME_PREFIX='production-daily'" in text
    assert "prune-production-daily-backups.sh" in text
    assert "production-daily-" in text
    assert 'artifact_retention="${ARTIFACT_RETENTION_DAYS:-14}"' in text
    assert 'local_retention="${LOCAL_RETENTION_DAYS:-7}"' in text
    assert 'process.env["INPUT_RETENTION-DAYS"]=process.env.ARTIFACT_RETENTION_DAYS' in text
    assert "set +x" in text
    assert "pg_dump" not in text
    assert "CHECK_RELEASE_GATE" not in text
    assert "observed_staging_sha" not in text
    assert "STAGING_URL" not in text
    assert "echo \"$SSH_PRIVATE_KEY" not in text
    assert "PRODUCTION_BACKUP_ENCRYPTION_KEY" not in text
    recovery = RECOVERY_WORKFLOW.read_text(encoding="utf-8")
    assert "observed_staging_sha" in recovery
    assert 'process.env["INPUT_RETENTION-DAYS"]="30"' in recovery


def test_recovery_script_keeps_manual_names_unless_daily_prefix() -> None:
    text = RECOVERY_SCRIPT.read_text(encoding="utf-8")
    assert 'backup_prefix="production-pre-migration"' in text
    assert 'production-daily) backup_prefix="production-daily"' in text
    assert "openssl enc -aes-256-cbc -salt -pbkdf2" in text
    assert "usage_percent < 70" in text
    assert "flock -n 9" in text


def test_prune_deletes_only_expired_daily_files(tmp_path: Path) -> None:
    backup = tmp_path / "backups"
    nested = backup / "nested"
    nested.mkdir(parents=True)
    now = 1_700_000_000
    limit = 7 * 86400
    names = {
        "production-daily-20261001T000000Z.dump.enc": now - limit - 60,
        "production-daily-20261001T000000Z-restore-verify.json": now - limit - 60,
        "production-daily-20261007T000000Z.dump.enc": now - limit,
        "production-daily-20261007T000000Z-restore-verify.json": now - limit,
        "production-daily-20261014T000000Z.dump.enc": now - 3600,
        "production-daily-20261014T000000Z-restore-verify.json": now - 3600,
        "production-pre-migration-20261001T000000Z.dump.enc": now - limit - 3600,
        "production-pre-migration-20261001T000000Z-restore-verify.json": now - limit - 3600,
        "production-daily-notes.txt": now - limit - 3600,
        "production-daily-20261001T000000Z.dump.enc.bak": now - limit - 3600,
        "unrelated.txt": now - limit - 3600,
    }
    for name, epoch in names.items():
        path = backup / name
        write_text(path, "fictional\n")
        subprocess.run(
            [BASH, "-lc", 'touch -d "@$1" "$2"', "touch", str(epoch), bash_path(path)],
            check=True,
        )
    write_text(nested / "production-daily-20261001T000000Z.dump.enc", "nested\n")
    completed = run_bash(PRUNE_SCRIPT, {}, [bash_path(backup), "7", str(now)])
    assert completed.returncode == 0, completed.stderr
    deleted = completed.stdout.splitlines()
    assert deleted == ["production-daily-20261001T000000Z.dump.enc"]
    assert not (backup / "production-daily-20261001T000000Z.dump.enc").exists()
    assert not (backup / "production-daily-20261001T000000Z-restore-verify.json").exists()
    for name in names:
        if name.startswith("production-daily-20261001T000000Z.dump") or name.startswith(
            "production-daily-20261001T000000Z-restore"
        ):
            continue
        assert (backup / name).exists(), name
    assert (nested / "production-daily-20261001T000000Z.dump.enc").exists()


def test_health_alerts_after_three_failures_and_recovery_once(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "500"
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        for _ in range(2):
            completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
            assert completed.returncode == 0, completed.stderr
        assert alert_lines(state) == []
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        lines = alert_lines(state)
        assert len(lines) == 1
        assert "errorCode=health_down" in lines[0]
        assert "count=3" in lines[0]
        assert "route=/api/health" in lines[0]
        server.kind = "ok"
        server.commit = "b" * 40
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        lines = alert_lines(state)
        assert len(lines) == 2
        assert "errorCode=recovered" in lines[1]
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert len(alert_lines(state)) == 2
    finally:
        server.shutdown()


def test_health_timeout_and_missing_fields_count_as_failures(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "timeout"
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "1",
    }
    try:
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert alert_lines(state) == []
        server.kind = "missing"
        run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert any("errorCode=health_down" in line for line in alert_lines(state))
    finally:
        server.shutdown()


def test_commit_change_alerts_once_outside_release_window(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "ok"
    server.commit = "a" * 40
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert alert_lines(state) == []
        server.commit = "b" * 40
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        lines = alert_lines(state)
        assert len(lines) == 1
        assert "errorCode=commit_changed" in lines[0]
        assert "b" * 40 not in lines[0]
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert len(alert_lines(state)) == 1
        marker = state / "release-window"
        state.mkdir(parents=True, exist_ok=True)
        write_text(marker, "open\n")
        server.commit = "c" * 40
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert len(alert_lines(state)) == 1
    finally:
        server.shutdown()


def test_disk_alerts_once_and_no_webhook_stays_local(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    write_text(
        bin_dir / "df",
        "#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' \"overlay 100 90 10 ${FAKE_DISK_PERCENT:?}% /\"\n",
    )
    subprocess.run([BASH, "-lc", 'chmod +x "$1/df"', "chmod", bash_path(bin_dir)], check=True)
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "FAKE_DISK_PERCENT": "81",
    }
    env.pop("FEISHU_WEBHOOK_URL", None)
    env.pop("FEISHU_SIGN_SECRET", None)
    completed = subprocess.run(
        [BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
        cwd=ROOT,
        env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    assert completed.returncode == 0, completed.stderr
    assert "errorCode=disk_high" in alert_lines(state)[0]
    assert "count=81" in alert_lines(state)[0]
    env["FAKE_DISK_PERCENT"] = "81"
    again = subprocess.run(
        [BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
        cwd=ROOT,
        env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    assert again.returncode == 0, again.stderr
    assert len(alert_lines(state)) == 1
    env["FAKE_DISK_PERCENT"] = "10"
    recovered = subprocess.run(
        [BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
        cwd=ROOT,
        env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    assert recovered.returncode == 0, recovered.stderr
    assert len(alert_lines(state)) == 2
    assert "errorCode=recovered" in alert_lines(state)[1]


def test_feishu_payload_keeps_only_allowed_fields(tmp_path: Path) -> None:
    hook, _thread = serve(HookHandler)
    hook.bodies = []
    health, _thread = serve(HealthHandler)
    health.kind = "500"
    state = tmp_path / "state"
    secret = "fixture-sign-secret"
    env_file = tmp_path / "monitor.env"
    write_text(env_file, f"FEISHU_WEBHOOK_URL=http://127.0.0.1:{hook.server_address[1]}/hook\nFEISHU_SIGN_SECRET={secret}\n")
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(env_file),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{health.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        for _ in range(3):
            completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
            assert completed.returncode == 0, completed.stderr
        assert len(hook.bodies) == 1
        payload = json.loads(hook.bodies[0].decode("utf-8"))
        assert set(payload) == {"timestamp", "sign", "msg_type", "content"}
        assert payload["msg_type"] == "text"
        text = payload["content"]["text"]
        assert secret not in text
        assert secret not in hook.bodies[0].decode("utf-8")
        assert "errorCode=health_down" in text
        assert "@" not in text
        expected = subprocess.run(
            [
                sys.executable,
                "-c",
                "import base64,hashlib,hmac,os,sys; ts=sys.argv[1]; secret=sys.argv[2]; key=f'{ts}\\n{secret}'.encode(); print(base64.b64encode(hmac.new(key, digestmod=hashlib.sha256).digest()).decode())",
                payload["timestamp"],
                secret,
            ],
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()
        assert payload["sign"] == expected
    finally:
        hook.shutdown()
        health.shutdown()


def uuid(index: int) -> str:
    return f"00000000-0000-4000-8000-{index:012d}"


def test_error_digest_drops_personal_text_and_respects_thresholds(tmp_path: Path) -> None:
    safe = [
        f"2026-10-15T00:00:{index:02d}Z web [agent-observability] {json.dumps({'requestId': uuid(index), 'errorCode': 'settlement_failed'})}"
        for index in range(1, 11)
    ]
    unsafe = [
        "2026-10-15T00:01:00Z web person@example.com errorCode=should_drop",
        "2026-10-15T00:01:01Z web note 1900-01-01 birth_date",
        "2026-10-15T00:01:02Z web 对话内容 今天工作不顺",
    ]
    fixture = tmp_path / "logs.txt"
    write_text(fixture, "\n".join([*safe, *unsafe]) + "\n")
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_LOG_FIXTURE": bash_path(fixture),
        "JYOTISHA_NOW": "2026-10-15T01:00:00Z",
    }
    below = tmp_path / "below.txt"
    write_text(below, "\n".join(safe[:9]) + "\n")
    quiet = run_bash(DIGEST_SCRIPT, {**env, "JYOTISHA_LOG_FIXTURE": bash_path(below)}, ["--window"])
    assert quiet.returncode == 0, quiet.stderr
    assert alert_lines(state) == []
    completed = run_bash(DIGEST_SCRIPT, env, ["--window"])
    assert completed.returncode == 0, completed.stderr
    lines = alert_lines(state)
    assert len(lines) == 1
    assert "errorCode=settlement_failed" in lines[0]
    assert "count=10" in lines[0]
    assert lines[0].count(",") == 4
    assert uuid(6) not in lines[0]
    assert "person@example.com" not in lines[0]
    assert "1900-01-01" not in lines[0]
    assert "今天工作不顺" not in lines[0]
    assert "should_drop" not in lines[0]

    fivexx = tmp_path / "fivexx.txt"
    write_text(fivexx, "\n".join(f"2026-10-15T00:02:{index:02d}Z api status=502" for index in range(20)) + "\n")
    state_5xx = tmp_path / "state5"
    completed = run_bash(
        DIGEST_SCRIPT,
        {**env, "JYOTISHA_MONITOR_STATE_DIR": bash_path(state_5xx), "JYOTISHA_LOG_FIXTURE": bash_path(fivexx)},
        ["--window"],
    )
    assert completed.returncode == 0, completed.stderr
    joined = "\n".join(alert_lines(state_5xx))
    assert "errorCode=http_5xx" in joined
    assert "count=20" in joined


def test_daily_digest_summarizes_previous_shanghai_day(tmp_path: Path) -> None:
    fixture = tmp_path / "daily.txt"
    write_text(
        fixture,
        "\n".join(
            [
                '2026-10-14T02:00:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000001","errorCode":"settlement_failed"}',
                '2026-10-15T01:30:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000002","errorCode":"settlement_failed"}',
            ]
        )
        + "\n",
    )
    state = tmp_path / "state"
    completed = run_bash(
        DIGEST_SCRIPT,
        {
            "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
            "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
            "JYOTISHA_LOG_FIXTURE": bash_path(fixture),
            "JYOTISHA_NOW": "2026-10-15T01:00:00Z",
        },
        ["--daily"],
    )
    assert completed.returncode == 0, completed.stderr
    lines = alert_lines(state)
    assert len(lines) == 1
    assert "count=1" in lines[0]
    assert "00000000-0000-4000-8000-000000000001" in lines[0]
    assert "00000000-0000-4000-8000-000000000002" not in lines[0]


def test_find_request_log_sorts_and_rejects_injection(tmp_path: Path) -> None:
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    request_id = "11111111-2222-4333-8444-555555555555"
    write_text(log_dir / "caddy.log", f"2026-10-15T03:00:00Z caddy {request_id}\n")
    write_text(log_dir / "web.log", f"2026-10-15T01:00:00Z web {request_id}\nignore\n")
    write_text(log_dir / "api.log", f"2026-10-15T02:00:00Z api {request_id}\n")
    canary = tmp_path / "canary.txt"
    write_text(canary, "keep\n")
    found = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h"])
    assert found.returncode == 0, found.stderr
    assert found.stdout.splitlines() == [
        f"2026-10-15T01:00:00Z web {request_id}",
        f"2026-10-15T02:00:00Z api {request_id}",
        f"2026-10-15T03:00:00Z caddy {request_id}",
    ]
    rejected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["; rm -rf /", "--since", "24h"])
    assert rejected.returncode == 2
    assert canary.read_text(encoding="utf-8") == "keep\n"
    injected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["$(rm canary.txt)", "--since", "1h"])
    assert injected.returncode == 2
    assert canary.exists()
    bad_since = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h;rm"])
    assert bad_since.returncode == 2


def service_logging_blocks(text: str) -> dict[str, str]:
    blocks: dict[str, str] = {}
    service: str | None = None
    in_services = False
    capturing = False
    buf: list[str] = []
    for line in text.splitlines():
        if line.startswith("services:"):
            in_services = True
            continue
        if in_services and line and not line.startswith(" "):
            in_services = False
        if not in_services:
            continue
        if line.startswith("  ") and not line.startswith("    ") and line.endswith(":"):
            if capturing and service is not None:
                blocks[service] = "\n".join(buf)
            service = line.strip()[:-1]
            capturing = False
            buf = []
            continue
        if service and line == "    logging:":
            capturing = True
            buf = [line]
            continue
        if capturing:
            if line.startswith("    ") and not line.startswith("      "):
                blocks[service] = "\n".join(buf)
                capturing = False
                buf = []
            else:
                buf.append(line)
    if capturing and service is not None:
        blocks[service] = "\n".join(buf)
    return blocks


def test_compose_logging_is_present_and_identical() -> None:
    expected = "\n".join(
        [
            "    logging:",
            "      driver: json-file",
            "      options:",
            '        max-size: "20m"',
            '        max-file: "10"',
        ]
    )
    expected_names = {
        SERVER_COMPOSE: {"api", "web", "caddy"},
        POSTGRES_COMPOSE: {"postgres", "migrator", "migration-checker"},
    }
    for path, names in expected_names.items():
        blocks = service_logging_blocks(path.read_text(encoding="utf-8"))
        assert set(blocks) == names
        for name in names:
            assert blocks[name] == expected, name


def test_shell_scripts_pass_bash_n() -> None:
    for script in (PRUNE_SCRIPT, HEALTH_SCRIPT, DIGEST_SCRIPT, FIND_SCRIPT, NOTIFY_SCRIPT, RECOVERY_SCRIPT):
        completed = subprocess.run(
            [BASH, "-n", bash_path(script)],
            capture_output=True,
            text=True,
            encoding="utf-8",
        )
        assert completed.returncode == 0, script.name + completed.stderr