"""Section 16 — MCP Elicitation: 7-layer test suite.

Covers gaps not addressed by the 43 tests in test_mcp_elicitation.py and the
14 tests in test_stress_elicitation_bypass.py.

New coverage:

  Layer 1 Unit:
    - build_form_elicitation: all available schema keys produce correct shape
    - build_url_elicitation: caller-supplied elicitation_id preserved
    - ElicitationRequest TypedDict has required fields (mode, message)
    - ElicitationAction TypedDict has required action field
    - AVAILABLE_PLATFORMS and AVAILABLE_DAW_CLOUDS are non-empty
    - compose_preferences schema has all expected properties

  Layer 2 Integration (session elicitation primitives):
    - create_pending_elicitation stores Future in session.pending
    - resolve_elicitation matching ID → Future set, returns True
    - resolve_elicitation non-matching ID → returns False
    - resolve_elicitation already-done Future → returns False
    - _signal_elicitation_complete resolves matching Future across sessions
    - _signal_elicitation_complete pushes SSE notification to queue
    - _signal_elicitation_complete with no matching session → returns 0
    - delete_session cancels all pending elicitation Futures

  Layer 3 E2E (HTTP UI elicitation routes):
    - GET /mcp/elicitation/{id}/callback?status=accepted → 200 HTML
    - GET /mcp/elicitation/{id}/callback?status=declined → 200 HTML
    - GET /mcp/connect/{invalid_slug} → 404
    - GET /mcp/connect/daw/{invalid_slug} → 404
    - GET /mcp/connect/{valid_slug} unauthenticated → 302 redirect
    - GET /mcp/connect/daw/{valid_slug} unauthenticated → 302 redirect

  Layer 4 Stress:
    - 50 pending elicitations in one session, all created without collision
    - _signal_elicitation_complete resolves across 5 simultaneous sessions

  Layer 5 Data Integrity:
    - Accepted content preserved in Future result
    - Declined action stored (action="decline", no content key in result)
    - resolve_elicitation stores exact result dict
    - build_form_elicitation requestedSchema matches SCHEMAS entry exactly

  Layer 6 Security:
    - elicitation_callback with unknown elicitation_id → 200 (no crash)
    - platform_connect_start unknown platform → 404 (not 500)
    - Double resolve → second call returns False (can't resolve twice)
    - execute_review_proposal_interactive with malformed args → graceful

  Layer 7 Performance:
    - 1000× build_form_elicitation under 50 ms
    - 1000× build_url_elicitation under 50 ms
    - 100 pending elicitations create+resolve under 100 ms
"""
from __future__ import annotations

import asyncio
import time
from typing import get_type_hints

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from unittest.mock import patch
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.muse_contracts.mcp_types import ElicitationAction, ElicitationRequest
from musehub.main import app
from musehub.mcp.elicitation import (
    AVAILABLE_DAW_CLOUDS,
    AVAILABLE_PLATFORMS,
    SCHEMAS,
    build_form_elicitation,
    build_url_elicitation,
    daw_cloud_connect_url,
    oauth_connect_url,
)
from musehub.mcp.session import (
    MCPSession,
    create_pending_elicitation,
    create_session,
    delete_session,
    push_to_session,
    resolve_elicitation,
)


# ── Fixtures ──────────────────────────────────────────────────────────────────


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


@pytest_asyncio.fixture
async def http_client(db_session: AsyncSession) -> AsyncClient:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://localhost",
    ) as c:
        yield c


# ── Layer 1 — Unit ────────────────────────────────────────────────────────────


class TestUnitSchemaBuilders:
    def test_build_form_all_schema_keys(self) -> None:
        """build_form_elicitation must succeed for every key in SCHEMAS."""
        for key in SCHEMAS:
            result = build_form_elicitation(key, message=f"Test: {key}")
            assert result["mode"] == "form", f"mode wrong for key={key}"
            assert result["message"] == f"Test: {key}"
            assert "requestedSchema" in result, f"missing requestedSchema for key={key}"

    def test_build_form_schema_matches_schemas_dict(self) -> None:
        """requestedSchema must be the exact SCHEMAS[key] dict."""
        for key in SCHEMAS:
            result = build_form_elicitation(key, message="m")
            assert result["requestedSchema"] is SCHEMAS[key]

    def test_build_url_caller_supplied_id_preserved(self) -> None:
        params, eid = build_url_elicitation("https://example.com", "Connect", elicitation_id="stable-id")
        assert eid == "stable-id"
        assert params["elicitationId"] == "stable-id"

    def test_build_url_auto_generates_id_when_omitted(self) -> None:
        params, eid1 = build_url_elicitation("https://a.com", "m1")
        params2, eid2 = build_url_elicitation("https://a.com", "m2")
        assert eid1 != eid2  # unique IDs
        assert len(eid1) > 8

    def test_build_url_mode_and_url_set(self) -> None:
        params, _ = build_url_elicitation("https://musehub.ai/connect", "msg")
        assert params["mode"] == "url"
        assert params["url"] == "https://musehub.ai/connect"
        assert params["message"] == "msg"


