"""Offline advanced API isolation, quotas, revocation and version tests."""
import asyncio
import os
from types import SimpleNamespace

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient

from control.accounts import Accounts
from control.app import Settings
from control.chat_features import FeatureLimits, FeatureProvider, FeatureStore, create_feature_router, local_capabilities


@pytest.fixture
def env(tmp_path):
    accounts = Accounts(str(tmp_path / 'accounts.db'))
    users = [accounts.create(name, name, 'user', 'correct-horse-password') for name in ('alice', 'bob')]
    for user in users:
        accounts.update(user['id'], status='ready')
    tokens = [accounts.issue_session(u['id'], 3600, u['password_hash'])[0] for u in users]
    limits = FeatureLimits(upload_bytes=128, account_bytes=180, disk_bytes=256, artifact_bytes=128, artifact_count=3, artifact_versions=3)
    store = FeatureStore(accounts, tmp_path / 'private', limits)
    app = FastAPI()
    app.state.accounts = accounts
    app.state.settings = Settings(allowed_origins=('https://testserver',))
    app.include_router(create_feature_router(store))
    with TestClient(app, base_url='https://testserver') as client:
        yield SimpleNamespace(accounts=accounts, store=store, client=client, users=users, tokens=tokens, headers=[{'Authorization': 'Bearer ' + t} for t in tokens])
    accounts.close()


def upload(env, data=b'hello', mime='text/plain', filename='note.txt', account=0):
    return env.client.post('/api/attachments', content=data, headers={**env.headers[account], 'Content-Type': mime, 'X-Filename': filename})


def artifact(env):
    result = env.client.post('/api/artifacts', headers=env.headers[0], json={'title': 'Draft', 'kind': 'markdown', 'content': '# Hello'})
    assert result.status_code == 201, result.text
    return result.json()['artifact']


def test_upload_download_private_and_owned(env):
    result = upload(env)
    assert result.status_code == 201
    item = result.json()['attachment']
    assert set(item) == {'id', 'filename', 'mime', 'size', 'extraction_status', 'created_at'}
    assert item['extraction_status'] == 'ready'
    path = env.store.root / item['id']
    assert os.stat(path).st_mode & 0o777 == 0o600
    assert os.stat(env.store.root).st_mode & 0o777 == 0o700
    url = '/api/attachments/' + item['id']
    response = env.client.get(url + '/content', headers=env.headers[0])
    assert response.content == b'hello'
    assert response.headers['cache-control'] == 'no-store'
    for method, suffix in (('get', ''), ('get', '/content'), ('delete', '')):
        assert getattr(env.client, method)(url + suffix, headers=env.headers[1]).status_code == 404
    assert env.client.get('/api/attachments', headers=env.headers[1]).json()['attachments'] == []
    assert env.client.get(url).status_code == 401
    content = asyncio.run(env.store.provider_content(env.users[0]['id'], [item['id']]))
    assert content == [{'type': 'text', 'text': 'Attachment: note.txt\nhello'}]
    with pytest.raises(HTTPException) as error:
        asyncio.run(env.store.provider_content(env.users[1]['id'], [item['id']]))
    assert error.value.status_code == 404


@pytest.mark.parametrize('data,mime,filename,status', [
    (b'hello', 'text/html', 'a.html', 415),
    (b'hello', 'image/png', 'a.png', 415),
    (b'hello', 'application/pdf', 'a.pdf', 415),
    (b'RIFFbad', 'image/webp', 'a.webp', 415),
    (b'\xff', 'text/plain', 'a.txt', 415),
    (b'\x00', 'text/plain', 'a.txt', 415),
    (b'hello', 'text/plain', '../escape', 422),
    (b'hello', 'text/plain', 'dir\\escape', 422),
    (b'x' * 129, 'text/plain', 'a.txt', 413),
    (b'', 'text/plain', 'a.txt', 413),
])
def test_mime_bytes_paths(env, data, mime, filename, status):
    assert upload(env, data, mime, filename).status_code == status
    assert list(env.store.root.iterdir()) == []


