#!/usr/bin/env python3
"""Exercise the read-only client against a local HTTP server, without credentials."""
import datetime as dt
import email.utils
import http.server
import importlib.util
import json
from pathlib import Path
import socketserver
import threading
import unittest


spec = importlib.util.spec_from_file_location("generate", Path(__file__).with_name("generate.py"))
g = importlib.util.module_from_spec(spec)
spec.loader.exec_module(g)


class Clock:
    def __init__(self):
        self.t = 0.0
        self.sleeps = []

    def monotonic(self):
        return self.t

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.t += seconds

    def now(self):
        return dt.datetime(2026, 9, 8, tzinfo=g.UTC) + dt.timedelta(seconds=self.t)


class ReadTransportTest(unittest.TestCase):
    def setUp(self):
        self.responses, self.calls = [], []
        self.clock = Clock()
        owner = self

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_POST(self):
                body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
                owner.calls.append((self.path, body, owner.clock.t))
                status, headers, value = owner.responses.pop(0)
                self.send_response(status)
                for key, val in headers.items():
                    self.send_header(key, val)
                self.end_headers()
                self.wfile.write(json.dumps(value).encode())

            def log_message(self, *args):
                pass

        class LoopbackServer(http.server.HTTPServer):
            def server_bind(self):
                # Avoid HTTPServer's unrelated reverse-DNS lookup in an
                # otherwise local-only transport test.
                socketserver.TCPServer.server_bind(self)
                self.server_name, self.server_port = self.server_address

        self.server = LoopbackServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.reader = g.HubSpotReader(
            base=f"http://127.0.0.1:{self.server.server_port}", sleep=self.clock.sleep,
            monotonic=self.clock.monotonic, now=self.clock.now)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join()

    def read(self, path="/crm/v3/objects/contacts/search"):
        return self.reader.request("synthetic-test-token", path, "POST", {"limit": 1})

    def test_429_retries_identical_search_and_paces_next_batch_read(self):
        self.responses = [(429, {"Retry-After": "2"}, {"policyName": "SECONDLY"}),
                          (200, {}, {"total": 12}), (200, {}, {"results": []})]
        self.assertEqual(self.read(), {"total": 12})
        self.assertEqual(self.read("/crm/v3/objects/contacts/batch/read"), {"results": []})
        self.assertEqual(self.calls[0][:2], self.calls[1][:2])
        self.assertEqual([c[2] for c in self.calls], [0.0, 2.0, 2.25])

    def test_http_date_and_malformed_retry_after_have_bounded_backoff(self):
        after = email.utils.format_datetime(self.clock.now() + dt.timedelta(seconds=3), usegmt=True)
        self.responses = [(429, {"Retry-After": after}, {}),
                          (429, {"Retry-After": "invalid"}, {}), (200, {}, {"total": 1})]
        self.assertEqual(self.read(), {"total": 1})
        self.assertEqual([c[2] for c in self.calls], [0.0, 3.0, 5.0])

    def test_exhaustion_is_four_attempts_and_never_empty_success(self):
        self.responses = [(429, {}, {"policyName": "SECONDLY"})] * 4
        with self.assertRaises(g.HubSpotError) as err:
            self.read()
        self.assertEqual(err.exception.code, 429)
        self.assertEqual([c[2] for c in self.calls], [0.0, 1.0, 3.0, 7.0])

    def test_long_retry_after_is_not_clamped_to_an_early_retry(self):
        self.responses = [(429, {"Retry-After": "86400"}, {})]
        with self.assertRaises(g.HubSpotError):
            self.read()
        self.assertEqual(len(self.calls), 1)
        self.assertEqual(self.clock.sleeps, [])

    def test_permission_and_server_errors_are_not_retried(self):
        for status in (403, 500):
            self.responses = [(status, {}, {"message": "failure"})]
            before = len(self.calls)
            with self.assertRaises(g.HubSpotError) as err:
                self.read()
            self.assertEqual(err.exception.code, status)
            self.assertEqual(len(self.calls), before + 1)

    def test_mutations_and_unrecognized_paths_are_rejected_before_http(self):
        for method, path in [
            ("PATCH", "/crm/v3/objects/contacts/123"),
            ("POST", "/crm/v3/objects/contacts/batch/update"),
            ("POST", "/crm/v3/objects/contacts/search?different=1"),
            ("GET", "/crm/v3/objects/contacts/search"),
        ]:
            with self.assertRaises(ValueError):
                self.reader.request("synthetic", path, method, {})
        self.assertEqual(self.calls, [])


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