class TestUnitElicitationTypesDicts:
    def test_elicitation_action_has_action_field(self) -> None:
        action: ElicitationAction = {"action": "accept"}
        assert action["action"] == "accept"

    def test_elicitation_action_with_content(self) -> None:
        action: ElicitationAction = {"action": "accept", "content": {"key": "C major"}}
        assert action["content"]["key"] == "C major"

    def test_elicitation_request_form_shape(self) -> None:
        req: ElicitationRequest = {
            "mode": "form",
            "message": "Pick preferences",
            "requestedSchema": {"type": "object"},
        }
        assert req["mode"] == "form"
        assert "requestedSchema" in req

    def test_elicitation_request_url_shape(self) -> None:
        req: ElicitationRequest = {
            "mode": "url",
            "message": "Connect platform",
            "url": "https://musehub.ai/connect/spotify",
            "elicitationId": "eid-123",
        }
        assert req["url"] == "https://musehub.ai/connect/spotify"


class TestUnitAvailableLists:
    def test_available_platforms_non_empty(self) -> None:
        assert len(AVAILABLE_PLATFORMS) > 0

    def test_available_daw_clouds_non_empty(self) -> None:
        assert len(AVAILABLE_DAW_CLOUDS) > 0

    def test_platforms_are_strings(self) -> None:
        assert all(isinstance(p, str) for p in AVAILABLE_PLATFORMS)

    def test_daw_clouds_are_strings(self) -> None:
        assert all(isinstance(d, str) for d in AVAILABLE_DAW_CLOUDS)


class TestUnitComposePreferencesSchema:
    def test_schema_is_object_type(self) -> None:
        schema = SCHEMAS["compose_preferences"]
        assert schema["type"] == "object"

    def test_schema_has_key_property(self) -> None:
        props = SCHEMAS["compose_preferences"]["properties"]
        assert "key" in props

    def test_schema_has_tempo_bpm_property(self) -> None:
        props = SCHEMAS["compose_preferences"]["properties"]
        assert "tempo_bpm" in props

    def test_tempo_bpm_has_min_max_constraints(self) -> None:
        tempo = SCHEMAS["compose_preferences"]["properties"]["tempo_bpm"]
        assert tempo["minimum"] >= 1
        assert tempo["maximum"] <= 500


# ── Layer 2 — Integration ─────────────────────────────────────────────────────


