import asyncio
import base64
import json
import threading
import time

import pytest

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

from hermes_mcp.bridge import (Activity, BearerAuth, FacesBuffer, LatestFrame, build_app, build_mcp,  # noqa: E402
                               HealthBuffer, people_summary)

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_have_no_head_tools_without_a_head_and_keep_their_names_with_one():
    base = {'look', 'faces', 'list_people'}
    assert tool_names(build_mcp(LatestFrame())) == base
    assert tool_names(build_mcp(LatestFrame(), head=FakeHead())) == base | {'turn_head', 'head_position'}
    assert 'enroll' not in tool_names(build_mcp(LatestFrame(), head=FakeHead()))   # enrolment is the owner's step


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', {}))
    text, image = result.content
    assert announced == [1]
    assert text.type == 'text' and json.loads(text.text)['faces']['available'] is False   # no detector running here
    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']


def face(track_id, x, name='', sim=0.0, conf=0.9, w=300, h=400):
    return {'track_id': track_id, 'bbox': {'x': x, 'y': 100, 'w': w, 'h': h}, 'confidence': conf, 'name': name, 'similarity': sim}


def call(mcp, name, args=None):
    return asyncio.run(mcp.call_tool(name, args or {}))


def test_faces_reports_identity_null_for_a_stranger_and_never_makes_up_a_name():
    buf = FacesBuffer()
    buf.put([face(1, 100, 'Kyle', 0.82), face(2, 600)])
    out = call(build_mcp(LatestFrame(), faces_buf=buf), 'faces').structured_content
    assert out['available'] and out['count'] == 2 and out['image_size'] == {'w': 1296, 'h': 972}
    kyle, stranger = out['faces']
    assert kyle['identity'] == {'name': 'Kyle', 'similarity': 0.82}
    assert stranger['identity'] is None                       # unknown, not guessed
    assert set(kyle) == {'track_id', 'bbox', 'bbox_norm', 'confidence', 'identity', 'seen_for_s'}
    assert 'embedding' not in json.dumps(out)


def test_track_ids_persist_and_seen_for_grows():
    buf = FacesBuffer()
    buf.put([face(7, 100)], t=time.monotonic() - 3.0)
    buf.put([face(7, 105)])
    (f,) = buf.summary()['faces']
    assert f['track_id'] == 7 and 2.5 < f['seen_for_s'] < 3.6


def test_faces_says_when_detection_is_not_running_instead_of_reporting_an_empty_room():
    out = call(build_mcp(LatestFrame(), faces_buf=FacesBuffer()), 'faces').structured_content
    assert out['available'] is False and 'reason' in out
    buf = FacesBuffer()
    buf.put([face(1, 100)], t=time.monotonic() - 30)           # stale data is not "now"
    assert buf.summary(max_age_s=1.5)['available'] is False


def test_an_empty_room_is_available_with_zero_faces():
    buf = FacesBuffer()
    buf.put([])
    assert buf.summary()['available'] and buf.summary()['count'] == 0


def test_look_can_skip_the_photo():
    frames, buf = LatestFrame(), FacesBuffer()
    buf.put([face(1, 100, 'Kyle', 0.8)])
    frames.put(b'\xff\xd8x')
    result = call(build_mcp(frames, faces_buf=buf), 'look', {'send_image': 'never'})
    assert len(result.content) == 1 and result.content[0].type == 'text'
    assert json.loads(result.content[0].text)['faces']['faces'][0]['identity']['name'] == 'Kyle'


def test_list_people_gives_names_and_counts_never_face_data(tmp_path):
    path = tmp_path / 'faces.json'
    path.write_text(json.dumps({'Kyle': [[0.1] * 128] * 10, 'Alex': [[0.2] * 128] * 3}))
    out = call(build_mcp(LatestFrame(), people_path=str(path)), 'list_people').structured_content
    assert out == {'count': 2, 'people': [{'name': 'Alex', 'samples': 3}, {'name': 'Kyle', 'samples': 10}]}
    assert people_summary(str(tmp_path / 'missing.json')) == {'count': 0, 'people': []}


def test_the_server_can_be_discovered_resources_and_prompts_are_not_empty():
    mcp = build_mcp(LatestFrame(), hardware={'speakers': False})
    assert [str(r.uri) for r in asyncio.run(mcp.list_resources())] == ['robot://status']
    assert {p.name for p in asyncio.run(mcp.list_prompts())} == {'who_is_here', 'look_around'}
    status = json.loads(asyncio.run(mcp.read_resource('robot://status'))[0].content)
    assert status['hardware'] == {'camera': True, 'head': False, 'speakers': False}
    assert status['detects'] == ['faces'] and 'people' in status['not_detected']
    assert 'embeddings never leave it' in status['recognition']['registry']
    assert 'turn_head' not in status['tools']


def test_status_reports_health_when_known_and_says_when_it_is_not():
    h = HealthBuffer()
    mcp = build_mcp(LatestFrame(), health=h)
    unknown = json.loads(asyncio.run(mcp.read_resource('robot://status'))[0].content)
    assert unknown['health'] == {'available': False}
    h.update(temperature_c=63.4, thermal_level='ok', throttled={'available': True})
    known = json.loads(asyncio.run(mcp.read_resource('robot://status'))[0].content)
    assert known['health']['temperature_c'] == 63.4 and known['health']['thermal_level'] == 'ok'