"""The reference client in client/dsh_sync_client.py.

Run against a stub HTTP server rather than the real app so these stay fast and
can inject failures (corrupt artifacts, 409s, dropped connections) that are
awkward to provoke through the live API.
"""
from __future__ import annotations

import hashlib
import io
import json
import sys
import tarfile
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from threading import Thread

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "client"))
from dsh_sync_client import DshSync, SyncError, _parse_value, semver_tuple  # noqa: E402


def _tar(plugin_id="demo", version="1.0.0", files=("manifest.json", "files/main.py")):
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        manifest = json.dumps({"id": plugin_id, "version": version}).encode()
        for name in files:
            blob = manifest if name.endswith("manifest.json") else b"print('hi')\n"
            info = tarfile.TarInfo(name)
            info.size = len(blob)
            tf.addfile(info, io.BytesIO(blob))
    return buf.getvalue()


class StubServer:
    """Minimal dsh-sync surface; `routes` maps (method, path) -> callable."""

    def __init__(self):
        self.routes: dict[tuple[str, str], callable] = {}
        self.requests: list[tuple[str, str, dict]] = []
        outer = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def _go(self, method):
                body = b""
                length = int(self.headers.get("Content-Length") or 0)
                if length:
                    body = self.rfile.read(length)
                parsed = json.loads(body) if body else None
                outer.requests.append((method, self.path, {"body": parsed,
                                                           "headers": dict(self.headers)}))
                fn = outer.routes.get((method, self.path.split("?")[0]))
                if fn is None:
                    self.send_response(404)
                    self.end_headers()
                    self.wfile.write(b'{"detail":"no route"}')
                    return
                status, payload, headers = fn(self.path, parsed, dict(self.headers))
                blob = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
                self.send_response(status)
                for k, v in (headers or {}).items():
                    self.send_header(k, v)
                self.send_header("Content-Length", str(len(blob)))
                self.end_headers()
                self.wfile.write(blob)

            def do_GET(self):
                self._go("GET")

            def do_POST(self):
                self._go("POST")

            def do_PUT(self):
                self._go("PUT")

        self.httpd = HTTPServer(("127.0.0.1", 0), Handler)
        self.thread = Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()

    @property
    def url(self) -> str:
        return f"http://127.0.0.1:{self.httpd.server_port}"

    def stop(self):
        self.httpd.shutdown()
        self.httpd.server_close()


@pytest.fixture()
def server():
    s = StubServer()
    yield s
    s.stop()


@pytest.fixture()
def client(server, tmp_path, monkeypatch):
    monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "state"))
    c = DshSync(tmp_path / "state")
    c.state.update(base_url=server.url, token="dsh_test", user={"id": 1, "username": "u"})
    c._save_state()
    return c


# ------------------------------------------------------------------- helpers


def test_semver_tuple_orders_numerically():
    assert semver_tuple("1.10.0") > semver_tuple("1.9.0")
    assert semver_tuple("2.0.0-beta") > semver_tuple("1.99.99")
    assert semver_tuple("1") == (1, 0, 0)


def test_parse_value_prefers_json_then_falls_back_to_text():
    assert _parse_value("0.7") == 0.7
    assert _parse_value('{"a":1}') == {"a": 1}
    assert _parse_value("dark") == "dark"


# --------------------------------------------------------------------- login


def test_login_stores_token_and_user(server, tmp_path, monkeypatch):
    monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
    server.routes[("POST", "/v1/tokens")] = lambda p, b, h: (
        200, {"token": "dsh_abc", "user": {"id": 5, "username": "alice"}}, {})
    c = DshSync(tmp_path / "s")
    out = c.login(server.url, "invite", "alice")
    assert out["token"] == "dsh_abc"
    assert c.state["token"] == "dsh_abc"
    assert DshSync(tmp_path / "s").state["token"] == "dsh_abc"  # persisted


def test_login_reports_bad_invite(server, tmp_path, monkeypatch):
    monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
    server.routes[("POST", "/v1/tokens")] = lambda p, b, h: (403, {"detail": "bad"}, {})
    with pytest.raises(SyncError, match="login failed"):
        DshSync(tmp_path / "s").login(server.url, "wrong", "alice")


def test_request_without_auth_is_refused(tmp_path, monkeypatch):
    monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
    with pytest.raises(SyncError, match="not logged in"):
        DshSync(tmp_path / "s").pull()


# ---------------------------------------------------------------------- pull


def test_pull_returns_values_and_caches_etag(server, client):
    server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
        200,
        {"settings": {"model.temperature": {"value": 0.7, "version": 3, "scope": "user"}}},
        {"ETag": '"abc"'},
    )
    assert client.pull() == {"model.temperature": 0.7}
    assert client.state["settings_etag"] == '"abc"'
    assert client.state["settings_versions"]["model.temperature"]["version"] == 3


def test_pull_sends_if_none_match_on_second_call(server, client):
    calls = []

    def handler(p, b, h):
        calls.append(h.get("If-None-Match"))
        if len(calls) == 1:
            return 200, {"settings": {"k": {"value": 1, "version": 1, "scope": "user"}}}, {"ETag": '"e1"'}
        return 304, b"", {"ETag": '"e1"'}

    server.routes[("GET", "/v1/settings")] = handler
    client.pull()
    client.pull()
    assert calls[0] is None
    assert calls[1] == '"e1"'


def test_pull_falls_back_to_cache_when_offline(server, client):
    server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
        200, {"settings": {"k": {"value": "cached", "version": 1, "scope": "user"}}}, {})
    client.pull()
    server.stop()  # server goes away
    assert client.pull() == {"k": "cached"}
    assert client.state["stale"] is True


