import hashlib
import http.client
import json
import os
import select
import socket
import stat
import subprocess
import sys
import threading
import time
from pathlib import Path

import pytest

SERVER = Path(__file__).resolve().parents[1] / "server.py"
TRAILER_SIZE = 512
RECENT_MAX = 8


def make_image(path: Path, device: str, version: str, payload: bytes) -> bytes:
    """Builds a file with the same trailer layout the firmware verifies."""
    trailer = bytearray(b"\xff" * TRAILER_SIZE)
    trailer[0:4] = b"NINL"
    trailer[4] = 1
    trailer[5:21] = device.encode().ljust(16, b"\x00")
    trailer[21:37] = version.encode().ljust(16, b"\x00")
    trailer[37:69] = hashlib.sha256(payload).digest()
    data = payload + bytes(trailer)
    path.write_bytes(data)
    return data


def free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


class Station:
    def __init__(self, tmp_path, *extra):
        self.image_dir = tmp_path / "images"
        self.image_dir.mkdir()
        self.state = tmp_path / "status.json"
        self.port = free_port()
        self.proc = subprocess.Popen(
            [sys.executable, str(SERVER), "--image-dir", str(self.image_dir),
             "--bind", "127.0.0.1", "--port", str(self.port),
             "--state", str(self.state), *extra],
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
        )
        deadline = time.time() + 10
        while time.time() < deadline:
            try:
                socket.create_connection(("127.0.0.1", self.port), 0.2).close()
                return
            except OSError:
                if self.proc.poll() is not None:
                    raise RuntimeError("server died: " + (self.proc.stdout.read() or ""))
                time.sleep(0.05)
        raise RuntimeError("server did not come up")

    def conn(self):
        return http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)

    def read_state(self, predicate=None):
        """Reads status.json, optionally waiting for it to satisfy `predicate`.

        The server only writes the entry for a request after that request's response has
        actually finished (or failed) on the wire - by design, so it never claims success
        before the last byte is confirmed sent (see server.py). A client that has just
        finished reading its own response can therefore still see a status.json that is a
        few milliseconds stale. Rather than have the server promise something before it is
        true, we wait here for the state to catch up.
        """
        last = None
        for _ in range(100):
            if self.state.exists():
                try:
                    last = json.loads(self.state.read_text())
                    if predicate is None or predicate(last):
                        return last
                except json.JSONDecodeError:
                    pass
            time.sleep(0.05)
        if last is None:
            raise AssertionError("no state file written")
        raise AssertionError(f"state never satisfied predicate, last seen: {last!r}")

    def stop(self):
        self.proc.terminate()
        self.proc.wait(timeout=5)

    def wait_for_log_line(self, contains, timeout=3.0):
        """Reads the server's merged stdout/stderr (see __init__) until a line containing
        `contains` shows up, skipping unrelated lines (e.g. the startup banner) - used to
        check that a request actually produced a log line (F3: every request must,
        including rejected ones and errors the base handler catches before routing).

        Reads the raw fd directly with os.read() instead of self.proc.stdout.readline():
        readline() goes through TextIOWrapper's internal BufferedReader, whose underlying
        read() can slurp everything currently sitting in the pipe (banner *and* the log
        line we're waiting for) into its own buffer on the very first call, while only
        handing back the first line. select() then reports the fd as not-readable on the
        next iteration - the data isn't gone, it is just buffered somewhere select() can't
        see - and a fixed-size test timeout would flake on exactly the line it's checking
        for.
        """
        fd = self.proc.stdout.fileno()
        buf = getattr(self, "_log_buf", "")
        deadline = time.time() + timeout
        while True:
            while "\n" in buf:
                line, buf = buf.split("\n", 1)
                self._log_buf = buf
                if contains in line:
                    return line + "\n"
            remaining = deadline - time.time()
            if remaining <= 0:
                self._log_buf = buf
                return None
            r, _, _ = select.select([fd], [], [], remaining)
            if not r:
                self._log_buf = buf
                return None
            chunk = os.read(fd, 65536)
            if not chunk:
                self._log_buf = buf
                return None
            buf += chunk.decode("utf-8", "replace")


@pytest.fixture
def station(tmp_path):
    s = Station(tmp_path)
    yield s
    s.stop()


def test_serves_a_known_image(station):
    data = make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.2.3", b"payload" * 500)
    c = station.conn()
    c.request("GET", "/ninlight_evk_ota.img")
    r = c.getresponse()
    body = r.read()
    assert r.status == 200
    assert int(r.getheader("Content-Length")) == len(data)
    assert body == data


def test_missing_image_is_404(station):
    c = station.conn()
    c.request("GET", "/ninlight_g95_cw_ota.img")
    assert c.getresponse().status == 404


def test_unknown_target_is_404(station):
    make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"x" * 100)
    for path in ("/ninlight_nosuch_ota.img", "/ninlight_evk_ota.img.bak", "/", "/images/"):
        c = station.conn()
        c.request("GET", path)
        assert c.getresponse().status == 404, path


def test_path_traversal_is_refused(station):
    secret = station.image_dir.parent / "secret.txt"
    secret.write_text("nicht ausliefern")
    c = station.conn()
    c.request("GET", "/../secret.txt")
    assert c.getresponse().status == 404


