test_async_http_infrastructure.py

"""Tests for async HTTP infrastructure (issue #6369).

Covers:
- WebhookCircuitBreaker state machine (CLOSED -> OPEN -> HALF_OPEN -> CLOSED)
- Per-target circuit breaker registry
- Latest-wins dropping pattern for audio byte webhooks
- Semaphore bounded concurrency getters
- Shared executors from utils/executors.py
"""

import asyncio
import sys
import time
import types
from pathlib import Path
from unittest.mock import patch

import pytest

BACKEND_DIR = Path(__file__).resolve().parents[2]


def _ensure_package(name, path):
    module = sys.modules.get(name)
    if module is None or not hasattr(module, "__path__"):
        module = types.ModuleType(name)
        sys.modules[name] = module
    module.__path__ = [str(path)]

    if "." in name:
        parent_name, attr_name = name.rsplit(".", 1)
        parent = sys.modules.get(parent_name)
        if parent is not None:
            setattr(parent, attr_name, module)


def _drop_stale_module(name, required_attrs):
    module = sys.modules.get(name)
    if module is None:
        return
    if (
        isinstance(module, types.ModuleType)
        and getattr(module, "__file__", None)
        and all(hasattr(module, attr) for attr in required_attrs)
    ):
        return

    sys.modules.pop(name, None)
    parent_name, attr_name = name.rsplit(".", 1)
    parent = sys.modules.get(parent_name)
    if parent is not None and getattr(parent, attr_name, None) is module:
        delattr(parent, attr_name)


_ensure_package("utils", BACKEND_DIR / "utils")
_drop_stale_module("utils.http_client", ["WebhookCircuitBreaker", "get_webhook_circuit_breaker"])
_drop_stale_module("utils.executors", ["critical_executor", "storage_executor", "shutdown_executors"])

from utils.http_client import (
    WebhookCircuitBreaker,
    get_webhook_circuit_breaker,
    latest_wins_start,
    latest_wins_check,
    get_webhook_semaphore,
    get_maps_semaphore,
    get_auth_semaphore,
    get_stt_semaphore,
    _webhook_circuit_breakers,
    _latest_wins_versions,
    _semaphores,
    _SEMAPHORE_CACHE_MAX,
    _CIRCUIT_BREAKER_FAILURE_THRESHOLD,
    _CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
)
from utils.executors import critical_executor, storage_executor

# ============================================================================
# WebhookCircuitBreaker
# ============================================================================


