"""Offline control-plane tests: no Docker, provider, LAN, or deployment access."""
import asyncio
import sqlite3
import time

import pytest
from fastapi.testclient import TestClient

from control.accounts import Accounts, password_matches, public_user
from control.app import Settings, create_app

PASSWORD = 'correct-horse-password'
SECRET = 'internal-secret-' + 'x' * 32


class FakeProvisioner:
    def __init__(self):
        self.ensured = []
        self.removed = []
        self.fail_ensure = False
        self.fail_remove = False
        self.healthy = True

    async def ensure(self, user):
        self.ensured.append(user['id'])
        if self.fail_ensure:
            raise RuntimeError('sensitive-provider-key-must-not-leak')
        return {'container_name': user['container_name'], 'hermes_url': 'http://127.0.0.1:1234', 'hermes_api_key': 'private-hermes-key'}

    async def ready(self, user):
        return self.healthy

    async def remove(self, user):
        self.removed.append(user['id'])
        if self.fail_remove:
            raise RuntimeError('container ownership verification failed')


@pytest.fixture
def environment(tmp_path):
    fake = FakeProvisioner()
    settings = Settings(db_path=str(tmp_path / 'accounts.db'), admin_password=PASSWORD, service_secret=SECRET, allowed_origins=('https://testserver',), reconcile_interval=3600)
    app = create_app(settings, fake)
    with TestClient(app, base_url='https://testserver') as client:
        yield client, app, fake, settings


def login(client, username='admin', password=PASSWORD):
    result = client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': username, 'password': password})
    assert result.status_code == 200, result.text
    return result.json()


def auth(result):
    return {'Authorization': 'Bearer ' + result['token']}


def wait_state(client, headers, user_id, status):
    # Requests yield to the ASGI lifecycle loop; bounded wait, no network.
    deadline = time.monotonic() + 3
    while time.monotonic() < deadline:
        users = client.get('/api/admin/users', headers=headers).json()['users']
        found = next((u for u in users if u['id'] == user_id), None)
        if (status is None and found is None) or (found and found['status'] == status):
            return found
    raise AssertionError(f'Account did not reach {status}: {found}')


def create(client, headers, username='john-doe'):
    response = client.post('/api/admin/users', headers=headers, json={'username': username, 'display_name': 'John Doe', 'password': PASSWORD})
    assert response.status_code == 201, response.text
    return response.json()['user']


def test_bootstrap_required(tmp_path):
    app = create_app(Settings(db_path=str(tmp_path / 'missing.db'), admin_password=None), FakeProvisioner())
    with pytest.raises(RuntimeError, match='CONTROL_ADMIN_PASSWORD'):
        with TestClient(app):
            pass


def test_password_sessions_cookie_schema(environment):
    client, app, fake, settings = environment
    assert client.get('/api/auth/me').status_code == 401
    session = login(client)
    assert set(session['user']) == {'id', 'username', 'display_name', 'role', 'status', 'container_name', 'error'}
    assert len(session['user']['id']) == 32
    assert session['expires_at'] > time.time()
    cookie = client.cookies.get('hope_session')
    assert cookie == session['token']
    raw = app.state.accounts.get(session['user']['id'])
    assert raw['password_hash'] != PASSWORD
    assert password_matches(PASSWORD, raw['password_hash'])
    row = app.state.accounts.db.execute('SELECT hash FROM sessions').fetchone()
    assert row['hash'] != session['token']
    assert client.get('/api/auth/me').status_code == 200
    assert client.post('/api/auth/logout').status_code == 403
    assert client.post('/api/auth/logout', headers={'Origin': 'https://evil.example'}).status_code == 403
    assert client.post('/api/auth/logout', headers=auth(session)).status_code == 200
    assert client.get('/api/auth/me', headers=auth(session)).status_code == 401
    again = client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'admin', 'password': PASSWORD})
    cookie_header = again.headers['set-cookie'].lower()
    assert all(x in cookie_header for x in ('httponly', 'secure', 'samesite=strict'))
    assert client.post('/api/auth/logout', headers={'Origin': 'https://testserver'}).status_code == 200


