"""Integration: macOS Container launch topology.

End-to-end against Apple's real `container` runtime. The smoke launches
a bottle with the experimental macOS Container backend and verifies the
properties that make the explicit-proxy launch acceptable:

  - the agent can exec commands after provisioning;
  - HTTP(S)_PROXY points at the sidecar's internal-network IP;
  - allowlisted HTTPS reaches the egress sidecar;
  - direct egress with proxy env removed fails from the internal-only
    agent network;
  - non-allowlisted proxy traffic is blocked.

Skipped under Gitea Actions and on hosts without Apple's `container`.
"""

from __future__ import annotations

import os
import platform
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path

from bot_bottle.backend import BottleSpec, get_bottle_backend
from bot_bottle.backend.macos_container.util import (
    dns_server as _container_dns_server,
    is_available as _container_available,
)
from bot_bottle.manifest import ManifestIndex


_AGENT_PROMPT = "You are a launch smoke-test agent. Be brief."


def _minimal_agent_dockerfile(path: Path) -> None:
    path.write_text(
        "\n".join((
            "FROM node:22-slim",
            "RUN apt-get update \\",
            "  && apt-get install -y --no-install-recommends \\",
            "    ca-certificates curl git \\",
            "  && rm -rf /var/lib/apt/lists/*",
            "USER node",
            "WORKDIR /home/node",
            "CMD [\"sleep\", \"infinity\"]",
            "",
        )),
        encoding="utf-8",
    )


def _minimal_manifest(dockerfile: Path) -> ManifestIndex:
    return ManifestIndex.from_json_obj({
        "bottles": {
            "dev": {
                "agent_provider": {
                    "template": "pi",
                    "dockerfile": str(dockerfile),
                    "settings": {
                        "provider": "example",
                        "base_url": "https://example.com/v1",
                        "models": ["smoke"],
                    },
                },
                "egress": {
                    "routes": [
                        {"host": "example.com"},
                    ],
                },
            },
        },
        "agents": {
            "demo": {
                "skills": [],
                "prompt": _AGENT_PROMPT,
                "bottle": "dev",
            },
        },
    })