def test_quotas_delete_tombstones_and_reuse(env):
    a = upload(env, b'x' * 100).json()['attachment']
    assert upload(env, b'x' * 90).status_code == 413
    assert upload(env, b'y' * 128, account=1).status_code == 201
    assert upload(env, b'z' * 29, account=1).status_code == 413
    assert env.client.delete('/api/attachments/' + a['id'], headers=env.headers[0]).status_code == 200
    assert not (env.store.root / a['id']).exists()
    assert env.client.get('/api/attachments/' + a['id'], headers=env.headers[0]).status_code == 404
    changes = env.client.get('/api/features/sync', headers=env.headers[0]).json()
    assert [c['deleted'] for c in changes['changes']] == [False, True]
    assert len(env.client.get('/api/features/sync', headers=env.headers[1]).json()['changes']) == 1
    assert upload(env, b'x' * 100).status_code == 201


def test_symlink_denial(env, tmp_path):
    item = upload(env).json()['attachment']
    path = env.store.root / item['id']
    outside = tmp_path / 'outside'
    outside.write_bytes(b'private')
    path.unlink()
    path.symlink_to(outside)
    assert env.client.get('/api/attachments/' + item['id'] + '/content', headers=env.headers[0]).status_code == 404
    assert env.client.delete('/api/attachments/' + item['id'], headers=env.headers[0]).status_code == 404
    assert outside.read_bytes() == b'private'


def test_image_payload_requires_explicit_vision(env):
    image = b'\x89PNG\r\n\x1a\n' + b'fake-offline-image'
    item = upload(env, image, 'image/png', 'a.png').json()['attachment']
    with pytest.raises(HTTPException) as error:
        asyncio.run(env.store.provider_content(env.users[0]['id'], [item['id']]))
    assert error.value.status_code == 409
    parts = asyncio.run(env.store.provider_content(env.users[0]['id'], [item['id']], vision=True))
    assert parts[0]['image_url']['url'].startswith('data:image/png;base64,')
    assert not local_capabilities(env.store)['image_input']


def test_artifact_version_conflict_export_and_tombstones(env):
    item = artifact(env)
    url = '/api/artifacts/' + item['id']
    assert env.client.patch(url, headers=env.headers[0], json={'revision': 0, 'content': 'No'}).status_code == 409
    response = env.client.patch(url, headers=env.headers[0], json={'revision': 1, 'content': 'Second'})
    assert response.status_code == 200
    assert response.json()['artifact']['revision'] == 2
    assert env.client.get(url + '?revision=1', headers=env.headers[0]).json()['artifact']['content'] == '# Hello'
    assert env.client.get(url + '/export?revision=1', headers=env.headers[0]).content == b'# Hello'
    assert env.client.get(url + '/export', headers=env.headers[0]).content == b'Second'
    for method, suffix in (('get', ''), ('get', '/export'), ('delete', '')):
        assert getattr(env.client, method)(url + suffix, headers=env.headers[1]).status_code == 404
    assert env.client.patch(url, headers=env.headers[1], json={'revision': 2, 'content': 'theft'}).status_code == 404
    assert env.client.delete(url, headers=env.headers[0]).status_code == 200
    assert env.client.get(url + '?revision=1', headers=env.headers[0]).status_code == 404
    assert env.client.get('/api/features/sync', headers=env.headers[0]).json()['changes'][-1]['deleted']


def test_revocation_and_delete_fk(env):
    attachment = upload(env).json()['attachment']
    item = artifact(env)
    env.accounts.revoke(env.tokens[0])
    assert env.client.get('/api/attachments/' + attachment['id'] + '/content', headers=env.headers[0]).status_code == 401
    assert env.client.get('/api/artifacts/' + item['id'], headers=env.headers[0]).status_code == 401
    env.accounts.begin_delete(env.users[0]['id'])
    with pytest.raises(HTTPException) as error:
        asyncio.run(env.store.provider_content(env.users[0]['id'], [attachment['id']]))
    assert error.value.status_code == 401
    env.accounts.finish_delete(env.users[0]['id'])
    assert env.accounts.db.execute('SELECT COUNT(*) FROM hope_attachments').fetchone()[0] == 0
    assert env.accounts.db.execute('SELECT COUNT(*) FROM hope_artifact_versions').fetchone()[0] == 0
    # Retained private disk is deliberate; account deletion is not guarded erase.
    assert (env.store.root / attachment['id']).exists()


