"""Spawning a child asks endpoints that never answer once, together.

A bare model the parent's provider does not list is looked up on every other
profile, and a generic profile asks its endpoint's `/models`. An endpoint that
accepts connections and never answers costs one fetch timeout: asked one after
another, three of them held a spawn for three timeouts, and two spawns that
arrived together each opened their own connection and waited their own
timeout. The wait is the daemon's, so this runs through the real daemon.
"""

import json
import socket
import threading
import time
import unittest

from harness import Albedo, Provider, operation_id, text

STALLED = 3
# One fetch timeout is 5 s; asked in sequence, three took at least 15.
TOGETHER_SECONDS = 12


class StalledEndpointsTests(unittest.TestCase):
    def setUp(self):
        self.provider = Provider(lambda _request: text("ok"))
        self.addCleanup(self.provider.close)
        self.held = []
        profiles = {}
        for name in ("alpha", "beta"):
            profiles[f"{name}-{self.provider.route}"] = self.profile(
                self.provider.url + f"/{name}/v1",
                f"fixture-{name}-{self.provider.route}",
            )
        for number in range(STALLED):
            profiles[f"stalled{number}-{self.provider.route}"] = self.profile(
                self.stall(), f"fixture-stalled{number}-{self.provider.route}"
            )
        self.alpha = f"alpha-{self.provider.route}"
        self.beta = f"beta-{self.provider.route}"
        self.model = f"fixture-beta-{self.provider.route}"
        self.app = Albedo(self.provider, providers=profiles).__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def profile(self, url, model):
        return {
            "extension": "openai",
            "baseUrl": url,
            "apiKey": "key",
            "model": model,
            "protocol": "chat_completions",
        }

    def stall(self):
        """An endpoint url whose server accepts connections and never answers."""
        server = socket.create_server(("127.0.0.1", 0))
        connections = []

        def hold():
            while True:
                try:
                    connections.append(server.accept()[0])
                except OSError:
                    return

        threading.Thread(target=hold, daemon=True).start()
        self.held.append(connections)
        self.addCleanup(lambda: [c.close() for c in connections])
        self.addCleanup(server.close)
        return f"http://127.0.0.1:{server.getsockname()[1]}/v1"

    def spawn(self, parent, name):
        session = operation_id()
        body = {
            "kind": "child",
            "parent_id": parent,
            "address": name,
            "name": name,
            "initial_input_id": operation_id(),
            "task": "scan",
            "model": self.model,
        }
        with self.app.api(
            f"/sessions/{session}", body, method="PUT", headers={"If-None-Match": "*"}
        ) as response:
            return json.load(response)

    def test_stalled_endpoints_are_asked_together_not_in_sequence(self):
        parent = self.app.session()
        began = time.monotonic()
        made = self.spawn(parent, "scout")
        elapsed = time.monotonic() - began
        self.assertEqual(made["provider_profile"], self.beta)
        self.assertLess(elapsed, TOGETHER_SECONDS)
        self.assertEqual([len(c) for c in self.held], [1] * STALLED)

    def test_spawns_that_arrive_together_share_one_connection_per_endpoint(self):
        parent = self.app.session()
        made = []
        threads = [
            threading.Thread(target=lambda n=n: made.append(self.spawn(parent, n)))
            for n in ("scout-1", "scout-2")
        ]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        self.assertEqual(len(made), 2)
        self.assertEqual([len(c) for c in self.held], [1] * STALLED)


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