def test_create_auth_reset_and_validation(environment):
    client, app, fake, settings = environment
    admin = auth(login(client))
    user = create(client, admin)
    ready = wait_state(client, admin, user['id'], 'ready')
    assert ready['container_name'] == f"hope-hermes-john-doe-{user['id'][:8]}"
    assert client.post('/api/admin/users', headers=admin, json={'username': 'Bad/Name', 'display_name': 'Bad', 'password': PASSWORD}).status_code == 400
    assert client.post('/api/admin/users', headers=admin, json={'username': 'john-doe', 'display_name': 'John', 'password': PASSWORD}).status_code == 409
    ordinary = auth(login(client, 'john-doe'))
    assert client.get('/api/admin/users', headers=ordinary).status_code == 403
    assert client.get('/internal/users/' + user['id'], headers=ordinary).status_code == 401
    assert client.post('/api/admin/users/' + user['id'] + '/password', headers=admin, json={'password': 'new-safe-password'}).status_code == 200
    assert client.get('/api/auth/me', headers=ordinary).status_code == 401
    assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'john-doe', 'password': PASSWORD}).status_code == 401
    login(client, 'john-doe', 'new-safe-password')


def test_last_admin_and_delete_revokes_sessions_rooms(environment):
    client, app, fake, settings = environment
    session = login(client)
    admin = auth(session)
    assert client.delete('/api/admin/users/' + session['user']['id'], headers=admin).status_code == 409
    user = create(client, admin)
    wait_state(client, admin, user['id'], 'ready')
    ordinary = login(client, 'john-doe')
    internal = {'X-Control-Secret': SECRET}
    resolve = client.post('/internal/resolve-session', headers=internal, json={'token': ordinary['token']})
    assert resolve.status_code == 200
    assert resolve.json()['hermes_api_key'] == 'private-hermes-key'
    assert 'password_hash' not in resolve.text
    assert client.post('/internal/rooms', headers=internal, json={'user_id': user['id'], 'room_name': 'voice-test', 'session_token': ordinary['token']}).status_code == 201
    assert client.get('/internal/rooms/voice-test', headers=internal).status_code == 200
    fake.fail_remove = True
    assert client.delete('/api/admin/users/' + user['id'], headers=admin).status_code == 202
    assert client.get('/api/auth/me', headers=auth(ordinary)).status_code == 401
    assert client.get('/internal/rooms/voice-test', headers=internal).status_code == 404
    assert client.post('/internal/resolve-session', headers=internal, json={'token': ordinary['token']}).status_code == 401
    wait_state(client, admin, user['id'], 'deleting')
    fake.fail_remove = False
    # Wait for failed first operation to finish before retry scheduling.
    async def settled():
        await asyncio.gather(*app.state.lifecycle.tasks.values())
    client.portal.call(settled)
    client.post('/api/admin/users/' + user['id'] + '/retry', headers=admin)
    wait_state(client, admin, user['id'], None)
    assert user['id'] in fake.removed


def test_failure_visible_and_restart_reconcile(tmp_path):
    fake = FakeProvisioner()
    fake.fail_ensure = True
    settings = Settings(db_path=str(tmp_path / 'restart.db'), admin_password=PASSWORD, service_secret=SECRET, reconcile_interval=3600)
    with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
        session = login(client)
        header = auth(session)
        user = wait_state(client, header, session['user']['id'], 'error')
        assert 'sensitive' not in user['error']
    fake.fail_ensure = False
    settings.admin_password = None
    with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
        wait_state(client, header, session['user']['id'], 'ready')
        assert client.get('/api/auth/me', headers=header).status_code == 200


def test_expiry_throttle_and_bounded_input(environment):
    client, app, fake, settings = environment
    session = login(client)
    with app.state.accounts.lock, app.state.accounts.db:
        app.state.accounts.db.execute('UPDATE sessions SET expires_at=0')
    assert client.get('/api/auth/me', headers=auth(session)).status_code == 401
    client.cookies.clear()
    assert client.post('/api/auth/login', json={'username': 'admin', 'password': 'p' * 257}).status_code == 400
    assert client.post('/api/auth/login', content='x' * 9000, headers={'Content-Type': 'application/json'}).status_code == 413
    assert client.post('/api/auth/login', content='{}', headers={'Content-Type': 'text/plain'}).status_code == 415
    client.cookies.clear()
    for _ in range(settings.login_limit):
        result = client.post('/api/auth/login', json={'username': 'missing', 'password': PASSWORD})
    assert result.status_code == 401
    assert client.post('/api/auth/login', json={'username': 'missing', 'password': PASSWORD}).status_code == 429


def test_internal_fail_closed(environment):
    client, app, fake, settings = environment
    assert client.post('/internal/resolve-session', json={'token': 'x' * 32}).status_code == 401
    assert client.get('/internal/users/' + 'a' * 32, headers={'X-Control-Secret': 'wrong'}).status_code == 401
    settings.service_secret = None
    assert client.get('/internal/rooms/voice-test', headers={'X-Control-Secret': SECRET}).status_code == 401


