import asyncio
import base64
import threading
import time

import pytest

pytest.importorskip('mcp')
from starlette.testclient import TestClient  # noqa: E402

from hermes_mcp.bridge import Activity, BearerAuth, LatestFrame, build_app, build_mcp  # noqa: E402

TOKEN = 'x' * 32


class FakeHead:
    def move(self, pan, tilt):
        return max(-80, min(80, pan)), max(-45, min(45, tilt))

    def position(self):
        return 0.0, 0.0


def tool_names(mcp):
    return {t.name for t in asyncio.run(mcp.list_tools())}


def test_tools_are_look_only_without_a_head_and_keep_their_names_with_one():
    assert tool_names(build_mcp(LatestFrame())) == {'look'}
    assert tool_names(build_mcp(LatestFrame(), head=FakeHead())) == {'look', 'turn_head', 'head_position'}


def test_look_returns_the_next_frame_not_a_stale_one():
    frames = LatestFrame()
    frames.put(b'stale', time.monotonic() - 5)
    announced = []
    mcp = build_mcp(frames, announce_look=lambda: announced.append(1), look_timeout_s=2.0)
    threading.Timer(0.3, lambda: frames.put(b'\xff\xd8fresh')).start()
    result = asyncio.run(mcp.call_tool('look', {}))
    image = result.content[0]
    assert announced == [1]
    assert image.type == 'image' and image.mime_type == 'image/jpeg'
    assert base64.b64decode(image.data) == b'\xff\xd8fresh'       # the frame that arrived after the call


def test_look_raises_when_the_camera_is_silent():
    mcp = build_mcp(LatestFrame(), look_timeout_s=0.3)
    with pytest.raises(Exception) as err:
        asyncio.run(mcp.call_tool('look', {}))
    assert 'camera' in str(err.value.__cause__)                  # the SDK wraps it; the cause is ours


def test_wait_newer_semantics():
    f = LatestFrame()
    t0 = time.monotonic()
    assert f.wait_newer(t0, 0.1) is None                    # nothing yet
    f.put(b'a', t0 - 1)
    assert f.wait_newer(t0, 0.1) is None                    # older than t0
    threading.Timer(0.1, lambda: f.put(b'b')).start()
    assert f.wait_newer(t0, 1.0) == b'b'


def test_auth_rejects_missing_and_wrong_tokens_and_counts_only_good_requests():
    app, activity = build_app(build_mcp(LatestFrame()), TOKEN, '127.0.0.1')
    with TestClient(app) as client:
        assert client.get('/mcp').status_code == 401
        assert client.get('/mcp', headers={'Authorization': 'Bearer wrong'}).status_code == 401
        assert not activity.connected(time.monotonic(), 30.0)       # rejected requests are not a client
        r = client.get('/mcp', headers={'Authorization': f'Bearer {TOKEN}'})
        assert r.status_code != 401
        assert activity.connected(time.monotonic(), 30.0)           # an authenticated one is
        assert not activity.connected(time.monotonic() + 31.0, 30.0)   # ...until the grace runs out


def test_activity_counts_inflight_requests_as_connected():
    async def run():
        release = asyncio.Event()

        async def slow_app(scope, receive, send):
            await release.wait()

        act = Activity(slow_app)
        task = asyncio.create_task(act({'type': 'http'}, None, None))
        await asyncio.sleep(0.05)
        assert act.inflight == 1 and act.connected(time.monotonic() + 1000, 1.0)   # a held stream stays connected
        release.set()
        await task
        assert act.inflight == 0
    asyncio.run(run())


def test_bearer_auth_passes_lifespan_through():
    seen = []

    async def inner(scope, receive, send):
        seen.append(scope['type'])

    asyncio.run(BearerAuth(inner, TOKEN)({'type': 'lifespan'}, None, None))
    assert seen == ['lifespan']