def test_other_methods_are_405(station):
    c = station.conn()
    c.request("POST", "/ninlight_evk_ota.img", body=b"")
    assert c.getresponse().status == 405


def test_rejected_method_produces_a_log_line(station):
    """F3: do_POST (and PUT/DELETE/HEAD) used to call reject() without logging anything -
    the station is the only diagnostic tool during a rescue, and the spec requires a line
    per request, rejections included."""
    c = station.conn()
    c.request("POST", "/ninlight_evk_ota.img", body=b"")
    assert c.getresponse().status == 405
    line = station.wait_for_log_line(" 405 ")
    assert line is not None, "kein Log-Eintrag fuer die abgelehnte POST-Anfrage"
    assert "/ninlight_evk_ota.img" in line


def test_log_error_is_not_silenced_by_log_message(capsys):
    """F3: BaseHTTPRequestHandler.log_error() calls log_message() by default (used e.g. for
    a malformed request line or a client timeout, cases that never reach do_GET/do_POST).
    server.py silences log_message() in favour of its own per-request log() calls - which
    used to silence log_error() right along with it, since it wasn't overridden separately.
    This exercises Handler.log_error() directly rather than over a raw socket: this
    sandbox's network layer intercepts/rewrites malformed non-HTTP payloads before they
    reach the server (observed via a spoofed 'Server: aiohttp' response), which makes a
    genuine malformed-request-line test unreliable here."""
    import importlib.util

    spec = importlib.util.spec_from_file_location("ninlight_station_server", SERVER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    class _Stub:
        client_address = ("203.0.113.5", 54321)

    module.Handler.log_error(_Stub(), "code %d, message %s", 400, "Bad request version")
    out = capsys.readouterr().out
    assert "ERROR" in out
    assert "203.0.113.5" in out
    assert "code 400, message Bad request version" in out


def test_range_request_returns_206_and_the_right_slice(station):
    data = make_image(station.image_dir / "ninlight_e27_rgb_ota.img", "e27_rgb", "2.0.0", b"abc" * 2000)
    c = station.conn()
    c.request("GET", "/ninlight_e27_rgb_ota.img", headers={"Range": "bytes=1000-"})
    r = c.getresponse()
    body = r.read()
    assert r.status == 206
    assert r.getheader("Content-Range") == f"bytes 1000-{len(data)-1}/{len(data)}"
    assert body == data[1000:]


def test_state_file_records_a_delivery(station):
    make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "0.3.4", b"z" * 1000)
    c = station.conn()
    c.request("GET", "/ninlight_evk_ota.img")
    c.getresponse().read()
    state = station.read_state(lambda s: s["served"] >= 1)
    assert state["served"] == 1
    entry = state["recent"][-1]
    assert entry["target"] == "evk"
    assert entry["version"] == "0.3.4"
    assert entry["status"] == "ok"
    assert entry["ip"] == "127.0.0.1"
    assert entry["bytes"] == 1000 + TRAILER_SIZE


def test_state_file_records_a_missing_image(station):
    c = station.conn()
    c.request("GET", "/ninlight_g95_cw_ota.img")
    c.getresponse().read()
    state = station.read_state(lambda s: len(s["recent"]) >= 1)
    assert state["served"] == 0
    assert state["recent"][-1]["status"] == "notfound"
    assert state["recent"][-1]["target"] == "g95_cw"


def test_state_file_is_world_readable_even_under_a_restrictive_umask(tmp_path):
    """status.json is written by a root service but read by a normal-user CLI.

    tempfile.mkstemp() creates its file 0600 and os.replace() preserves whatever mode
    the temp file had, so the server must explicitly widen the mode before the rename -
    and must do so with chmod() rather than relying on the mode passed to mkstemp/open,
    because that one is still narrowed by the process umask. A restrictive umask (as
    root services on the Pi may well run under) must not bring the 0600 back.
    """
    old_umask = os.umask(0o077)
    try:
        s = Station(tmp_path)
    finally:
        os.umask(old_umask)
    try:
        make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"w" * 100)

        def world_readable() -> bool:
            mode = stat.S_IMODE(s.state.stat().st_mode)
            return bool(mode & stat.S_IROTH)

        # Written once already at startup (Station.__init__ -> write_state()).
        s.read_state(lambda _: world_readable())
        assert world_readable(), "status.json is not world-readable after the initial write"

        # A second write (a real delivery) must not lose the mode either - e.g. if the
        # widening were only applied to the file's very first creation instead of to
        # every temp file, mode 0600 would silently come back on every later write.
        c = s.conn()
        c.request("GET", "/ninlight_evk_ota.img")
        c.getresponse().read()
        s.read_state(lambda st: st["served"] >= 1)
        assert world_readable(), "status.json lost its mode on a later overwrite"
    finally:
        s.stop()


def test_recent_is_capped_at_eight(station):
    make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"q" * 10)
    for _ in range(12):
        c = station.conn()
        c.request("GET", "/ninlight_evk_ota.img")
        c.getresponse().read()
    state = station.read_state(lambda s: s["served"] >= 12)
    assert state["served"] == 12
    assert len(state["recent"]) == RECENT_MAX