def test_room_session_binding_and_logout(environment):
    client, app, fake, settings = environment
    admin_session = login(client)
    header = auth(admin_session)
    wait_state(client, header, admin_session['user']['id'], 'ready')
    user = create(client, header, 'alice')
    wait_state(client, header, user['id'], 'ready')
    session = login(client, 'alice')
    internal = {'X-Control-Secret': SECRET}
    mismatch = {'user_id': user['id'], 'room_name': 'wrong', 'session_token': admin_session['token']}
    assert client.post('/internal/rooms', headers=internal, json=mismatch).status_code == 409
    grant = {'user_id': user['id'], 'room_name': 'voice-alice', 'session_token': session['token']}
    assert client.post('/internal/rooms', headers=internal, json=grant).status_code == 201
    assert client.get('/internal/rooms/voice-alice', headers=internal).status_code == 200
    assert client.post('/api/auth/logout', headers=auth(session)).status_code == 200
    assert client.get('/internal/rooms/voice-alice', headers=internal).status_code == 404


def test_failed_removal_restart_retries(tmp_path):
    path = str(tmp_path / 'removal.db')
    store = Accounts(path)
    store.create('admin', 'Administrator', 'admin', PASSWORD)
    user = store.create('alice', 'Alice', 'user', PASSWORD)
    store.begin_delete(user['id'])
    store.close()
    settings = Settings(db_path=path, admin_password=None, service_secret=SECRET, reconcile_interval=3600)
    fake = FakeProvisioner()
    fake.fail_remove = True
    with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
        header = auth(login(client))
        wait_state(client, header, user['id'], 'deleting')
        assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'alice', 'password': PASSWORD}).status_code == 401
    fake.fail_remove = False
    with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
        header = auth(login(client))
        wait_state(client, header, user['id'], None)
    assert fake.removed.count(user['id']) >= 2


def test_reset_between_password_verification_and_session_issue(environment, monkeypatch):
    client, app, fake, settings = environment
    original = app.state.accounts.issue_session

    def reset_then_issue(user_id, ttl, expected_hash):
        app.state.accounts.reset_password(user_id, 'changed-safe-password')
        return original(user_id, ttl, expected_hash)

    monkeypatch.setattr(app.state.accounts, 'issue_session', reset_then_issue)
    response = client.post('/api/auth/login', json={'username': 'admin', 'password': PASSWORD})
    assert response.status_code == 401
    assert app.state.accounts.db.execute('SELECT count(*) FROM sessions').fetchone()[0] == 0
    monkeypatch.setattr(app.state.accounts, 'issue_session', original)
    login(client, password='changed-safe-password')


def test_successes_and_other_accounts_do_not_share_failure_budget(environment):
    client, app, fake, settings = environment
    for _ in range(settings.login_limit + 2):
        login(client)
    for _ in range(settings.login_limit):
        assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'missing', 'password': PASSWORD}).status_code == 401
    assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'missing', 'password': PASSWORD}).status_code == 429
    login(client)  # Same IP, different account remains usable.


@pytest.mark.parametrize('peer,trusted', [('127.0.0.1', True), ('::1', True), ('203.0.113.10', False)])
def test_real_ip_only_trusted_from_loopback(tmp_path, peer, trusted):
    settings = Settings(db_path=str(tmp_path / 'proxy.db'), admin_password=PASSWORD, login_limit=2)
    with TestClient(create_app(settings, FakeProvisioner()), base_url='https://testserver', client=(peer, 1234)) as client:
        body = {'username': 'missing', 'password': PASSWORD}
        for _ in range(settings.login_limit):
            assert client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.1'}, json=body).status_code == 401
        assert client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.1'}, json=body).status_code == 429
        other = client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.2', 'X-Forwarded-For': '198.51.100.3'}, json=body)
        assert other.status_code == (401 if trusted else 429)


def test_database_delete_intent_persists(tmp_path):
    db = str(tmp_path / 'durable.db')
    store = Accounts(db)
    admin = store.create('admin', 'Administrator', 'admin', PASSWORD)
    user = store.create('alice', 'Alice', 'user', PASSWORD)
    token, expiry = store.issue_session(user['id'], 60, user['password_hash'])
    store.begin_delete(user['id'])
    store.close()
    store = Accounts(db)
    assert store.get(user['id'])['desired'] == 'deleted'
    assert store.resolve_session(token) is None
    assert store.get(admin['id'])['role'] == 'admin'
    store.close()