planet/scripts/harness/test_error_diagnostics.py

"""Verify catalog-driven diagnostics without touching Docker or live services."""

import os
from pathlib import Path
import re
import shlex
import subprocess
import tempfile
import unittest

from test_database_startup import run_shell, shell_function

ROOT = Path(__file__).resolve().parents[2]
MODULE = ROOT / "scripts/lib/error-diagnostics.zsh"


def catalog(language: str) -> dict[str, list[str]]:
    text = (ROOT / f"docs/technical/{language}/ops-runbook.md").read_text()
    entries = {}
    for line in text.splitlines():
        if line.startswith("| P_"):
            fields = [field.strip() for field in line.split("|")[1:-1]]
            assert len(fields) == 4, line
            assert fields[0] not in entries, fields[0]
            entries[fields[0]] = fields[1:]
    return entries


class ErrorDiagnosticsTests(unittest.TestCase):
    def diagnose(self, message: str, evidence: str = "") -> list[str]:
        with tempfile.TemporaryDirectory() as folder:
            log = Path(folder) / "build.log"
            log.write_text(evidence)
            result = subprocess.run(
                [
                    "zsh",
                    "-f",
                    "-c",
                    f"""
                    SCRIPT_DIR={shlex.quote(str(ROOT))}
                    source {shlex.quote(str(MODULE))}
                    planet_error_record "$1" "$2"
                """,
                    "test-diagnostics",
                    message,
                    str(log),
                ],
                text=True,
                capture_output=True,
                timeout=10,
                check=True,
            )
        return result.stdout.strip().split("\t")

    def test_bilingual_catalogs_have_identical_codes_and_matching_fragments(self) -> None:
        zh, en = catalog("zh"), catalog("en")
        self.assertGreater(len(zh), 0)
        self.assertEqual(list(zh), list(en))
        self.assertEqual(list(zh)[-1], "P_UNKNOWN")
        for code, (signals, reason, action) in zh.items():
            with self.subTest(code=code):
                self.assertEqual(signals, en[code][0])
                self.assertTrue(reason and action)
                self.assertNotEqual(reason, en[code][1])

    def test_every_recorded_fragment_returns_the_exact_catalog_wording(self) -> None:
        for code, (signals, reason, action) in catalog("zh").items():
            for signal in signals.split(";"):
                with self.subTest(code=code, signal=signal):
                    self.assertEqual(self.diagnose(signal), [code, reason, action])

    def test_specific_build_evidence_precedes_generic_summary(self) -> None:
        for evidence, code in (
            (
                'Head "https://registry-1.docker.io/v2/test": dial tcp [::1]:443: i/o timeout',
                "P_NETWORK_TIMEOUT",
            ),
            (
                'Head "https://ghcr.io/v2/test": net/http: TLS handshake timeout',
                "P_NETWORK_TIMEOUT",
            ),
            ("failed to solve: lookup registry.example: no such host", "P_DNS"),
            ("X509: certificate signed by unknown authority", "P_TLS_CERT"),
            (
                "failed to solve: unexpected status from HEAD request: 429 Too Many Requests",
                "P_REGISTRY_RATE_LIMIT",
            ),
            ("new unexpected build error", "P_BUILD_FAILED"),
        ):
            with self.subTest(code=code):
                result = self.diagnose("AI Provider 镜像构建失败", evidence + " TOKEN_SENTINEL")
                self.assertEqual(result[0], code)
                self.assertNotIn("TOKEN_SENTINEL", " ".join(result))

    def test_unknown_error_does_not_claim_a_network_or_proxy_cause(self) -> None:
        self.assertEqual(self.diagnose("an unrecognized failure")[0], "P_UNKNOWN")

    def test_existing_literal_shell_errors_have_catalog_entries(self) -> None:
        for path in (ROOT / "planet.sh", ROOT / "scripts/lib/docker-bootstrap.zsh"):
            for message in re.findall(r'^\s*log_error "([^"\n]+)"', path.read_text(), re.M):
                message = re.sub(r"\$\{[^}]+\}|\$[0-9]+", "", message)
                if not message.strip():
                    continue  # Dynamic retry failures are classified using their runtime message.
                with self.subTest(path=path.name, message=message):
                    self.assertNotEqual(self.diagnose(message)[0], "P_UNKNOWN")

    def test_log_error_reads_cause_and_remedy_from_the_table(self) -> None:
        result = subprocess.run(
            ["zsh", "-f"],
            input=f"""
                SCRIPT_DIR={shlex.quote(str(ROOT))}
                source {shlex.quote(str(MODULE))}
                log_line() {{ echo "$3"; }}
                log_note() {{ echo "$1"; }}
                {shell_function('log_error')}
                log_error 'new failure'
            """,
            text=True,
            capture_output=True,
            timeout=10,
            check=True,
        )
        _, reason, action = catalog("zh")["P_UNKNOWN"]
        self.assertIn(f"原因 [P_UNKNOWN]: {reason}", result.stdout)
        self.assertIn(f"处理: {action}", result.stdout)

    def test_verbose_build_preserves_failure_and_literal_log_path(self) -> None:
        with tempfile.TemporaryDirectory() as folder:
            binary = Path(folder) / "docker"
            binary.write_text("#!/bin/sh\nprintf 'TLS handshake timeout\\n'\nexit 17\n")
            binary.chmod(0o755)
            for verbose in (0, 1):
                with self.subTest(verbose=verbose):
                    log = Path(folder) / "build ' quoted $(not-a-command).log"
                    result = run_shell(
                        ["run_ai_provider_build_command"],
                        f"""
                        PATH={shlex.quote(folder + ':' + os.environ['PATH'])}
                        VERBOSE={verbose}
                        AI_PROVIDER_BUILD_LOG_FILE={shlex.quote(str(log))}
                        run_command_with_spinner() {{ shift; "$@"; }}
                        """,
                        'if run_ai_provider_build_command "docker compose"; '
                        "then exit 0; else exit $?; fi",
                    )
                    self.assertEqual(result.returncode, 17, result.stderr)
                    self.assertEqual(log.read_text(), "TLS handshake timeout\n")
                    self.assertNotIn("not-a-command", result.stderr)


if __name__ == "__main__":
    unittest.main()