def test_aborted_transfer_is_recorded(tmp_path):
    s = Station(tmp_path, "--stall-after", "256")
    try:
        make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 100000)
        sock = socket.create_connection(("127.0.0.1", s.port), timeout=5)
        sock.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\n\r\n")
        sock.recv(4096)
        time.sleep(0.3)
        sock.close()
        deadline = time.time() + 10
        while time.time() < deadline:
            try:
                entry = json.loads(s.state.read_text())["recent"][-1]
                if entry["status"] == "aborted":
                    assert entry["bytes"] < 100000
                    return
            except (OSError, json.JSONDecodeError, IndexError, KeyError):
                pass
            time.sleep(0.1)
        raise AssertionError("abort was not recorded")
    finally:
        s.stop()


def test_stall_after_stops_sending_but_keeps_the_connection(tmp_path):
    s = Station(tmp_path, "--stall-after", "512")
    try:
        make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)
        sock = socket.create_connection(("127.0.0.1", s.port), timeout=5)
        sock.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\n\r\n")
        got = b""
        sock.settimeout(2)
        try:
            while len(got) < 200000:
                chunk = sock.recv(65536)
                if not chunk:
                    break
                got += chunk
        except socket.timeout:
            pass
        # far less than the full image
        assert len(got) < 20000
        # and the connection is genuinely still open: a server that (wrongly) closed it
        # after the stall boundary would make this recv() return b"" (EOF) instead of
        # timing out. Just counting received bytes above would not catch that - a server
        # closing the connection right after --stall-after also sends far less than
        # 20000 bytes.
        sock.settimeout(1)
        with pytest.raises(socket.timeout):
            sock.recv(1)
        sock.close()
    finally:
        s.stop()


def test_stall_after_budget_survives_a_resumed_request(tmp_path):
    """--stall-after must stop a client for good, not just for one request.

    On real hardware, a lamp whose download stalls hits its own receive timeout, gives up
    on the connection, and opens a new one with a Range header picking up right where it
    left off - exactly what test_stall_after_stops_sending_but_keeps_the_connection above
    exercises for a single request. A per-request byte counter would hand that resumed
    request a brand-new --stall-after allowance, so the "stall" would only ever delay
    delivery instead of actually stopping it - which is what was observed on the EVK: the
    server stalled at 65536 bytes each time, but the lamp reconnected five times and walked
    away with the complete image. This test reproduces that reconnect and checks that the
    second, resumed request gets no further bytes at all.
    """
    s = Station(tmp_path, "--stall-after", "1000")
    try:
        make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)

        # First request: burns through the whole --stall-after budget, then the server
        # holds the connection open and sends nothing further.
        sock = socket.create_connection(("127.0.0.1", s.port), timeout=5)
        sock.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\n\r\n")
        got = b""
        sock.settimeout(2)
        try:
            while len(got) < 200000:
                chunk = sock.recv(65536)
                if not chunk:
                    break
                got += chunk
        except socket.timeout:
            pass
        assert len(got) < 20000  # far less than the full image, same bound as above
        sock.close()  # the lamp's watchdog gives up and reconnects, like on real hardware

        # Second request, resuming with Range exactly like a real lamp would. This
        # client's cumulative budget is already spent, so it must get zero more bytes of
        # body - not a fresh --stall-after allowance just because it's a new connection.
        sock2 = socket.create_connection(("127.0.0.1", s.port), timeout=5)
        sock2.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\nRange: bytes=1000-\r\n\r\n")
        sock2.settimeout(2)
        header = b""
        while b"\r\n\r\n" not in header:
            header += sock2.recv(4096)
        assert b" 206 " in header.split(b"\r\n", 1)[0]
        body_after_header = header.split(b"\r\n\r\n", 1)[1]
        assert body_after_header == b"", "resumed request must not receive any body bytes"
        try:
            more = sock2.recv(1)
        except socket.timeout:
            more = None
        assert more in (None, b""), "server sent data after the client's stall budget was spent"
        sock2.close()
    finally:
        s.stop()


def test_concurrent_downloads_are_all_recorded(station):
    """Several lamps hitting the station at once must not lose or corrupt status.json.

    Station.served is a shared counter and Station.recent a shared list, mutated from one
    handler thread per connection under ThreadingHTTPServer. Without a lock around the
    read-modify-write of `served` and the recent-list mutation + serialize-to-disk step,
    concurrent requests can lose increments (two threads read the same old value) or have
    one thread's status.json write clobber another's.
    """
    make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"c" * 5000)
    expected_len = 5000 + TRAILER_SIZE
    n_workers = 20
    errors = []
    results_lock = threading.Lock()

    def worker():
        try:
            c = station.conn()
            c.request("GET", "/ninlight_evk_ota.img")
            r = c.getresponse()
            body = r.read()
            if r.status != 200 or len(body) != expected_len:
                with results_lock:
                    errors.append((r.status, len(body)))
        except Exception as e:  # pragma: no cover - surfaced via the errors list below
            with results_lock:
                errors.append(e)

    threads = [threading.Thread(target=worker) for _ in range(n_workers)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=10)

    assert not errors, f"some downloads failed or returned the wrong body: {errors}"

    state = station.read_state(lambda s: s["served"] >= n_workers)
    assert state["served"] == n_workers
    assert len(state["recent"]) == min(n_workers, RECENT_MAX)
    for entry in state["recent"]:
        assert entry["status"] == "ok"
        assert entry["target"] == "evk"
        assert entry["bytes"] == expected_len


