"""Targeted WSS ownership, token/session reuse and late-response guards."""
import asyncio
import json
from pathlib import Path
from types import SimpleNamespace
import threading

import pytest
import httpx
from fastapi import FastAPI
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect

from control_sidecar.auth import Auth
from control_sidecar.main import create_app
from control_sidecar.protocol import BoundaryError
from control_sidecar.websocket import ControlSocket
from control_sidecar.transport import Transport
from app.secure_control import build_router
from test_secure_bridge import configuration, FakeIdentity, FakeTransport, MANAGER, CONTROL, AI_TOKEN, login_response


def cfg(**kwargs):
    return configuration(login_body=dict(advertisedIp='127.0.0.1', controlPort=9443,
        redis=dict(host='localhost',port=6379,db=0,username='reader',password='secret-fixture',tls=False),
        bboxStreamKey='fixture',connectionConfigVersion=1), **kwargs)


def test_wss_bootstrap_redaction_unknown_action_and_single_session():
    tx = FakeTransport(body={'status':'READY'})
    auth = FakeIdentity()
    app = create_app(cfg(), tx, auth)
    with TestClient(app) as client:
        with pytest.raises(WebSocketDisconnect):
            with client.websocket_connect('/ws/v1'):
                pass
        with client.websocket_connect('/ws/v1', headers={'Authorization':'Bearer '+MANAGER}) as ws:
            for i,action in enumerate(('control.server.status','tokens.flushCache','ai.runtimeStatus.get','ai.connectionInfo.get')):
                ws.send_json(dict(requestId=str(i),action=action,params={}))
                r=ws.receive_json()
                assert r['ok'] and r['requestId']==str(i)
                assert 'secret-fixture' not in json.dumps(r)
            assert not auth.invalidated
            with pytest.raises(WebSocketDisconnect):
                with client.websocket_connect('/ws/v1', headers={'Authorization':'Bearer '+MANAGER}):
                    pass
            ws.send_json(dict(requestId='x',action='ai.configuration.apply',params={}))
            r=ws.receive_json()
            assert not r['ok'] and r['error']['code']=='UNKNOWN_ACTION'
    assert app.state.sockets.active is None and auth.closed and tx.closed


def test_wss_runtime_unavailable_and_bounded_heartbeat(monkeypatch):
    monkeypatch.setattr(ControlSocket, 'interval', .02)
    app=create_app(cfg(),FakeTransport(status=503),FakeIdentity())
    with TestClient(app) as client:
        with client.websocket_connect('/ws/v1',headers={'Authorization':'Bearer '+MANAGER}) as ws:
            ws.send_json(dict(requestId='r',action='ai.runtimeStatus.get',params={}))
            result=ws.receive_json()
            assert result['result']['status']=='NOT_READY'
            ping=ws.receive_json()
            ws.send_json({'type':'pong','pingId':ping['pingId']})
            for _ in range(3):
                assert ws.receive_json()['type']=='ping'
            with pytest.raises(WebSocketDisconnect):
                ws.receive_json()
    assert app.state.sockets.stats['pongs']==1
    assert app.state.sockets.last_error=='HEARTBEAT_TIMEOUT'


def test_auth_singleflight_restore_and_late_401(tmp_path):
    async def run():
        c=cfg(session_file=str(tmp_path/'session.json'))
        tx=FakeTransport(body=login_response())
        a=Auth(c,tx)
        await asyncio.gather(a.login(),a.login())
        assert len(tx.calls)==1 and Path(c.session_file).stat().st_mode & 0o077 == 0
        b=Auth(c,tx)
        assert b.restore() and b.current()==AI_TOKEN
        b.token='new-fixture-token-1234567890'
        b.invalidate(AI_TOKEN)
        assert b.token=='new-fixture-token-1234567890'
        b.invalidate(b.token)
        assert b.token is None and not Path(c.session_file).exists()
        a.failed(BoundaryError('LOGIN_TEMPORARY',500))
        for _ in range(8): a.failed(BoundaryError('UPSTREAM_OUTCOME_UNKNOWN'))
        assert not a.terminal
        a.failed(BoundaryError('UPSTREAM_TLS_INVALID'))
        assert a.terminal
    asyncio.run(run())


def test_runtime_snapshot_auth_and_parent_only():
    state=dict(observed_state='running',desired_state='running',run_id='fixture-run',run_generation=1,last_error=None)
    group=SimpleNamespace(spec={'source_id':'2'},stream_id='fixture-stream',
                          lifecycle=SimpleNamespace(snapshot=lambda:dict(state)))
    manager=SimpleNamespace(_lock=threading.RLock(),_groups={'fixture-stream':group})
    app=FastAPI()
    app.include_router(build_router(manager,CONTROL,'AI-test'))
    with TestClient(app) as client:
        assert client.get('/internal/secure-control/runtime-status').status_code==401
        r=client.get('/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+CONTROL}).json()
        assert r['streamCounts']['running']==1 and r['streams'][0]['cameraId']=='2'
        state['observed_state']='stopping'
        r=client.get('/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+CONTROL}).json()
        assert r['streams'][0]['lastError']=='TRANSITION_STOPPING'


def test_retry_after_is_login_only_event_response_is_unchanged():
    async def run():
        client=httpx.AsyncClient(transport=httpx.MockTransport(lambda req:
            httpx.Response(503,headers={'Retry-After':'90'},content=b'{"original":true}')))
        tx=Transport(None,client=client)
        assert await tx.request('POST','https://fixture.invalid') == (503,b'{"original":true}')
        with pytest.raises(BoundaryError) as e:
            await tx.request('POST','https://fixture.invalid',retry_after_error=True)
        assert e.value.retry_after==90
        await tx.close()
    asyncio.run(run())


def test_session_writer_does_not_delete_another_writer(tmp_path):
    c=cfg(session_file=str(tmp_path/'session.json'))
    pending=tmp_path/'session.json.pending'
    pending.write_text('other-writer')
    a=Auth(c,FakeTransport())
    a.token=AI_TOKEN
    with pytest.raises(FileExistsError):
        a.save_session()
    assert pending.read_text()=='other-writer'