def test_cookie_csrf_limits_and_artifact_versions(env):
    env.client.cookies.set('hope_session', env.tokens[0])
    assert env.client.post('/api/artifacts', json={'title': 'x'}).status_code == 403
    item = artifact(env)
    url = '/api/artifacts/' + item['id']
    assert env.client.patch(url, headers=env.headers[0], json={'revision': 1, 'content': 'x' * 129}).status_code == 413
    assert env.client.patch(url, headers=env.headers[0], json={'revision': 1}).status_code == 200
    assert env.client.patch(url, headers=env.headers[0], json={'revision': 2}).status_code == 200
    assert env.client.patch(url, headers=env.headers[0], json={'revision': 3}).status_code == 413
    assert env.client.post('/api/artifacts', headers=env.headers[0], json={'title': 'x', 'user_id': env.users[1]['id']}).status_code == 422
    assert env.client.get('/api/features/sync?after=-1', headers=env.headers[0]).status_code == 422


class FakeFeatures:
    def __init__(self):
        self.calls = []
        self.qualified = True

    async def feature_capabilities(self, owner):
        return {'scheduled_tasks': self.qualified}

    async def feature_request(self, owner, method, path, *, json=None, params=None):
        self.calls.append((owner, method, path, json, params))
        job = {'id': 'abcdef123456', 'name': 'Reminder', 'schedule': 'every 2h', 'prompt': 'Hello', 'enabled': True,
               'provider': 'secret', 'base_url': 'secret', 'api_key': 'secret', 'script': 'secret'}
        if method == 'GET':
            return {'jobs': [job]}
        if method == 'DELETE':
            return {'ok': True}
        return {'job': job}


def test_tasks_exact_gateway_mapping_no_secret_projection(env):
    fake = FakeFeatures()
    app = FastAPI()
    app.state.accounts = env.accounts
    app.state.settings = Settings()
    app.include_router(create_feature_router(env.store, fake))
    with TestClient(app) as client:
        result = client.get('/api/tasks', headers=env.headers[0])
        assert result.status_code == 200
        assert 'secret' not in result.text
        assert fake.calls[-1] == (env.users[0]['id'], 'GET', '/api/jobs', None, {'include_disabled': 'true'})
        body = {'name': 'Reminder', 'schedule': 'every 2h', 'prompt': 'Hello', 'repeat': 2}
        result = client.post('/api/tasks', headers=env.headers[0], json=body)
        assert result.status_code == 201
        assert fake.calls[-1][3] == {**body, 'deliver': 'local'}
        assert client.post('/api/tasks', headers=env.headers[0], json={**body, 'deliver': 'external'}).status_code == 422
        assert client.patch('/api/tasks/abcdef123456', headers=env.headers[0], json={'enabled': False}).status_code == 200
        for action in ('pause', 'resume', 'run'):
            assert client.post('/api/tasks/abcdef123456/' + action, headers=env.headers[1]).status_code == 200
            assert fake.calls[-1][0] == env.users[1]['id']
            assert fake.calls[-1][2] == '/api/jobs/abcdef123456/' + action
        assert client.delete('/api/tasks/abcdef123456', headers=env.headers[0]).status_code == 200
        assert client.post('/api/tasks/bad/run', headers=env.headers[0]).status_code == 404
        fake.qualified = False
        assert client.get('/api/tasks', headers=env.headers[0]).status_code == 501
        assert client.get('/api/connectors', headers=env.headers[0]).json()['available'] is False
        assert client.get('/api/research', headers=env.headers[0]).json()['available'] is False
        assert client.get('/api/memory', headers=env.headers[0]).status_code == 501


