"""Exercise stalled Docker substitutes without contacting a daemon."""

import os
from pathlib import Path
import signal
import subprocess
import sys
import tempfile
import unittest


RUNNER = Path(__file__).resolve().parents[1] / "docker_command.py"


class DockerCommandTests(unittest.TestCase):
    def run_command(self, source: str) -> tuple[subprocess.CompletedProcess[str], str]:
        with tempfile.TemporaryDirectory() as folder:
            log = Path(folder) / "quoted ' $(literal).log"
            status = Path(folder) / "status"
            result = subprocess.run(
                [sys.executable, str(RUNNER), "--label", "Compose up postgres redis",
                 "--timeout", "0.6", "--heartbeat", "0.15", "--log", str(log),
                 "--status-file", str(status), "--",
                 sys.executable, "-c", source],
                capture_output=True, text=True, timeout=5,
            )
            self.assertEqual(log.stat().st_mode & 0o777, 0o600)
            self.last_status = status.read_text()
            return result, log.read_text()

    def test_silent_hang_has_heartbeat_deadline_and_no_invented_cause(self) -> None:
        result, log = self.run_command("import time; time.sleep(30)")
        self.assertEqual(result.returncode, 124, result.stderr)
        self.assertIn("Compose up postgres redis", self.last_status)
        self.assertIn("超时", self.last_status)
        self.assertEqual(result.stdout + result.stderr, "")
        self.assertIn("PLANET_DOCKER_COMMAND_TIMEOUT", log)
        self.assertNotIn("代理", log)

    def test_partial_and_carriage_return_output_survives_timeout(self) -> None:
        result, log = self.run_command(
            "import time; print('postgres Pulling\\rredis Downloading', end='', flush=True); "
            "time.sleep(30)"
        )
        self.assertEqual(result.returncode, 124)
        self.assertIn("postgres Pulling", log)
        self.assertIn("redis Downloading", self.last_status)

    def test_failure_is_streamed_and_original_exit_code_preserved(self) -> None:
        result, log = self.run_command(
            "import sys; print('TLS handshake timeout', file=sys.stderr); sys.exit(17)"
        )
        self.assertEqual(result.returncode, 17)
        self.assertIn("TLS handshake timeout", result.stdout)
        self.assertEqual(log, "TLS handshake timeout\n")
        self.assertEqual(self.last_status, "TLS handshake timeout")

    def test_success_and_proxy_credentials(self) -> None:
        result, log = self.run_command("print('https://user:secret@example.test/v2/ ready')")
        self.assertEqual(result.returncode, 0)
        self.assertEqual(self.last_status, "")
        self.assertEqual(result.stderr, "")
        self.assertIn("https://***@example.test/v2/ ready", log)
        self.assertNotIn("secret", result.stdout + result.stderr + log)

    def test_long_compose_network_error_keeps_the_port_in_secondary_line(self) -> None:
        result, _ = self.run_command(
            "import sys; print('Error response from daemon: failed to set up container networking: '"
            " + 'x' * 160 + ': Bind for 127.0.0.1:5432 failed: port is already allocated'); sys.exit(1)"
        )
        self.assertEqual(result.returncode, 1)
        self.assertEqual(self.last_status,
                         "Bind for 127.0.0.1:5432 failed: port is already allocated")

    def test_timeout_kills_descendant_even_after_cli_parent_exits(self) -> None:
        result, _ = self.run_command(
            "import subprocess, sys; "
            "p = subprocess.Popen([sys.executable, '-c', "
            "'import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
            "time.sleep(30)']); print(p.pid, flush=True)"
        )
        self.assertEqual(result.returncode, 124)
        child = int(result.stdout.strip())
        state = Path(f"/proc/{child}/stat")
        try:
            self.assertTrue(not state.exists() or state.read_text().split()[2] == "Z")
        finally:
            if state.exists() and state.read_text().split()[2] != "Z":
                os.kill(child, signal.SIGKILL)


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