# --- Migrationsroute (Task 3) -------------------------------------------------------
#
# Eigener Testklient statt der Station-Klasse oben: die migration_server/server-Fixtures
# brauchen get()/status()/set_control()/read_control()/delete_control()/write_raw_control(),
# waehrend die Station-Klasse und die "station"-Fixture oben unveraendert fuer die
# bestehenden Rettungsmodus-Tests weiterlaufen sollen.


class _Response:
    def __init__(self, status, headers, body):
        self.status = status
        self.headers = headers
        self.body = body


class MigrationTestServer:
    """Startet server.py mit --mode/--control und bietet die vom Brief geforderten
    Hilfsmethoden.

    status()/read_control() folgen demselben Muster wie Station.read_state() oben
    (Zeilen 68-90): sie pollen bis zu einigen Sekunden auf ein optionales Praedikat, statt
    sich auf eine feste Pause nach get() zu verlassen. Grund ist derselbe wie dort -
    server.py schreibt status.json/control.json fuer eine Anfrage erst, nachdem der
    Antwortkoerper vollstaendig gesendet ist, aber http.client's read() kann zurueckkehren,
    sobald Content-Length Bytes angekommen sind, was diesem Schreiben knapp zuvorkommen
    kann. Jeder Aufrufer kennt sein eigenes Praedikat (das ohnehin schon als Assertion
    dasteht), daher kein pauschales Warten hier."""

    def __init__(self, tmp_path, mode, *extra_args):
        self.image_dir = tmp_path / "images"
        self.image_dir.mkdir()
        self.state_path = tmp_path / "status.json"
        # Eigenes Unterverzeichnis fuer die Steuerdatei, getrennt von status.json: ein Test
        # fuer I3 (Fixrunde 1) macht genau dieses Verzeichnis schreibgeschuetzt, um einen
        # fehlschlagenden write_control() zu erzwingen, und darf dabei nicht auch das
        # Schreiben von status.json lahmlegen, ueber das derselbe Test den Fehler beobachtet.
        control_dir = tmp_path / "control"
        control_dir.mkdir()
        self.control_path = control_dir / "control.json"
        # Der Umschalter des Bildsatzes. Die Datei existiert zunaechst NICHT - das ist
        # der Auslieferungszustand und muss "ohne-wlan" (migrate-nocreds/) bedeuten.
        self.imageset_path = tmp_path / "imageset"
        self.port = free_port()
        self.proc = subprocess.Popen(
            [sys.executable, str(SERVER), "--image-dir", str(self.image_dir),
             "--bind", "127.0.0.1", "--port", str(self.port),
             "--state", str(self.state_path), "--mode", mode,
             "--control", str(self.control_path),
             "--imageset-file", str(self.imageset_path), *extra_args],
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
        )
        deadline = time.time() + 10
        while time.time() < deadline:
            try:
                socket.create_connection(("127.0.0.1", self.port), 0.2).close()
                return
            except OSError:
                if self.proc.poll() is not None:
                    raise RuntimeError("server died: " + (self.proc.stdout.read() or ""))
                time.sleep(0.05)
        raise RuntimeError("server did not come up")

    def get(self, path, headers=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        try:
            conn.request("GET", path, headers=headers or {})
            r = conn.getresponse()
            body = r.read()
            return _Response(r.status, dict(r.getheaders()), body)
        finally:
            conn.close()

    def _poll_json(self, path, predicate=None, timeout=5.0):
        """Wie Station.read_state() oben: liest `path` wiederholt (bis zu `timeout`
        Sekunden), bis er sich parsen laesst und - falls angegeben - `predicate` erfuellt.
        Ohne predicate reicht ein einziges erfolgreiches Parsen, genau wie dort."""
        last = None
        last_err = None
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                last = json.loads(path.read_text())
                if predicate is None or predicate(last):
                    return last
            except (OSError, json.JSONDecodeError) as e:
                last_err = e
            time.sleep(0.02)
        if last is None:
            raise AssertionError(f"could not read {path}: {last_err}")
        raise AssertionError(f"{path} never satisfied predicate, last seen: {last!r}")

    def status(self, predicate=None):
        return self._poll_json(self.state_path, predicate)

    def read_control(self, predicate=None):
        return self._poll_json(self.control_path, predicate)

    def set_control(self, phase, armed_target=None):
        tmp = self.control_path.with_name(self.control_path.name + ".tmp")
        tmp.write_text(json.dumps({"phase": phase, "armed_target": armed_target}))
        os.chmod(tmp, 0o644)
        os.replace(tmp, self.control_path)

    def delete_control(self):
        try:
            self.control_path.unlink()
        except FileNotFoundError:
            pass

    def write_raw_control(self, text):
        self.control_path.write_text(text)

    def set_image_set(self, value):
        """Stellt den Umschalter so um, wie es das CLI tut: eine Datei mit dem Namen des
        Bildsatzes. `None` loescht sie wieder (= Auslieferungszustand)."""
        if value is None:
            try:
                self.imageset_path.unlink()
            except FileNotFoundError:
                pass
            return
        self.imageset_path.write_text(value)

    def stop(self):
        self.proc.terminate()
        self.proc.wait(timeout=5)


@pytest.fixture
def server(tmp_path):
    srv = MigrationTestServer(tmp_path, "recovery")
    yield srv, srv.image_dir
    srv.stop()


@pytest.fixture
def migration_server(tmp_path):
    srv = MigrationTestServer(tmp_path, "migration")
    yield srv, srv.image_dir
    srv.stop()


def test_migration_auto_serves_known_product(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG" * 1000)
    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img")
    assert r.status == 200
    assert r.body == b"IMG" * 1000
    st = srv.status(lambda s: s.get("last_seen") is not None)
    assert st["last_seen"]["target"] == "g95_cw"
    assert st["last_seen"]["variant"] == "dev"


def test_migration_dev_only_in_version_serves_the_dev_image(migration_server):
    """Der Hardwarebefund vom 2026-09-15: Produkt-ID OHNE "-dev", Version MIT. Frueher
    ging hier das produktionssignierte Image raus (200, und die Lampe lehnte es ab) -
    jetzt muss das dev-signierte gehen, und status.json muss die Quelle mitfuehren."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"DEV" * 1000)
    (imgdir / "migrate-nocreds" / "g95_cw_prod.img").write_bytes(b"PROD" * 1000)
    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img")
    assert r.status == 200
    assert r.body == b"DEV" * 1000
    st = srv.status(lambda s: s.get("last_seen") is not None)
    assert st["last_seen"]["target"] == "g95_cw"
    assert st["last_seen"]["variant"] == "dev"
    assert st["last_seen"]["variant_source"] == "version"
    assert st["recent"][-1]["variant"] == "dev"
    assert st["recent"][-1]["variant_source"] == "version"


def test_migration_production_lamp_still_gets_the_prod_image(migration_server):
    """Die erfolgreich migrierte GU10: nirgends ein "-dev", also prod - und die Quelle
    "none", damit am Panel/in status.json steht, worauf die Einstufung beruht."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "gu10_rgb_dev.img").write_bytes(b"DEV" * 1000)
    (imgdir / "migrate-nocreds" / "gu10_rgb_prod.img").write_bytes(b"PROD" * 1000)
    r = srv.get("/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img")
    assert r.status == 200
    assert r.body == b"PROD" * 1000
    st = srv.status(lambda s: s.get("last_seen") is not None)
    assert st["last_seen"]["variant"] == "prod"
    assert st["last_seen"]["variant_source"] == "none"


def test_migration_listen_records_but_serves_nothing(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.set_control(phase="listen")
    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
    assert r.status == 404
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["last_seen"]["target"] == "g95_cw"      # erkannt
    assert st["served"] == 0                          # aber nichts ausgeliefert
    assert st["recent"][-1]["status"] == "listen"


def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.set_control(phase="armed", armed_target="g95_cw")
    p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
    assert srv.get(p).status == 200
    assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen"
    assert srv.get(p).status == 404


def test_armed_for_other_target_does_not_serve(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.set_control(phase="armed", armed_target="gu10_rgb")
    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
    assert r.status == 404
    assert srv.read_control()["phase"] == "armed"   # Freigabe bleibt stehen


# --- Eine Freigabe wird NUR durch eine erfolgreiche Auslieferung verbraucht ----------
#
# Die Zusage (Spec 3.3): claim_delivery() nimmt die Freigabe schon unter der Sperre
# weg, BEVOR ueberhaupt gesendet wird (das ist der Punkt von C1 aus Fixrunde 1). Geht
# danach nichts raus, muss restore_armed() sie zurueckgeben - sonst steht der Bediener
# vor einer Lampe, die beim naechsten Boot nichts mehr bekommt, und muss raten, warum.
#
# Es gibt drei solche Stellen in _do_get_migration(): fehlender Bildsatz, fehlendes
# Einzelimage und abgebrochener Transfer. Nur die erste (die juengste) war getestet -
# das Abschlussreview hat die beiden anderen mit je einem "if False:" ausgehebelt, ohne
# dass ein einziger der 225 Tests rot wurde. Die beiden folgenden Tests schliessen das.


def test_a_missing_image_does_not_burn_the_release(migration_server):
    """Der Bildsatz ist da, nur DIESES Modell fehlt darin (Status "notfound"). Die
    Freigabe darf daran nicht verfallen: sie war fuer genau dieses Ziel gedacht, und ein
    fehlendes Image ist ein Deploy-Versaeumnis, kein Auslieferungsvorgang."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    # Der Satz existiert (ein anderes Modell liegt darin), e14_cw fehlt darin.
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.set_control(phase="armed", armed_target="e14_cw")

    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img")

    assert r.status == 404
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["recent"][-1]["status"] == "notfound"
    assert st["served"] == 0
    control = srv.read_control()
    assert control["phase"] == "armed", \
        "die Freigabe wurde verbraucht, obwohl es gar kein Image auszuliefern gab"
    assert control["armed_target"] == "e14_cw", "das freigegebene Ziel ging verloren"


def test_an_aborted_transfer_does_not_burn_the_release(tmp_path):
    """Bricht die Uebertragung ab (Lampe verliert Strom oder Funk mitten im Download),
    muss die Freigabe stehen bleiben, damit die Lampe es beim naechsten Boot erneut
    versuchen kann - sonst muss jemand vor Ort erneut freigeben, und genau diese
    Nachsicht ist der Grund, warum restore_armed() ueberhaupt existiert.

    Nachgestellt wie test_aborted_transfer_is_recorded oben: --stall-after laesst den
    Server nach n Byte verstummen, der Test schliesst die Verbindung danach, und
    _serve_file() meldet "aborted"."""
    srv = MigrationTestServer(tmp_path, "migration", "--stall-after", "256")
    try:
        imgdir = srv.image_dir
        (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
        (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"y" * 100000)
        srv.set_control(phase="armed", armed_target="g95_cw")

        sock = socket.create_connection(("127.0.0.1", srv.port), timeout=5)
        try:
            sock.sendall(("GET " + G95_DEV + " HTTP/1.1\r\nHost: x\r\n\r\n").encode())
            sock.recv(4096)
            time.sleep(0.3)
        finally:
            sock.close()

        st = srv.status(lambda s: s["recent"] and s["recent"][-1]["status"] == "aborted")
        assert st["recent"][-1]["bytes"] < 100000
        assert st["served"] == 0
        control = srv.read_control(lambda c: True)
        assert control["phase"] == "armed", \
            "die Freigabe wurde von einem ABGEBROCHENEN Transfer verbraucht"
        assert control["armed_target"] == "g95_cw", "das freigegebene Ziel ging verloren"
    finally:
        srv.stop()


def test_unknown_product_id_is_404_and_logged(migration_server):
    srv, _ = migration_server
    r = srv.get("/qcx_firmware_-acme.bulb_V1_ota_light.img")
    assert r.status == 404
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["recent"][-1]["status"] == "unknown"


def test_missing_migrate_image_is_404(migration_server):
    srv, imgdir = migration_server
    # Der Bildsatz ist da (er enthaelt ein anderes Modell), nur dieses eine Modell fehlt
    # darin - ein anderer Zustand als ein fehlender Bildsatz (siehe
    # test_a_selected_but_missing_image_set_delivers_nothing).
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img")
    assert r.status == 404
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["recent"][-1]["status"] == "notfound"


def test_recovery_path_is_404_in_migration_mode(migration_server):
    srv, imgdir = migration_server
    (imgdir / "ninlight_evk_ota.img").write_bytes(b"x" * 600)
    assert srv.get("/ninlight_evk_ota.img").status == 404


def test_migration_path_is_404_in_recovery_mode(server):
    srv, _ = server
    assert srv.get(
        "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 404


def test_missing_control_file_means_auto(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.delete_control()
    assert srv.get(
        "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200


def test_corrupt_control_file_means_auto(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.write_raw_control("{das ist kein json")
    assert srv.get(
        "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200


def test_range_resume_on_migration_image(migration_server):
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(bytes(range(256)) * 40)
    p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
    r = srv.get(p, headers={"Range": "bytes=1000-"})
    assert r.status == 206
    assert r.headers["Content-Range"] == "bytes 1000-10239/10240"
    assert r.body == (bytes(range(256)) * 40)[1000:]


def test_version_comes_from_manifest(migration_server):
    """Faellt auf das manifestweite Feld zurueck, wenn der Eintrag selbst keine eigene
    Version traegt - Bestandskompatibilitaet mit einem Manifest aus der Zeit vor
    Fixrunde 2 (N2)."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    (imgdir / "migrate-nocreds" / "manifest.json").write_text(json.dumps({
        "ninlight_version": "0.3.9",
        "images": [{"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img"}]}))
    srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["recent"][-1]["version"] == "0.3.9"


def test_version_prefers_the_delivered_entry_over_the_manifest_wide_field(migration_server):
    """Fixrunde 2, N2: nach einem Merge in deploy.sh (Voll-Deploy mit nur einem der
    beiden Signierschluessel) koennen Eintraege im selben Manifest unterschiedliche
    Firmware-Versionen tragen - z.B. ein neu signiertes g95_cw_dev.img auf 1.2.0, aber
    ein unveraendert liegen gebliebenes aelteres g95_cw_prod.img auf 1.0.0. Der Server
    darf dann nicht das manifestweite Feld fuer beide protokollieren, sondern muss die
    Version DES AUSGELIEFERTEN Eintrags nehmen."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG-DEV")
    (imgdir / "migrate-nocreds" / "g95_cw_prod.img").write_bytes(b"IMG-PROD")
    (imgdir / "migrate-nocreds" / "manifest.json").write_text(json.dumps({
        # Manifestweites Feld absichtlich falsch/veraltet, damit ein Zurueckfallen
        # darauf im Test auffallen wuerde.
        "ninlight_version": "1.2.0",
        "images": [
            {"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img",
             "ninlight_version": "1.2.0"},
            {"target": "g95_cw", "variant": "prod", "file": "g95_cw_prod.img",
             "ninlight_version": "1.0.0"},
        ]}))
    srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1_ota_light.img")
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["recent"][-1]["version"] == "1.0.0"


def test_control_file_is_world_readable_even_under_a_restrictive_umask(tmp_path):
    """Dieselbe 0600-Falle wie bei status.json (siehe
    test_state_file_is_world_readable_even_under_a_restrictive_umask oben), aber fuer
    control.json: das CLI aus Task 4 muss die Datei lesen UND schreiben koennen."""
    old_umask = os.umask(0o077)
    try:
        srv = MigrationTestServer(tmp_path, "migration")
    finally:
        os.umask(old_umask)
    try:
        (srv.image_dir / "migrate-nocreds").mkdir(exist_ok=True)
        (srv.image_dir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
        srv.set_control(phase="armed", armed_target="g95_cw")

        def world_readable():
            mode = stat.S_IMODE(srv.control_path.stat().st_mode)
            return bool(mode & stat.S_IROTH)

        # set_control() oben schreibt direkt aus dem Test, nicht ueber den Server - das
        # will hier nicht geprueft werden. Erst die Anfrage unten bringt den Server selbst
        # dazu, control.json zu schreiben (armed -> listen), also denselben Codepfad
        # (Station.write_control()) wie write_state() fuer status.json.
        assert srv.get(
            "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200
        srv.read_control(lambda _: world_readable())
        assert world_readable(), "control.json ist nach dem Schreiben durch den Server " \
            "nicht welt-lesbar"
    finally:
        srv.stop()


def test_armed_serves_exactly_once_under_concurrency(migration_server):
    """C1 (Fixrunde 1): read_control()/Phasenpruefung/Auslieferung/write_control("listen")
    liefen ohne gemeinsame Sperre ab - zwei gleichzeitige Anfragen (zwei Lampen, oder Erst-
    und Range-Resume-Verbindung derselben Lampe) konnten beide phase == "armed" sehen und
    beide die volle Auslieferung bekommen. station.claim_delivery() muss das Lesen und das
    Verbrauchen der Freigabe atomar unter einer Sperre machen, damit von n gleichzeitigen
    Anfragen gegen dieselbe Freigabe hoechstens eine durchkommt."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG" * 20000)
    srv.set_control(phase="armed", armed_target="g95_cw")
    p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"

    n_workers = 20
    results = []
    results_lock = threading.Lock()

    def worker():
        r = srv.get(p)
        with results_lock:
            results.append(r.status)

    threads = [threading.Thread(target=worker) for _ in range(n_workers)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=10)

    assert len(results) == n_workers, "nicht alle Worker sind fertig geworden"
    assert results.count(200) == 1, f"erwartet genau eine 200, bekommen: {results}"
    assert results.count(404) == n_workers - 1
    assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen"