def test_default_provider_auth_fixed_paths_memory_conflict(env, tmp_path):
    import httpx
    requests = []
    def handle(request):
        requests.append(request)
        return httpx.Response(200, json={'jobs': []})
    http = httpx.AsyncClient(transport=httpx.MockTransport(handle))
    provider = FeatureProvider(env.accounts, http_client=http, state_dir=tmp_path / 'hermes')
    env.accounts.update(env.users[0]['id'], hermes_url='http://127.0.0.1:20000', hermes_api_key='private')
    memory = provider.state_dir / env.users[0]['id'] / 'home' / 'memories'
    memory.mkdir(parents=True)
    (memory / 'MEMORY.md').write_text('old')
    result = asyncio.run(provider.read_memory(env.users[0]['id']))
    revision = result['memory']['revision']
    updated = asyncio.run(provider.write_memory(env.users[0]['id'], 'memory', 'new', revision))
    assert updated['revision'] != revision
    assert (memory / 'MEMORY.md').read_text() == 'new'
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.write_memory(env.users[0]['id'], 'memory', 'stale', revision))
    assert error.value.status_code == 409
    asyncio.run(provider.feature_request(env.users[0]['id'], 'GET', '/api/jobs'))
    assert requests[-1].headers['authorization'] == 'Bearer private'
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.feature_request(env.users[0]['id'], 'GET', '/api/config'))
    assert error.value.status_code == 501
    outside = tmp_path / 'private-file'
    outside.write_text('never touch')
    (memory / 'USER.md').symlink_to(outside)
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.read_memory(env.users[0]['id']))
    assert error.value.status_code == 409
    assert outside.read_text() == 'never touch'
    env.accounts.begin_delete(env.users[0]['id'])
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.feature_request(env.users[0]['id'], 'GET', '/api/jobs'))
    assert error.value.status_code == 401
    asyncio.run(http.aclose())


def test_memory_special_file_and_hardlink_rejected(env, tmp_path):
    provider = FeatureProvider(env.accounts, state_dir=tmp_path / 'hermes')
    env.accounts.update(env.users[0]['id'], hermes_url='http://127.0.0.1:20000', hermes_api_key='private')
    directory = provider.state_dir / env.users[0]['id'] / 'home' / 'memories'
    directory.mkdir(parents=True)
    target = directory / 'MEMORY.md'
    os.mkfifo(target)
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.read_memory(env.users[0]['id']))
    assert error.value.status_code == 409
    target.unlink()
    outside = tmp_path / 'outside-memory'
    outside.write_text('untouched')
    os.link(outside, target)
    with pytest.raises(HTTPException) as error:
        asyncio.run(provider.write_memory(env.users[0]['id'], 'memory', 'bad', 'x'))
    assert error.value.status_code == 409
    assert outside.read_text() == 'untouched'
    asyncio.run(provider.aclose())


def test_cumulative_artifact_quota_and_metadata_list(env):
    from dataclasses import replace
    env.store.limits = replace(env.store.limits, account_artifact_bytes=12, global_artifact_bytes=20)
    item = artifact(env)  # Seven bytes.
    listing = env.client.get('/api/artifacts', headers=env.headers[0]).json()['artifacts']
    assert 'content' not in listing[0]
    assert env.client.patch('/api/artifacts/' + item['id'], headers=env.headers[0], json={'revision': 1, 'content': '123456'}).status_code == 413
    assert env.client.patch('/api/artifacts/' + item['id'], headers=env.headers[0], json={'revision': 1, 'content': '12345'}).status_code == 200
    versions = env.client.get('/api/artifacts/' + item['id'] + '/versions', headers=env.headers[0]).json()['versions']
    assert [v['revision'] for v in versions] == [2, 1]
    assert env.client.get('/api/artifacts/' + item['id'] + '/versions', headers=env.headers[1]).status_code == 404


def test_pdf_subprocess_limits_and_admission(env, monkeypatch):
    import control.chat_features as module
    monkeypatch.setattr(module.importlib.util, 'find_spec', lambda name: object())
    calls = []
    def run(args, **kwargs):
        calls.append((args, kwargs))
        assert 'RLIMIT_AS' in args[3] and 'RLIMIT_CPU' in args[3]
        assert kwargs['timeout'] == 6
        return SimpleNamespace(stdout=b'["extracted", "ready"]')
    monkeypatch.setattr(module.subprocess, 'run', run)
    response = upload(env, b'%PDF-offline', 'application/pdf', 'file.pdf')
    assert response.status_code == 201
    assert response.json()['attachment']['extraction_status'] == 'ready'
    assert len(calls) == 1
    env.store.upload_admission.acquire()
    env.store.upload_admission.acquire()
    try:
        assert upload(env, b'%PDF-offline', 'application/pdf', 'file.pdf').status_code == 429
        assert len(calls) == 1
    finally:
        env.store.upload_admission.release()
        env.store.upload_admission.release()