def test_pull_without_cache_and_server_down_raises(server, client):
    server.stop()
    with pytest.raises(SyncError, match="unreachable"):
        client.pull()


# ---------------------------------------------------------------------- push


def test_push_sends_base_version_from_cache(server, client):
    server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
        200, {"settings": {"k": {"value": 1, "version": 4, "scope": "user"}}}, {})
    client.pull()
    sent = {}

    def put(path, body, headers):
        sent.update(body)
        return 200, {"key": "k", "version": 5}, {}

    server.routes[("PUT", "/v1/settings/user/k")] = put
    client.push("k", 2)
    assert sent["base_version"] == 4
    assert sent["value"] == 2


def test_push_without_cache_omits_base_version(server, client):
    sent = {}

    def put(path, body, headers):
        sent.update(body)
        return 200, {"key": "new", "version": 1}, {}

    server.routes[("PUT", "/v1/settings/user/new")] = put
    client.push("new", "v")
    assert "base_version" not in sent


def test_push_surfaces_conflict_as_readable_error(server, client):
    server.routes[("PUT", "/v1/settings/user/k")] = lambda p, b, h: (
        409, {"detail": {"message": "version conflict"}}, {})
    with pytest.raises(SyncError, match="conflict on 'k'"):
        client.push("k", 1)


# ------------------------------------------------------------------- install


def _install_routes(server, blob, *, sha=None, size=None, version="1.0.0"):
    digest = sha if sha is not None else hashlib.sha256(blob).hexdigest()
    server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (
        200, {"plugins": [{"id": "demo", "latest": {
            "version": version, "sha256": digest,
            "size_bytes": size if size is not None else len(blob),
            "min_harness_version": "0.0.0"}, "versions": []}]}, {})
    server.routes[("GET", f"/v1/plugins/demo/{version}/download")] = lambda p, b, h: (
        200, blob, {"X-Sha256": digest})


def test_install_verifies_and_unpacks(server, client, tmp_path):
    _install_routes(server, _tar())
    target = client.install("demo", target_root=tmp_path / "plugins")
    assert (target / "manifest.json").is_file()
    assert (target / "files" / "main.py").is_file()


def test_install_rejects_sha256_mismatch(server, client, tmp_path):
    _install_routes(server, _tar(), sha="0" * 64)
    with pytest.raises(SyncError, match="sha256 mismatch"):
        client.install("demo", target_root=tmp_path / "plugins")


def test_install_rejects_size_mismatch(server, client, tmp_path):
    _install_routes(server, _tar(), size=999999)
    with pytest.raises(SyncError, match="size mismatch"):
        client.install("demo", target_root=tmp_path / "plugins")


def test_install_unknown_plugin(server, client, tmp_path):
    server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (200, {"plugins": []}, {})
    with pytest.raises(SyncError, match="not found"):
        client.install("nope", target_root=tmp_path / "plugins")


def test_install_rejects_path_traversal(server, client, tmp_path):
    _install_routes(server, _tar(files=("manifest.json", "../../evil.py")))
    with pytest.raises(SyncError, match="unsafe path"):
        client.install("demo", target_root=tmp_path / "plugins")


def test_install_replaces_existing_atomically(server, client, tmp_path):
    root = tmp_path / "plugins"
    stale = root / "demo"
    stale.mkdir(parents=True)
    (stale / "old.txt").write_text("stale", encoding="utf-8")

    _install_routes(server, _tar())
    client.install("demo", target_root=root)
    assert not (root / "demo" / "old.txt").exists()
    assert (root / "demo" / "manifest.json").is_file()
    assert not list(root.glob(".*"))  # temp/bak dirs cleaned up


def test_install_honours_min_harness_version(server, client, tmp_path):
    server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (
        200, {"plugins": [{"id": "demo", "versions": [], "latest": {
            "version": "9.0.0", "sha256": "x", "size_bytes": 1,
            "min_harness_version": "5.0.0"}}]}, {})
    with pytest.raises(SyncError, match="requires harness"):
        client.install("demo", harness_version="1.0.0", target_root=tmp_path / "p")


# -------------------------------------------------------------------- status


def test_status_reports_only_safe_fields(client):
    s = client.status()
    assert set(s) == {"base_url", "user", "stale", "last_pull", "settings_etag"}
    assert "token" not in s


def test_wrapped_archive_has_its_wrapper_stripped(server, client, tmp_path):
    """`plugin-x/manifest.json` + `plugin-x/files/y` must land flat."""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        for name, blob in (("demo-1.0.0/manifest.json", b'{"id":"demo"}'),
                           ("demo-1.0.0/files/main.py", b"print(1)\n")):
            info = tarfile.TarInfo(name)
            info.size = len(blob)
            tf.addfile(info, io.BytesIO(blob))
    _install_routes(server, buf.getvalue())
    target = client.install("demo", target_root=tmp_path / "plugins")
    assert (target / "manifest.json").is_file()
    assert (target / "files" / "main.py").is_file(), "wrapper must not eat files/"


def test_push_succeeds_even_when_refresh_fails(server, client, tmp_path):
    """The write landed; a broken refresh must not look like a failed push."""
    server.routes[("PUT", "/v1/settings/user/k")] = lambda p, b, h: (
        200, {"key": "k", "version": 1}, {})
    # No GET /v1/settings route -> refresh 404s.
    out = client.push("k", "v")
    assert out["version"] == 1
    assert client.state["stale"] is True