def test_control_write_failure_blocks_further_armed_deliveries(migration_server):
    """I3 (Fixrunde 1): schlaegt das Schreiben von control.json fehl (z.B. volles oder
    schreibgeschuetztes /run), darf die Freigabe nicht dauerhaft auf "armed" stehen bleiben
    - sonst wuerde jede weitere Lampe des Ziels ebenfalls bedient, genau die Luecke aus C1,
    nur diesmal durch einen Schreibfehler statt durch Nebenlaeufigkeit ausgeloest. Die eine
    Anfrage, die den Fehler ausloest, bekommt noch ihre Auslieferung (wie eine normale
    Freigabe); jede weitere schlaegt fehl, bis das Schreiben wieder klappt. Der Fehler muss
    ausserdem in status.json sichtbar sein, sonst wundert sich niemand, warum "arm" nicht
    mehr wirkt."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
    srv.set_control(phase="armed", armed_target="g95_cw")
    p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"

    control_dir = srv.control_path.parent
    old_mode = control_dir.stat().st_mode
    os.chmod(control_dir, 0o500)  # r-x fuer den Besitzer, kein Schreiben mehr moeglich
    try:
        r1 = srv.get(p)
        assert r1.status == 200, "die eine vorgesehene Auslieferung muss noch rausgehen"
        st = srv.status(lambda s: s.get("control_error"))
        assert st["control_error"], "Schreibfehler muss in status.json sichtbar sein"

        r2 = srv.get(p)
        assert r2.status == 404, \
            "ein Schreibfehler darf die Freigabe nicht dauerhaft offen lassen"
    finally:
        os.chmod(control_dir, old_mode)


def _load_server_module():
    """Laedt server.py wie test_log_error_is_not_silenced_by_log_message oben - als
    eigenstaendiges Modulobjekt, ohne es in sys.modules einzutragen, damit mehrfache
    Aufrufe (einer je resolve_mode-Test) sich nicht in die Quere kommen."""
    import importlib.util

    spec = importlib.util.spec_from_file_location("ninlight_station_server_rm", SERVER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_resolve_mode_explicit_mode_wins_over_file(tmp_path):
    """I5 (Fixrunde 1): --mode schlaegt --mode-file."""
    module = _load_server_module()
    mode_file = tmp_path / "mode"
    mode_file.write_text("migration")
    assert module.resolve_mode("recovery", mode_file) == "recovery"


def test_resolve_mode_reads_the_mode_file(tmp_path):
    module = _load_server_module()
    mode_file = tmp_path / "mode"
    mode_file.write_text("migration")
    assert module.resolve_mode(None, mode_file) == "migration"


def test_resolve_mode_missing_file_means_recovery(tmp_path):
    module = _load_server_module()
    mode_file = tmp_path / "mode"  # bewusst nicht angelegt
    assert module.resolve_mode(None, mode_file) == "recovery"


def test_resolve_mode_garbage_file_means_recovery(tmp_path):
    module = _load_server_module()
    mode_file = tmp_path / "mode"
    mode_file.write_text("garbage\n")
    assert module.resolve_mode(None, mode_file) == "recovery"


def test_resolve_mode_default_without_mode_file_is_recovery():
    module = _load_server_module()
    assert module.resolve_mode(None, None) == "recovery"


# --- Umschalter: mit oder ohne einkompilierte WLAN-Zugangsdaten ---------------------
#
# Zwei Bildsaetze liegen auf dem Pi (images/migrate-nocreds und images/migrate-wlan);
# welcher ausgeliefert wird, entscheidet allein der Umschalter. Geprueft wird deshalb
# immer an der TATSAECHLICH AUSGELIEFERTEN DATEI, nicht an einem Pfad oder einem
# Statusfeld - der Fehlgriff, gegen den der Umschalter da ist, ist ein falscher Inhalt
# in einer Lampe.

def _both_sets(imgdir):
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"OHNE" * 100)
    (imgdir / "migrate-wlan").mkdir(exist_ok=True)
    (imgdir / "migrate-wlan" / "g95_cw_dev.img").write_bytes(b"MITW" * 100)


G95_DEV = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"


def test_the_default_image_set_is_the_one_without_wifi_credentials(migration_server):
    """Ohne Umschalterdatei (Auslieferungszustand) geht der Abzug OHNE Zugangsdaten
    raus. Die sichere Vorgabe ist die, bei der ein Fehlgriff niemandem das eigene
    WLAN-Passwort in ein fremdes Geraet schreibt."""
    srv, imgdir = migration_server
    _both_sets(imgdir)
    r = srv.get(G95_DEV)
    assert r.status == 200
    assert r.body == b"OHNE" * 100


def test_the_switch_decides_which_file_is_actually_delivered(migration_server):
    srv, imgdir = migration_server
    _both_sets(imgdir)
    srv.set_image_set("mit-wlan")
    assert srv.get(G95_DEV).body == b"MITW" * 100
    # ... und wieder zurueck, ohne Dienstneustart: der Umschalter wirkt sofort.
    srv.set_image_set("ohne-wlan")
    assert srv.get(G95_DEV).body == b"OHNE" * 100


def test_a_broken_switch_file_falls_back_to_the_safe_set(migration_server):
    """Unbekannter Inhalt ist kein Grund, den teureren Satz zu nehmen."""
    srv, imgdir = migration_server
    _both_sets(imgdir)
    srv.set_image_set("voellig-anderes")
    assert srv.get(G95_DEV).body == b"OHNE" * 100


def test_a_selected_but_missing_image_set_delivers_nothing(migration_server):
    """Der gewaehlte Satz liegt nicht auf dem Pi: es darf NICHTS rausgehen - insbesondere
    nicht die Datei des anderen Satzes, die daneben liegt und genauso heisst."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    (imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"OHNE" * 100)
    srv.set_image_set("mit-wlan")          # migrate-wlan/ existiert nicht

    r = srv.get(G95_DEV)
    assert r.status == 404
    assert b"OHNE" not in r.body
    st = srv.status(lambda s: len(s["recent"]) >= 1)
    assert st["served"] == 0
    # Eigener Status, damit am Panel/in status.json "Satz fehlt" von "dieses Modell
    # fehlt" unterscheidbar bleibt.
    assert st["recent"][-1]["status"] == "noimageset"


