"""HTTP-слой (Http, EvaClient, HubClient) и конфиг. Перенос из jira-agent: сценарии Http без Jira-специфики."""
import contextlib
import io
import json
import os
import socket
import tempfile
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from unittest import mock

from _common import agent, eva_cfg


class MiniServer:
    """Крошечный сервер: routes — путь → (код, заголовки, тело-байты); requests — что пришло."""

    def __init__(self):
        self.routes = {}
        self.requests = []
        self.last_headers = {}
        srv = self

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

            def do_GET(self):
                srv.requests.append(self.path)
                srv.last_headers = {k.lower(): v for k, v in self.headers.items()}
                code, headers, body = srv.routes.get(self.path, (404, {}, b"{}"))
                self.send_response(code)
                for k, v in headers.items():
                    self.send_header(k, v)
                if "Content-Length" not in headers:
                    self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                if self.command != "HEAD":
                    self.wfile.write(body)

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), H)
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    @property
    def base_url(self):
        return "http://127.0.0.1:%d" % self.server.server_port

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


class ClientsTest(unittest.TestCase):
    def setUp(self):
        self.srv = MiniServer()
        self.eva = agent.EvaClient(eva_cfg(self.srv.base_url))

    def tearDown(self):
        self.srv.stop()

    def test_get_and_counters(self):
        self.srv.routes["/x"] = (200, {"Content-Type": "application/json"}, b'{"a": 1}')
        self.assertEqual(agent.Http.call(self.eva, "GET", "/x"), {"a": 1})
        self.assertEqual(self.eva.calls, 1)
        self.assertIsNotNone(self.eva.last_call_at)
        self.assertIsNotNone(self.eva.last_call_iso)
        self.eva.reset_counters()
        self.assertEqual((self.eva.calls, self.eva.rate_limited), (0, 0))

    def test_bearer_header_and_no_secret_in_repr(self):
        self.assertEqual(self.eva.headers["Authorization"], "Bearer eva-secret")
        self.assertNotIn("eva-secret", repr(self.eva))
        self.srv.routes["/x"] = (200, {}, b"{}")
        agent.Http.call(self.eva, "GET", "/x")
        self.assertEqual(self.srv.last_headers["authorization"], "Bearer eva-secret")

    def test_secret_not_in_log_output(self):
        buf = io.StringIO()
        self.srv.routes["/x"] = (200, {}, b"{}")
        with contextlib.redirect_stdout(buf):
            agent.Http.call(self.eva, "GET", "/x")
            agent.log("проверка")
        self.assertNotIn("eva-secret", buf.getvalue())

    def test_http_error_carries_code_text_headers(self):
        self.srv.routes["/x"] = (429, {"Retry-After": "7"}, b"slow down")
        with self.assertRaises(agent.HttpError) as cm:
            agent.Http.call(self.eva, "GET", "/x")
        self.assertEqual(cm.exception.code, 429)
        self.assertEqual(cm.exception.headers.get("retry-after"), "7")
        self.assertIn("slow down", cm.exception.text)
        self.assertEqual(self.eva.calls, 1)

    def test_fetch_bytes(self):
        self.srv.routes["/f"] = (200, {}, b"\x89PNG\x00\x01")
        self.assertEqual(self.eva.fetch_bytes("/f", 100), b"\x89PNG\x00\x01")
        self.assertEqual(self.srv.last_headers["authorization"], "Bearer eva-secret")
        self.assertEqual(self.eva.calls, 1)

    def test_fetch_bytes_limit(self):
        self.srv.routes["/f"] = (200, {}, b"x" * 20)
        with self.assertRaises(agent.TooBig):
            self.eva.fetch_bytes("/f", 10)
        self.assertEqual(self.eva.fetch_bytes("/f", 20), b"x" * 20)

    def test_fetch_bytes_redirect_same_host(self):
        self.srv.routes["/f"] = (302, {"Location": self.srv.base_url + "/moved"}, b"")
        self.srv.routes["/moved"] = (200, {}, b"abc")
        self.assertEqual(self.eva.fetch_bytes("/f", 100), b"abc")

    def test_fetch_bytes_redirect_other_host(self):
        port = self.srv.server.server_port
        self.srv.routes["/f"] = (302, {"Location": "http://localhost:%d/moved" % port}, b"")
        self.srv.routes["/moved"] = (200, {}, b"abc")
        with self.assertRaises(agent.RedirectAway) as e:
            self.eva.fetch_bytes("/f", 100)
        self.assertEqual(e.exception.host, "localhost")
        self.assertNotIn("/moved", self.srv.requests)

    def test_fetch_bytes_cut_body_is_connection_error(self):
        self.srv.routes["/f"] = (200, {"Content-Length": "40"}, b"x" * 10)       # обещано 40, отдано 10
        with self.assertRaises(ConnectionError):
            self.eva.fetch_bytes("/f", 1000)

    def test_json_call_redirect_other_host_refused(self):
        port = self.srv.server.server_port
        self.srv.routes["/a"] = (302, {"Location": "http://localhost:%d/b" % port}, b"")
        self.srv.routes["/b"] = (200, {}, b"{}")
        with self.assertRaises(agent.HttpError) as e:
            agent.Http.call(self.eva, "GET", "/a")
        self.assertIn("localhost", e.exception.text)
        self.assertNotIn("/b", self.srv.requests)

    def test_json_call_redirect_same_host_followed(self):
        self.srv.routes["/a"] = (302, {"Location": self.srv.base_url + "/b"}, b"")
        self.srv.routes["/b"] = (200, {}, b'{"ok": true}')
        self.assertEqual(agent.Http.call(self.eva, "GET", "/a"), {"ok": True})

    def test_error_body_read_failure_is_connection_error(self):
        import urllib.error

        class Broken(io.BytesIO):
            def read(self, *a):
                raise TimeoutError("timed out")

        err = urllib.error.HTTPError("http://x", 500, "boom", {}, Broken())
        with mock.patch("urllib.request.OpenerDirector.open", side_effect=err):
            with self.assertRaises(ConnectionError):
                agent.Http.call(self.eva, "GET", "/x")
        self.assertEqual(self.eva.calls, 1)

    def test_incomplete_read_is_connection_error(self):
        import http.client
        with mock.patch("urllib.request.OpenerDirector.open", side_effect=http.client.IncompleteRead(b"ab", 10)):
            with self.assertRaises(ConnectionError):
                agent.Http.call(self.eva, "GET", "/x")

    def test_incomplete_read_inside_response_body(self):
        srv = socket.socket()
        srv.bind(("127.0.0.1", 0))
        srv.listen(1)

        def serve():
            c, _ = srv.accept()
            c.recv(4096)
            c.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{\"a\":")
            c.close()

        t = threading.Thread(target=serve, daemon=True)
        t.start()
        client = agent.EvaClient(eva_cfg("http://127.0.0.1:%d" % srv.getsockname()[1]))
        try:
            with self.assertRaises(ConnectionError):
                agent.Http.call(client, "GET", "/x")
        finally:
            t.join(2)
            srv.close()

    def test_unreachable_raises_connection_error(self):
        with self.assertRaises(ConnectionError):
            agent.Http.call(agent.EvaClient(eva_cfg("http://127.0.0.1:1")), "GET", "/x")

    def test_non_json_200_becomes_http_error(self):
        self.srv.routes["/x"] = (200, {}, b"<html>\xff\xfe not json")
        with self.assertRaises(agent.HttpError) as cm:
            agent.Http.call(self.eva, "GET", "/x")
        self.assertEqual(cm.exception.code, 200)
        self.assertEqual(self.eva.calls, 1)

    def test_hub_client_claim_and_wiki_pages_routes(self):
        calls = []
        hub = agent.HubClient(eva_cfg(hub_url="http://x", hub_token="t"))
        hub.call = lambda method, path, body=None, params=None, timeout=None: (calls.append((method, path, body, timeout)), {})[1]
        hub.claim(1, 2, None)
        hub.wiki_pages(7, [{"code": "DOC-1"}])
        hub.upload_file(b"abc")
        self.assertEqual(calls[0][:2], ("POST", "/agent/claim"))
        self.assertEqual(calls[0][2]["last_call_at"], None)
        self.assertGreaterEqual(calls[0][3], 45)
        self.assertEqual(calls[1], ("POST", "/agent/wiki/pages", {"command_id": 7, "pages": [{"code": "DOC-1"}]}, None))
        self.assertEqual(calls[2][:2], ("POST", "/agent/files"))
        self.assertEqual(calls[2][2], {"content": "YWJj"})

    def test_load_config_requires_fields(self):
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")

            def write(cfg):
                with open(path, "w", encoding="utf-8") as f:
                    json.dump(cfg, f)

            for bad in ({"hub_url": "http://x", "hub_token": "t"},
                        {"hub_url": "http://x", "hub_token": "t", "eva": {"base_url": "https://e"}},
                        {"hub_url": "http://x", "hub_token": "t", "eva": {"token": "z"}},
                        {"hub_url": "http://x", "hub_token": "t", "eva": "https://e"}):
                write(bad)
                with self.assertRaises(SystemExit):
                    agent.load_config(path)
            write({"hub_url": "http://x", "hub_token": "t", "eva": {"base_url": "https://e", "token": "z"}})
            cfg = agent.load_config(path)
            self.assertEqual((cfg["timeout"], cfg["poll_seconds"], cfg["verify_ssl"]), (60, 5, True))
            self.assertNotIn("auto_update", cfg)             # умолчание true применяет _run: `is not False`
            with self.assertRaises(SystemExit):
                agent.load_config(os.path.join(d, "нет.json"))

    def test_example_config_loads(self):
        cfg = agent.load_config(os.path.join(agent.HERE, "agent.example.json"))
        self.assertEqual(cfg["auto_update"], True)
        self.assertIn("base_url", cfg["eva"])

    def test_check_command(self):
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(eva_cfg(), f)
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                self.assertEqual(agent.main(["--config", path, "check"]), 3)     # хаб и Eva недоступны (порт 1)
        self.assertIn("хаб: НЕДОСТУПЕН", out.getvalue())

    def test_selftest_command(self):
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(eva_cfg(), f)
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                self.assertEqual(agent.main(["--config", path, "selftest"]), 3)  # недоступность — 3, а не 1
        self.assertIn("eva-agent v%d" % agent.AGENT_VERSION, out.getvalue())


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