"""Local integration tests for the sibling download_file.py. Python 3.10+."""
import hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import time

PAYLOAD = b"id,value\n1,42\n"
SHA256 = hashlib.sha256(PAYLOAD).hexdigest()
CLIENT = Path(__file__).with_name("download_file.py").resolve()


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

    def do_GET(self):
        if self.path == "/missing":
            self.send_error(404)
            return
        if self.path == "/redirect":
            self.send_response(302)
            self.send_header("Location", "/ok")
            self.end_headers()
            return
        if self.path == "/slow":
            time.sleep(0.3)
        self.send_response(200)
        if self.path not in {"/no-length", "/big-stream"}:
            length = len(PAYLOAD) + (10 if self.path == "/short" else 0)
            self.send_header("Content-Length", str(length))
        if self.path == "/encoded":
            self.send_header("Content-Encoding", "gzip")
        self.end_headers()
        try:
            self.wfile.write(PAYLOAD)
        except (BrokenPipeError, ConnectionResetError):
            pass


def main():
    assert CLIENT.is_file(), CLIENT
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    base = f"http://127.0.0.1:{server.server_port}"
    environment = dict(os.environ, no_proxy="127.0.0.1", NO_PROXY="127.0.0.1")
    try:
        with tempfile.TemporaryDirectory(prefix="download-selftest-") as work:
            root = Path(work)
            target = root / "result.csv"

            def run(route, digest=SHA256, limit=1024, timeout="1"):
                return subprocess.run(
                    [sys.executable, str(CLIENT), base + route, str(target),
                     "--sha256", digest, "--max-bytes", str(limit),
                     "--timeout", timeout], env=environment,
                    capture_output=True, text=True, timeout=5, check=False,
                )

            for route in ("/ok", "/no-length"):
                target.write_bytes(b"old result\n")
                result = run(route)
                assert result.returncode == 0, result
                assert not result.stderr, result
                report = json.loads(result.stdout)
                assert report["bytes"] == len(PAYLOAD) and report["sha256"] == SHA256
                assert target.read_bytes() == PAYLOAD
                assert not list(root.glob(".*.part"))

            cases = [
                ("/ok", "0" * 64, 1024, "1", "SHA-256 mismatch"),
                ("/short", SHA256, 1024, "1", "truncated response"),
                ("/ok", SHA256, 5, "1", "Content-Length exceeds"),
                ("/big-stream", SHA256, 5, "1", "body exceeds"),
                ("/missing", SHA256, 1024, "1", "404"),
                ("/redirect", SHA256, 1024, "1", "302"),
                ("/encoded", SHA256, 1024, "1", "Content-Encoding"),
                ("/slow", SHA256, 1024, "0.05", "timed out"),
            ]
            for route, digest, limit, timeout, message in cases:
                target.write_bytes(b"old result\n")
                result = run(route, digest, limit, timeout)
                assert result.returncode == 1, (route, result)
                assert not result.stdout and message in result.stderr, (route, result)
                assert target.read_bytes() == b"old result\n", route
                assert not list(root.glob(".*.part")), route

            target.unlink()
            result = run("/ok", "0" * 64)
            assert result.returncode == 1 and not target.exists()
            assert not list(root.glob(".*.part"))
            for digest, limit, timeout in [("bad", 1024, "1"), (SHA256, 0, "1"),
                                           (SHA256, 1024, "nan")]:
                result = run("/ok", digest, limit, timeout)
                assert result.returncode == 2 and not result.stdout, result
            print("PASS: 2 successes, 8 failures preserve old output; no partial files")
            print("PASS: failed first download leaves no output; 3 invalid arguments rejected")
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=2)


if __name__ == "__main__":
    main()