def _buildkit_dns_available() -> bool:
    if platform.system() != "Darwin" or not _container_available():
        return False
    stage = Path(tempfile.mkdtemp(prefix="cb-container-buildkit-dns."))
    image = "bot-bottle-buildkit-dns-check:latest"
    try:
        dockerfile = stage / "Dockerfile"
        dockerfile.write_text(
            "FROM debian:bookworm-slim\n"
            "RUN getent hosts deb.debian.org\n",
            encoding="utf-8",
        )
        result = subprocess.run(
            [
                "container", "build",
                "--dns", _container_dns_server(),
                "-t", image,
                "-f", str(dockerfile),
                str(stage),
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        return result.returncode == 0
    finally:
        subprocess.run(
            ["container", "image", "delete", image],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        shutil.rmtree(stage, ignore_errors=True)


@unittest.skipIf(
    os.environ.get("GITEA_ACTIONS") == "true",
    "skipped under act_runner: cannot host Apple Container VMs",
)
@unittest.skipUnless(
    platform.system() == "Darwin",
    "Apple Container is macOS-only",
)
@unittest.skipUnless(
    _container_available(),
    "Apple Container not on PATH; install from "
    "https://github.com/apple/container/releases",
)
@unittest.skipUnless(
    _buildkit_dns_available(),
    "Apple Container BuildKit cannot resolve deb.debian.org on this host",
)
class TestMacosContainerLaunch(unittest.TestCase):
    """Launch once and reuse the bottle across probes."""

    @classmethod
    def setUpClass(cls) -> None:
        cls.stage = Path(tempfile.mkdtemp(prefix="cb-macos-container-launch."))
        cls._launch = None
        cls.bottle = None
        dockerfile = cls.stage / "Dockerfile.agent-smoke"
        _minimal_agent_dockerfile(dockerfile)
        os.environ["BOT_BOTTLE_BACKEND"] = "macos-container"
        try:
            backend = get_bottle_backend()
            spec = BottleSpec(
                manifest=_minimal_manifest(dockerfile),
                agent_name="demo",
                copy_cwd=False,
                user_cwd=str(cls.stage),
            )
            cls.plan = backend.prepare(spec, stage_dir=cls.stage)
            cls._launch = backend.launch(cls.plan)
            cls.bottle = cls._launch.__enter__()
        except BaseException:
            if cls._launch is not None:
                cls._launch.__exit__(None, None, None)
            shutil.rmtree(cls.stage, ignore_errors=True)
            os.environ.pop("BOT_BOTTLE_BACKEND", None)
            raise

    @classmethod
    def tearDownClass(cls) -> None:
        try:
            if cls._launch is not None:
                cls._launch.__exit__(None, None, None)
        finally:
            shutil.rmtree(cls.stage, ignore_errors=True)
            os.environ.pop("BOT_BOTTLE_BACKEND", None)

    def test_smoke_exec_echo(self):
        r = self.bottle.exec(  # type: ignore[union-attr]
            "echo hello-from-macos-container"
        )
        self.assertEqual(0, r.returncode, msg=r.stderr)
        self.assertIn("hello-from-macos-container", r.stdout)

    def test_proxy_env_points_at_sidecar_internal_ip(self):
        r = self.bottle.exec(  # type: ignore[union-attr]
            "printf '%s\n' \"$HTTPS_PROXY\" \"$HTTP_PROXY\" "
            "\"$NO_PROXY\" \"$NODE_EXTRA_CA_CERTS\""
        )
        self.assertEqual(0, r.returncode, msg=r.stderr)
        values = [line.strip() for line in r.stdout.splitlines()]
        self.assertEqual(4, len(values), values)
        self.assertEqual(values[0], values[1], values)
        self.assertRegex(values[0], r"^http://[0-9.]+:9099$")
        self.assertNotIn("127.0.0.1", values[0])
        sidecar_host = values[0].removeprefix("http://").removesuffix(":9099")
        self.assertIn(sidecar_host, values[2])
        self.assertEqual(
            "/usr/local/share/ca-certificates/bot-bottle-mitm-ca.crt",
            values[3],
        )

    def test_allowlisted_https_reaches_egress_proxy(self):
        r = self.bottle.exec(  # type: ignore[union-attr]
            "curl -fsS --max-time 20 https://example.com >/dev/null && echo OK"
        )
        self.assertEqual(0, r.returncode, msg=r.stderr + r.stdout)
        self.assertIn("OK", r.stdout)

    def test_direct_egress_bypass_without_proxy_fails(self):
        r = self.bottle.exec(  # type: ignore[union-attr]
            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
            "curl -s --show-error --max-time 5 https://example.com 2>&1 || true"
        )
        self.assertTrue(
            "refused" in r.stdout.lower()
            or "timed out" in r.stdout.lower()
            or "unreachable" in r.stdout.lower()
            or "failed" in r.stdout.lower()
            or "could not resolve" in r.stdout.lower()
            or "connection reset" in r.stdout.lower(),
            f"expected direct egress to fail; got: {r.stdout!r}",
        )

    def test_non_allowlisted_host_fails_through_proxy(self):
        r = self.bottle.exec(  # type: ignore[union-attr]
            "curl -s --show-error --max-time 10 https://iana.org 2>&1 || true"
        )
        self.assertTrue(
            "403" in r.stdout
            or "502" in r.stdout
            or "blocked" in r.stdout.lower()
            or "not allowed" in r.stdout.lower()
            or "not in the bottle's egress.routes allowlist" in r.stdout.lower()
            or "forbidden" in r.stdout.lower()
            or "failed" in r.stdout.lower(),
            f"expected non-allowlisted proxy request to fail; got: {r.stdout!r}",
        )


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