class TestWebhookCircuitBreaker:
    """Circuit breaker state machine tests."""

    def test_initial_state_is_closed(self):
        cb = WebhookCircuitBreaker("test-host")
        assert cb.state == 'closed'
        assert cb.allow_request() is True

    def test_stays_closed_below_threshold(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD - 1):
            cb.record_failure()
        assert cb.state == 'closed'
        assert cb.allow_request() is True

    def test_opens_at_threshold(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        assert cb.state == 'open'
        assert cb.allow_request() is False

    def test_success_resets_failure_count(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD - 1):
            cb.record_failure()
        cb.record_success()
        # Now failures are reset, need full threshold again
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD - 1):
            cb.record_failure()
        assert cb.state == 'closed'

    def test_open_to_half_open_after_timeout(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        assert cb.state == 'open'

        # Simulate time passing beyond recovery timeout
        cb._last_failure_time = time.monotonic() - _CIRCUIT_BREAKER_RECOVERY_TIMEOUT - 1
        assert cb.state == 'half_open'

    def test_half_open_allows_one_probe(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        cb._last_failure_time = time.monotonic() - _CIRCUIT_BREAKER_RECOVERY_TIMEOUT - 1

        assert cb.state == 'half_open'
        assert cb.allow_request() is True  # first probe
        assert cb.allow_request() is False  # second blocked

    def test_half_open_success_closes(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        cb._last_failure_time = time.monotonic() - _CIRCUIT_BREAKER_RECOVERY_TIMEOUT - 1

        assert cb.allow_request() is True
        cb.record_success()
        assert cb.state == 'closed'
        assert cb.allow_request() is True

    def test_half_open_failure_reopens(self):
        cb = WebhookCircuitBreaker("test-host")
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        cb._last_failure_time = time.monotonic() - _CIRCUIT_BREAKER_RECOVERY_TIMEOUT - 1

        assert cb.allow_request() is True
        # Fail again — should go back to open
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        assert cb.state == 'open'

    def test_half_open_single_failed_probe_reopens_immediately(self):
        """A single failure during HALF_OPEN must reopen the breaker immediately.

        This tests the strict probe semantics: the breaker grants exactly one
        request in HALF_OPEN; if that probe fails, it must reopen without
        requiring the full failure threshold to be reached again.
        """
        cb = WebhookCircuitBreaker("test-host")
        # Drive to OPEN
        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
            cb.record_failure()
        assert cb.state == 'open'

        # Age the breaker past recovery timeout → HALF_OPEN
        cb._last_failure_time = time.monotonic() - _CIRCUIT_BREAKER_RECOVERY_TIMEOUT - 1
        assert cb.state == 'half_open'

        # Allow the single probe
        assert cb.allow_request() is True

        # One failure — must reopen without needing additional failures
        cb.record_failure()
        assert cb.state == 'open', (
            "A single probe failure in HALF_OPEN must immediately reopen the breaker, "
            "not wait for the full failure threshold"
        )
        assert cb.allow_request() is False, "Breaker must block requests after single probe failure"


# ============================================================================
# Circuit breaker registry
# ============================================================================


class TestCircuitBreakerRegistry:
    """Per-target circuit breaker lookup tests."""

    def setup_method(self):
        _webhook_circuit_breakers.clear()

    def test_same_url_returns_same_instance(self):
        cb1 = get_webhook_circuit_breaker("https://example.com/path1")
        cb2 = get_webhook_circuit_breaker("https://example.com/path1")
        assert cb1 is cb2

    def test_same_url_ignores_query_params(self):
        cb1 = get_webhook_circuit_breaker("https://example.com/hook?key=1")
        cb2 = get_webhook_circuit_breaker("https://example.com/hook?key=2")
        assert cb1 is cb2

    def test_different_paths_return_different_instances(self):
        cb1 = get_webhook_circuit_breaker("https://example.com/path1")
        cb2 = get_webhook_circuit_breaker("https://example.com/path2")
        assert cb1 is not cb2

    def test_different_hosts_return_different_instances(self):
        cb1 = get_webhook_circuit_breaker("https://foo.com/hook")
        cb2 = get_webhook_circuit_breaker("https://bar.com/hook")
        assert cb1 is not cb2

    def test_invalid_url_fallback(self):
        cb = get_webhook_circuit_breaker("not-a-url")
        assert cb is not None
        assert cb.state == 'closed'


# ============================================================================
# Latest-wins dropping
# ============================================================================


class TestLatestWins:
    """Latest-wins version tracking for audio byte webhooks."""

    def setup_method(self):
        _latest_wins_versions.clear()

    def test_start_increments_version(self):
        v1 = latest_wins_start("uid-1")
        v2 = latest_wins_start("uid-1")
        assert v2 == v1 + 1

    def test_check_passes_for_latest(self):
        v = latest_wins_start("uid-1")
        assert latest_wins_check("uid-1", v) is True

    def test_check_fails_for_stale(self):
        v1 = latest_wins_start("uid-1")
        latest_wins_start("uid-1")  # v2 supersedes v1
        assert latest_wins_check("uid-1", v1) is False

    def test_independent_uid_tracking(self):
        v_a = latest_wins_start("uid-a")
        v_b = latest_wins_start("uid-b")
        assert latest_wins_check("uid-a", v_a) is True
        assert latest_wins_check("uid-b", v_b) is True

    def test_check_unknown_uid_returns_false(self):
        assert latest_wins_check("nonexistent", 1) is False


# ============================================================================
# Semaphore getters
# ============================================================================


class TestSemaphoreGetters:
    """Verify semaphore creation and per-loop isolation."""

    def test_webhook_semaphore_returns_semaphore(self):
        sem = get_webhook_semaphore()
        assert isinstance(sem, asyncio.Semaphore)

    def test_maps_semaphore_returns_semaphore(self):
        sem = get_maps_semaphore()
        assert isinstance(sem, asyncio.Semaphore)

    def test_auth_semaphore_returns_semaphore(self):
        sem = get_auth_semaphore()
        assert isinstance(sem, asyncio.Semaphore)

    def test_stt_semaphore_returns_semaphore(self):
        sem = get_stt_semaphore()
        assert isinstance(sem, asyncio.Semaphore)

    @pytest.mark.asyncio
    async def test_same_loop_returns_same_instance(self):
        """Within the same event loop, getter returns the same semaphore."""
        sem1 = get_webhook_semaphore()
        sem2 = get_webhook_semaphore()
        assert sem1 is sem2

    @pytest.mark.asyncio
    async def test_pruning_keeps_the_running_loops_semaphore(self):
        """Crossing the cache cap must not swap the live loop's existing semaphore.

        The prune used _semaphores.clear(), which dropped the running loop's entries too.
        A caller already holding permits on the old Semaphore kept them while the next
        caller received a brand-new one, so the effective concurrency briefly doubled —
        the opposite of what the cap exists for. The docstring already promised the main
        loop's semaphores "are stable" and that only short-lived asyncio.run() entries
        are pruned.
        """
        _semaphores.clear()
        webhook_before = get_webhook_semaphore()
        live_loop_id = id(asyncio.get_running_loop())

        # Fill the cache past the cap with entries from other (destroyed) loops.
        for i in range(_SEMAPHORE_CACHE_MAX + 5):
            _semaphores[(live_loop_id + 1 + i, 'webhook')] = asyncio.Semaphore(1)

        # A new name for this loop is what triggers the prune.
        get_maps_semaphore()

        assert get_webhook_semaphore() is webhook_before, 'the running loop lost its semaphore to the prune'

    @pytest.mark.asyncio
    async def test_pruning_still_bounds_the_cache(self):
        """The prune must drop the foreign-loop entries it was added for."""
        _semaphores.clear()
        get_webhook_semaphore()
        live_loop_id = id(asyncio.get_running_loop())
        for i in range(_SEMAPHORE_CACHE_MAX + 5):
            _semaphores[(live_loop_id + 1 + i, 'webhook')] = asyncio.Semaphore(1)

        get_maps_semaphore()  # crosses the cap, triggering the prune

        assert all(key[0] == live_loop_id for key in _semaphores), 'foreign-loop entries survived'
        assert len(_semaphores) == 2  # webhook + maps, both for this loop

    def test_different_loops_return_different_instances(self):
        """Different asyncio.run() calls get isolated semaphores."""
        sems = []

        async def _get():
            return get_webhook_semaphore()

        sems.append(asyncio.run(_get()))
        _semaphores.clear()  # Ensure no stale entries from the destroyed loop
        sems.append(asyncio.run(_get()))
        assert sems[0] is not sems[1]


# ============================================================================
# Shared executors
# ============================================================================


class TestSharedExecutors:
    """Verify dedicated thread pool executors are functional."""

    def test_critical_executor_submits(self):
        future = critical_executor.submit(lambda: 42)
        assert future.result(timeout=5) == 42

    def test_storage_executor_submits(self):
        future = storage_executor.submit(lambda: "ok")
        assert future.result(timeout=5) == "ok"

    def test_critical_executor_thread_name_prefix(self):
        import threading

        result = critical_executor.submit(lambda: threading.current_thread().name).result(timeout=5)
        assert result.startswith("critical")

    def test_storage_executor_thread_name_prefix(self):
        import threading

        result = storage_executor.submit(lambda: threading.current_thread().name).result(timeout=5)
        assert result.startswith("storage")

    def test_critical_executor_parallel_work(self):
        """Verify critical executor handles concurrent submissions."""
        import time

        def slow_task(n):
            time.sleep(0.05)
            return n * 2

        futures = [critical_executor.submit(slow_task, i) for i in range(4)]
        results = [f.result(timeout=5) for f in futures]
        assert results == [0, 2, 4, 6]


class TestShutdownLifecycle:
    """Verify shutdown functions exist and are callable."""

    def test_shutdown_executors_callable(self):
        """shutdown_executors must be a callable function."""
        from utils.executors import shutdown_executors

        assert callable(shutdown_executors)

    def test_shutdown_executors_registered_with_atexit(self):
        """shutdown_executors must be registered via atexit."""
        import atexit

        from utils.executors import shutdown_executors

        # atexit._run_exitfuncs stores registered callables; check it's registered
        # We verify by checking the function exists and is registered
        # (atexit internals are implementation-dependent, so we just verify callability
        #  and that calling it on a fresh executor doesn't raise)
        from concurrent.futures import ThreadPoolExecutor

        test_exec = ThreadPoolExecutor(max_workers=1, thread_name_prefix="test-shutdown")
        test_exec.shutdown(wait=False, cancel_futures=True)  # Should not raise

    def test_close_all_clients_resets_semaphores(self):
        """close_all_clients must clear the semaphore cache."""
        # Populate semaphore cache
        sem = get_webhook_semaphore()
        assert isinstance(sem, asyncio.Semaphore)

        async def _close():
            from utils.http_client import close_all_clients

            await close_all_clients()

        asyncio.run(_close())

        # After close, semaphore cache should be cleared
        assert len(_semaphores) == 0


# ============================================================================
# Client configuration assertions (tester-requested)
# ============================================================================


class TestWebhookClientConfig:
    """Verify webhook client is configured with correct timeout and limits."""

    def test_webhook_client_read_timeout_is_30s(self):
        """Webhook client must use 30s read timeout to match previous per-call behavior."""

        async def _read_timeout():
            from utils.http_client import close_all_clients, get_webhook_client

            try:
                return get_webhook_client().timeout.read
            finally:
                await close_all_clients()

        assert asyncio.run(_read_timeout()) == 30.0

    def test_webhook_client_connect_timeout_is_2s(self):
        """Webhook client must use aggressive 2s connect timeout."""

        async def _connect_timeout():
            from utils.http_client import close_all_clients, get_webhook_client

            try:
                return get_webhook_client().timeout.connect
            finally:
                await close_all_clients()

        assert asyncio.run(_connect_timeout()) == 2.0


class TestClientEventLoopOwnership:
    """A shared client belongs to the event loop that opened its connections.

    Regression for the prod failure where one process-wide client outlived the
    `asyncio.run()` loop that pooled its keep-alive connections: the next live
    loop made the pool discard one, uvloop raised `RuntimeError: ... the
    handler is closed` from `write_eof()` on the freed handle, and httpcore
    re-raised it at the caller — intermittent HTTP 500s on `/v1/apps/enable`
    and dropped app-integration webhook deliveries.
    """

    def test_each_event_loop_gets_its_own_client(self):
        """The surviving client of a finished loop is never handed to the next one."""
        import utils.http_client as hc

        async def _client_id():
            return id(hc.get_webhook_client())

        # No close in between: this is the prod shape, where the process-wide
        # client outlived the asyncio.run() loop that pooled its connections.
        first = asyncio.run(_client_id())
        second = asyncio.run(_client_id())
        try:
            assert first != second
        finally:
            asyncio.run(hc.close_all_clients())

    def test_one_loop_reuses_a_single_pooled_client(self):
        import utils.http_client as hc

        async def _same_client():
            try:
                return hc.get_webhook_client() is hc.get_webhook_client()
            finally:
                await hc.close_all_clients()

        assert asyncio.run(_same_client()) is True

    def test_finished_loops_do_not_accumulate_clients(self):
        import utils.http_client as hc

        async def _touch():
            hc.get_webhook_client()

        for _ in range(5):
            asyncio.run(_touch())

        assert len(hc._clients) == 1  # only the most recent loop's entry survives


class TestExecutorConfiguration:
    """Verify executor pool sizing cannot silently regress."""

    def test_critical_executor_has_8_workers(self):
        """critical_executor documented as 8 workers for latency-sensitive work."""
        assert critical_executor._max_workers == 8

    def test_storage_executor_has_128_workers(self):
        """storage_executor sized for 128 workers to handle concurrent private cloud uploads (#7376)."""
        assert storage_executor._max_workers == 128


class TestNotificationWebhookWiring:
    """Pin how the daily-summary webhook is owned: run inline, never submitted to a pool.

    The name predates two rewirings. It was storage_executor, then postprocess_executor
    (#7387), and is now an inline awaited call bounded by its own budget (#12530) --
    because the coordinator was already running on postprocess_executor, so submitting
    back into it made the function its own child and left the coroutine unowned at exit.
    """

    def test_day_summary_webhook_is_awaited_inline_not_submitted(self):
        """The daily-summary webhook must run inline, never be submitted to a pool.

        This pin used to require the opposite — ``postprocess_executor.submit(asyncio.run,
        day_summary_webhook(...))`` (#7387). That wiring turned out to be wrong in both
        directions (#12530): ``_send_summary_notification`` already runs *on*
        postprocess_executor, so the submit made it its own child, and the Cloud Run Job
        exits without joining the pool, so a queued webhook was dropped outright.

        The behavioral coverage lives in test_daily_summary_generation.py; this stays a
        source pin because the defect is a wiring shape, and it is the negative half —
        "not submitted anywhere" — that a behavioral test cannot express.
        """
        import os

        # Read source to verify pattern without triggering Firestore imports
        backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        with open(os.path.join(backend_dir, 'utils', 'other', 'notifications.py'), encoding='utf-8') as f:
            src = f.read()

        assert '.submit(asyncio.run, day_summary_webhook(' not in src
        assert 'asyncio.run(' in src and 'day_summary_webhook(' in src
        # Bounded inside the per-user budget it now shares: a slow receiver must not be
        # able to spend someone else's recap. See DAILY_SUMMARY_WEBHOOK_BUDGET_SECONDS.
        assert 'timeout=DAILY_SUMMARY_WEBHOOK_BUDGET_SECONDS' in src
        assert 'critical_executor' not in src
        assert 'storage_executor' not in src


class TestPrivateCloudQueueCap:
    """Verify private_cloud_queue uses bounded deque to prevent OOM."""

    def test_pusher_uses_deque_with_maxlen(self):
        """private_cloud_queue must be deque(maxlen=PRIVATE_CLOUD_QUEUE_MAX_SIZE)."""
        import ast
        import os

        backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        with open(os.path.join(backend_dir, 'routers', 'pusher.py'), encoding='utf-8') as f:
            src = f.read()

        assert 'deque(maxlen=PRIVATE_CLOUD_QUEUE_MAX_SIZE)' in src
        assert 'private_cloud_queue: List[dict] = []' not in src

    def test_queue_max_size_is_20(self):
        """Queue cap should be 20 items (~18MB max per connection, safe for 30-conn pods)."""
        import ast
        import os

        backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        with open(os.path.join(backend_dir, 'utils', 'pusher_protocol.py'), encoding='utf-8') as f:
            src = f.read()

        tree = ast.parse(src)
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign):
                for target in node.targets:
                    if isinstance(target, ast.Name) and target.id == 'PRIVATE_CLOUD_QUEUE_MAX_SIZE':
                        assert isinstance(node.value, ast.Constant)
                        assert node.value.value == 20
                        return
        pytest.fail("PRIVATE_CLOUD_QUEUE_MAX_SIZE constant not found")

    def test_overflow_warning_at_all_enqueue_points(self):
        """All 3 enqueue points must log overflow warning before deque drops oldest."""
        import os

        backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        with open(os.path.join(backend_dir, 'routers', 'pusher.py'), encoding='utf-8') as f:
            src = f.read()

        # Count occurrences of the overflow warning pattern
        warning_count = src.count('private_cloud_queue full')
        assert warning_count == 3, f"Expected 3 overflow warnings, found {warning_count}"

    def test_deque_maxlen_drops_oldest(self):
        """Verify deque(maxlen=N) drops oldest item when full."""
        from collections import deque

        q = deque(maxlen=3)
        q.append({'id': 1})
        q.append({'id': 2})
        q.append({'id': 3})
        assert len(q) == 3
        q.append({'id': 4})  # oldest (id=1) should be dropped
        assert len(q) == 3
        assert q[0]['id'] == 2
        assert q[-1]['id'] == 4


class TestCircuitBreakerAccessTracking:
    """Verify circuit breaker eviction uses last-access time."""

    def test_active_breaker_not_evicted(self):
        """Actively used breaker should not be evicted even with 0 failures."""
        import time
        from utils.http_client import (
            _webhook_circuit_breakers,
            get_webhook_circuit_breaker,
            _evict_stale_circuit_breakers,
            _CIRCUIT_BREAKER_IDLE_TTL,
        )

        _webhook_circuit_breakers.clear()
        cb = get_webhook_circuit_breaker('https://active.test/hook')
        cb.allow_request()  # Updates _last_access_time to now
        assert cb._last_failure_time == 0.0  # Never failed

        _evict_stale_circuit_breakers()
        assert 'https://active.test/hook' in _webhook_circuit_breakers
        _webhook_circuit_breakers.clear()

    def test_stale_breaker_evicted(self):
        """Breaker not accessed for > TTL should be evicted."""
        import time
        from utils.http_client import (
            _webhook_circuit_breakers,
            get_webhook_circuit_breaker,
            _evict_stale_circuit_breakers,
            _CIRCUIT_BREAKER_IDLE_TTL,
        )

        _webhook_circuit_breakers.clear()
        cb = get_webhook_circuit_breaker('https://stale.test/hook')
        # Backdate access time to exceed TTL
        cb._last_access_time = time.monotonic() - _CIRCUIT_BREAKER_IDLE_TTL - 1

        _evict_stale_circuit_breakers()
        assert 'https://stale.test/hook' not in _webhook_circuit_breakers
        _webhook_circuit_breakers.clear()

    def test_allow_request_updates_access_time(self):
        """allow_request() must update _last_access_time."""
        import time
        from utils.http_client import _webhook_circuit_breakers, get_webhook_circuit_breaker

        _webhook_circuit_breakers.clear()
        cb = get_webhook_circuit_breaker('https://test.test/hook')
        old_access = cb._last_access_time
        time.sleep(0.01)
        cb.allow_request()
        assert cb._last_access_time > old_access
        _webhook_circuit_breakers.clear()