def test_a_missing_image_set_does_not_burn_the_release(migration_server):
    """Wie beim fehlenden Einzelimage: eine Freigabe darf nicht daran verfallen, dass es
    gar nichts auszuliefern gab."""
    srv, imgdir = migration_server
    (imgdir / "migrate-nocreds").mkdir(exist_ok=True)
    srv.set_image_set("mit-wlan")
    srv.set_control(phase="armed", armed_target="g95_cw")
    assert srv.get(G95_DEV).status == 404
    assert srv.read_control()["phase"] == "armed"


def test_the_version_comes_from_the_manifest_of_the_selected_set(migration_server):
    """Nach dem Umschalten darf nicht die Version des vorigen Satzes stehenbleiben - die
    beiden manifest.json entstehen im selben Deploy und koennen dieselbe mtime tragen."""
    srv, imgdir = migration_server
    _both_sets(imgdir)
    for name, version in (("migrate-nocreds", "0.4.0-onb"), ("migrate-wlan", "0.4.0-wlan")):
        (imgdir / name / "manifest.json").write_text(json.dumps({
            "images": [{"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img",
                        "ninlight_version": version}]}))
    # Beiden Manifesten dieselbe mtime geben - genau der Fall, den ein reiner
    # mtime-Vergleich nicht bemerkt haette.
    stamp = 1600000000
    for name in ("migrate-nocreds", "migrate-wlan"):
        os.utime(imgdir / name / "manifest.json", (stamp, stamp))

    assert srv.get(G95_DEV).status == 200
    assert srv.status(lambda s: s["recent"])["recent"][-1]["version"] == "0.4.0-onb"

    srv.set_image_set("mit-wlan")
    assert srv.get(G95_DEV).status == 200
    assert srv.status(
        lambda s: len(s["recent"]) >= 2
    )["recent"][-1]["version"] == "0.4.0-wlan"


def test_read_image_set_defaults_without_a_file(tmp_path):
    module = _load_server_module()
    assert module.read_image_set(None) == "ohne-wlan"
    assert module.read_image_set(tmp_path / "fehlt") == "ohne-wlan"
    p = tmp_path / "imageset"
    p.write_text("mit-wlan\n")
    assert module.read_image_set(p) == "mit-wlan"