class TestIntegrationPendingElicitation:
    @pytest.mark.anyio
    async def test_create_pending_stores_future(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            fut = create_pending_elicitation(s, "req-1")
            assert "req-1" in s.pending
            assert s.pending["req-1"] is fut
            assert not fut.done()
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_resolve_matching_id_returns_true(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            fut = create_pending_elicitation(s, "req-resolve")
            result = resolve_elicitation(s, "req-resolve", {"action": "accept", "content": {"key": "D minor"}})
            assert result is True
            assert fut.done()
            assert fut.result() == {"action": "accept", "content": {"key": "D minor"}}
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_resolve_non_matching_id_returns_false(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            create_pending_elicitation(s, "req-real")
            result = resolve_elicitation(s, "req-wrong", {"action": "accept"})
            assert result is False
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_resolve_already_done_future_returns_false(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            create_pending_elicitation(s, "req-done")
            resolve_elicitation(s, "req-done", {"action": "accept"})
            # Second resolve should fail.
            result = resolve_elicitation(s, "req-done", {"action": "accept"})
            assert result is False
        finally:
            delete_session(s.session_id)


class TestIntegrationSignalElicitation:
    @pytest.mark.anyio
    async def test_signal_resolves_matching_future(self) -> None:
        from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete

        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            fut = create_pending_elicitation(s, "sig-id-1")
            resolved = _signal_elicitation_complete("sig-id-1", action="accept")
            assert resolved >= 1
            assert fut.done()
            assert fut.result()["action"] == "accept"
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_signal_pushes_sse_notification_to_queue(self) -> None:
        from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete

        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            queue: asyncio.Queue[str | None] = asyncio.Queue()
            s.sse_queues.append(queue)
            create_pending_elicitation(s, "sig-sse-1")

            _signal_elicitation_complete("sig-sse-1", action="accept")

            item = queue.get_nowait()
            assert item is not None
            assert "notifications/elicitation/complete" in item
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_signal_no_matching_session_returns_zero(self) -> None:
        from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete

        # Signal an ID that no session has pending.
        resolved = _signal_elicitation_complete("completely-unknown-elicitation-id")
        assert resolved == 0

    @pytest.mark.anyio
    async def test_delete_session_cancels_pending_futures(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        sid = s.session_id
        fut1 = create_pending_elicitation(s, "cancel-1")
        fut2 = create_pending_elicitation(s, "cancel-2")

        delete_session(sid)

        assert fut1.cancelled()
        assert fut2.cancelled()


# ── Layer 3 — End-to-End ──────────────────────────────────────────────────────


class TestE2EElicitationCallbackRoute:
    @pytest.mark.anyio
    async def test_callback_accepted_returns_200(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await http_client.get(
            "/mcp/elicitation/test-eid-accepted/callback?status=accepted"
        )
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_callback_declined_returns_200(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await http_client.get(
            "/mcp/elicitation/test-eid-declined/callback?status=declined"
        )
        assert resp.status_code == 200


class TestE2EPlatformConnectRoute:
    @pytest.mark.anyio
    async def test_invalid_platform_slug_returns_404(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await http_client.get(
            "/mcp/connect/totally-made-up-platform?elicitation_id=eid-xyz"
        )
        assert resp.status_code == 404

    @pytest.mark.anyio
    async def test_unauthenticated_valid_platform_redirects(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        # Use first available platform slug.
        from musehub.api.routes.musehub.ui_mcp_elicitation import _PLATFORM_BY_SLUG
        if not _PLATFORM_BY_SLUG:
            pytest.skip("No platforms configured")
        slug = next(iter(_PLATFORM_BY_SLUG))
        # Patch _get_musehub_user_id to return None (unauthenticated) since the
        # test client has no SessionMiddleware installed.
        with patch(
            "musehub.api.routes.musehub.ui_mcp_elicitation._get_musehub_user_id",
            return_value=None,
        ):
            resp = await http_client.get(
                f"/mcp/connect/{slug}?elicitation_id=eid-auth-test",
                follow_redirects=False,
            )
        # Unauthenticated → redirect to login page.
        assert resp.status_code in (302, 303)
        assert "login" in resp.headers.get("location", "").lower()


class TestE2EDawConnectRoute:
    @pytest.mark.anyio
    async def test_invalid_daw_slug_returns_404(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await http_client.get(
            "/mcp/connect/daw/totally-unknown-daw?elicitation_id=eid-daw"
        )
        assert resp.status_code == 404

    @pytest.mark.anyio
    async def test_unauthenticated_valid_daw_redirects(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        from musehub.api.routes.musehub.ui_mcp_elicitation import _DAW_BY_SLUG
        if not _DAW_BY_SLUG:
            pytest.skip("No DAW services configured")
        slug = next(iter(_DAW_BY_SLUG))
        # Patch _get_musehub_user_id to return None (unauthenticated) since the
        # test client has no SessionMiddleware installed.
        with patch(
            "musehub.api.routes.musehub.ui_mcp_elicitation._get_musehub_user_id",
            return_value=None,
        ):
            resp = await http_client.get(
                f"/mcp/connect/daw/{slug}?elicitation_id=eid-daw-test",
                follow_redirects=False,
            )
        assert resp.status_code in (302, 303)
        assert "login" in resp.headers.get("location", "").lower()


# ── Layer 4 — Stress ──────────────────────────────────────────────────────────


class TestStressElicitation:
    @pytest.mark.anyio
    async def test_50_pending_elicitations_no_collision(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            futs = []
            for i in range(50):
                fut = create_pending_elicitation(s, f"elicit-{i}")
                futs.append((i, fut))
            # All 50 should be distinct, non-done futures.
            assert len(s.pending) == 50
            assert all(not fut.done() for _, fut in futs)
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_signal_resolves_across_5_sessions(self) -> None:
        from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete

        sessions = [create_session(None, {"elicitation": {"form": {}}}) for _ in range(5)]
        futs = []
        try:
            for s in sessions:
                fut = create_pending_elicitation(s, "shared-eid")
                futs.append(fut)

            resolved = _signal_elicitation_complete("shared-eid", action="accept")
            assert resolved == 5
            assert all(fut.done() for fut in futs)
        finally:
            for s in sessions:
                delete_session(s.session_id)


# ── Layer 5 — Data Integrity ──────────────────────────────────────────────────


class TestDataIntegrityElicitation:
    @pytest.mark.anyio
    async def test_accepted_content_preserved_in_result(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            fut = create_pending_elicitation(s, "di-accept")
            payload = {"action": "accept", "content": {"key": "G major", "tempo_bpm": 140}}
            resolve_elicitation(s, "di-accept", payload)
            assert fut.result() == payload
        finally:
            delete_session(s.session_id)

    @pytest.mark.anyio
    async def test_declined_action_in_result(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            fut = create_pending_elicitation(s, "di-decline")
            resolve_elicitation(s, "di-decline", {"action": "decline"})
            assert fut.result()["action"] == "decline"
            assert "content" not in fut.result()
        finally:
            delete_session(s.session_id)

    def test_build_form_requested_schema_is_schemas_entry(self) -> None:
        for key in list(SCHEMAS.keys())[:5]:  # check first 5 to keep test fast
            params = build_form_elicitation(key, message="m")
            assert params["requestedSchema"] == SCHEMAS[key]

    def test_build_url_elicitation_id_in_params(self) -> None:
        params, eid = build_url_elicitation("https://example.com", "msg")
        assert params["elicitationId"] == eid


# ── Layer 6 — Security ────────────────────────────────────────────────────────


class TestSecurityElicitation:
    @pytest.mark.anyio
    async def test_callback_unknown_id_does_not_crash(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """elicitation_callback with unknown ID must return 200, not 500."""
        resp = await http_client.get(
            "/mcp/elicitation/completely-unknown-id/callback?status=accepted"
        )
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_platform_unknown_slug_returns_404_not_500(
        self, http_client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await http_client.get("/mcp/connect/injected<script>?elicitation_id=x")
        # Must not be a 500; either 404 or 422 (validation).
        assert resp.status_code in (400, 404, 422)

    @pytest.mark.anyio
    async def test_double_resolve_returns_false(self) -> None:
        """Resolving the same elicitation twice must fail on the second call."""
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            create_pending_elicitation(s, "double-resolve")
            r1 = resolve_elicitation(s, "double-resolve", {"action": "accept"})
            r2 = resolve_elicitation(s, "double-resolve", {"action": "accept"})
            assert r1 is True
            assert r2 is False
        finally:
            delete_session(s.session_id)

# ── Layer 7 — Performance ─────────────────────────────────────────────────────


class TestPerformanceElicitation:
    def test_1000_build_form_under_50ms(self) -> None:
        keys = list(SCHEMAS.keys())
        start = time.perf_counter()
        for i in range(1000):
            build_form_elicitation(keys[i % len(keys)], message="perf test")
        elapsed_ms = (time.perf_counter() - start) * 1000
        assert elapsed_ms < 50, f"1000× build_form_elicitation took {elapsed_ms:.1f} ms"

    def test_1000_build_url_under_50ms(self) -> None:
        start = time.perf_counter()
        for i in range(1000):
            build_url_elicitation(f"https://example.com/flow/{i}", "Connect")
        elapsed_ms = (time.perf_counter() - start) * 1000
        assert elapsed_ms < 50, f"1000× build_url_elicitation took {elapsed_ms:.1f} ms"

    @pytest.mark.anyio
    async def test_100_pending_create_and_resolve_under_100ms(self) -> None:
        s = create_session(None, {"elicitation": {"form": {}}})
        try:
            start = time.perf_counter()
            for i in range(100):
                create_pending_elicitation(s, f"perf-{i}")
            for i in range(100):
                resolve_elicitation(s, f"perf-{i}", {"action": "accept"})
            elapsed_ms = (time.perf_counter() - start) * 1000
            assert elapsed_ms < 100, f"100 create+resolve took {elapsed_ms:.1f} ms"
        finally:
            delete_session(s.session_id)