import asyncio
import base64
import json
import os
from pathlib import Path

import httpx
import pytest
from fastapi.responses import PlainTextResponse
from fastapi.testclient import TestClient

from app import archive_auth, archive_download, bt, common, download, main, store, torrent_meta


def bencode(item):
    if isinstance(item, int):
        return b'i' + str(item).encode() + b'e'
    if isinstance(item, str):
        item = item.encode()
    if isinstance(item, bytes):
        return str(len(item)).encode() + b':' + item
    if isinstance(item, (list, tuple)):
        return b'l' + b''.join(bencode(x) for x in item) + b'e'
    if isinstance(item, dict):
        return b'd' + b''.join(bencode(k) + bencode(v) for k, v in sorted(item.items())) + b'e'
    raise TypeError(item)


def sample_torrent():
    return bencode({b'announce': b'https://tracker.example/announce', b'info': {
        b'name': b'MyGame', b'files': [
            {b'length': 100, b'path': [b'Game.iso']},
            {b'length': 40, b'path': [b'readme.txt']},
            {b'length': 12, b'path': [b'..padding', b'pad'], b'attr': b'p'},
        ]}})


@pytest.fixture
def isolated(tmp_path, monkeypatch):
    monkeypatch.setattr(store, 'DATA', tmp_path)
    monkeypatch.setattr(store, 'DB_PATH', tmp_path / 'queue.sqlite3')
    monkeypatch.setenv('WEB_USER', 'admin')
    monkeypatch.setenv('WEB_PASSWORD', 'mysecretpassword')
    monkeypatch.setenv('ARIA2_SECRET', 'secret-rpc-key')
    store.init_db()
    async def idle_worker():
        await asyncio.sleep(10000)
    monkeypatch.setattr(main, 'worker', idle_worker)
    return tmp_path


def test_torrent_metadata_parser_and_magnet_selection(tmp_path):
    file = tmp_path / 'test.torrent'
    file.write_bytes(sample_torrent())
    meta = torrent_meta.read_torrent(file)
    assert [row['name'] for row in meta['files']] == [
        'MyGame/Game.iso', 'MyGame/readme.txt', 'MyGame/..padding/pad']
    assert meta['files'][0]['index'] == 1
    assert meta['files'][2]['padding'] is True
    assert torrent_meta.suggested_indices('magnet:?xt=urn:btih:123&so=15,0-1', 20) == [1, 2, 16]


def test_torrent_path_traversal_rejected(tmp_path):
    file = tmp_path / 'bad.torrent'
    file.write_bytes(bencode({b'info': {b'name': b'Game', b'files': [
        {b'length': 5, b'path': [b'..', b'secret']}]}}))
    with pytest.raises(torrent_meta.MetadataError):
        torrent_meta.read_torrent(file)
    with pytest.raises(torrent_meta.MetadataError):
        torrent_meta.decode_bencode(b'd3:fooi1eeTRAILING')


def test_archive_session_secrets_and_logout(isolated):
    archive_auth._save_session('test@archive.org', {
        'logged-in-user': 'member%40archive.org', 'logged-in-sig': 'signedsession'})
    path = archive_auth.session_path()
    assert oct(path.stat().st_mode & 0o777) == '0o600'
    assert archive_auth.session_status() == {'authenticated': True, 'email': 'test@archive.org'}
    assert archive_auth.session_cookies().get('logged-in-sig') == 'signedsession'
    archive_auth.logout()
    assert archive_auth.session_status()['authenticated'] is False
    assert not path.exists()


def test_archive_xauthn_login_no_password_saved(isolated, monkeypatch):
    original = httpx.AsyncClient
    def handler(request):
        assert request.method == 'POST'
        assert request.url.host == 'archive.org'
        assert b'password=not-saved' in request.content
        return httpx.Response(200, json={'success': True, 'values': {'cookies': {
            'logged-in-user': 'user%40example.org; Path=/',
            'logged-in-sig': 'sig%3Dabc; Path=/'}}})
    transport = httpx.MockTransport(handler)
    monkeypatch.setattr(archive_auth.httpx, 'AsyncClient', lambda **kw: original(**kw, transport=transport))
    status = asyncio.run(archive_auth.login('user@example.org', 'not-saved'))
    assert status['authenticated'] is True
    content = archive_auth.session_path().read_text()
    assert 'not-saved' not in content
    assert 'sig%3Dabc' in content


def test_archive_relay_redirect_cookie_scoping_and_range(isolated, monkeypatch):
    archive_auth._save_session('user@example.org', {
        'logged-in-user': 'member', 'logged-in-sig': 'secret-cookie'})
    originals = httpx.AsyncClient
    seen = []
    def handler(request):
        seen.append((request.url.host, request.headers.get('cookie'), request.headers.get('range')))
        if request.url.host == 'archive.org':
            return httpx.Response(302, headers={'Location': 'https://ia808.us.archive.org/download/asset.zip'})
        class FakeStream(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield b'bcd'
        return httpx.Response(206, headers={'Content-Length': '3', 'Content-Range': 'bytes 1-3/4',
                                            'Content-Type': 'application/octet-stream'}, stream=FakeStream())
    transport = httpx.MockTransport(handler)
    monkeypatch.setattr(archive_auth.httpx, 'AsyncClient', lambda **kw: originals(**kw, transport=transport))

    async def call():
        response = await archive_auth.stream_archive('https://archive.org/download/item/asset.zip', 'GET',
                                                     {'range': 'bytes=1-3'})
        data = b''
        async for part in response.body_iterator:
            data += part
        return response, data
    response, data = asyncio.run(call())
    assert response.status_code == 206
    assert data == b'bcd'
    assert len(seen) == 2
    assert all('secret-cookie' in (item[1] or '') for item in seen)
    assert all(item[2] == 'bytes=1-3' for item in seen)
    assert archive_auth.allowed_redirect('https://attacker.example/') is False
    assert archive_auth.allowed_redirect('http://archive.org/') is False


def test_magnet_inspection_selection_and_signed_archive_proxy(isolated, monkeypatch):
    started = []
    stopped = []
    async def fake_start(preview_id, magnet):
        started.append((preview_id, magnet))
    async def fake_stop(preview_id):
        stopped.append(preview_id)
    monkeypatch.setattr(bt, 'start_inspection', fake_start)
    monkeypatch.setattr(bt, 'stop_inspection', fake_stop)
    monkeypatch.setattr(bt, 'inspect_state', lambda pid: {'connections': 2, 'known_peers': 5})
    with TestClient(main.app) as client:
        auth = ('admin', 'mysecretpassword')
        magnet = 'magnet:?xt=urn:btih:c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d&dn=Test&so=1'
        r = client.post('/api/magnet/inspect', auth=auth, json={'magnet': magnet})
        assert r.status_code == 200, r.text
        pid = r.json()['preview_id']
        status = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
        assert status['status'] == 'loading'
        assert status['connections'] == 2 and status['known_peers'] == 5
        assert started == [(pid, magnet)]
        folder = torrent_meta.inspect_dir(pid)
        (folder / 'hash.torrent').write_bytes(sample_torrent())
        info = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
        assert info['status'] == 'ready'
        assert info['suggested'] == [2]
        assert info['files'][1]['name'] == 'MyGame/readme.txt'
        assert client.post('/api/magnet/jobs', auth=auth,
                           json={'preview_id': pid, 'files': [4], 'format': 'raw'}).status_code == 400
        assert client.post('/api/magnet/jobs', auth=auth,
                           json={'preview_id': pid, 'files': [3], 'format': 'raw'}).status_code == 400
        r = client.post('/api/magnet/jobs', auth=auth,
                        json={'preview_id': pid, 'files': [2], 'format': 'raw'})
        assert r.status_code == 200, r.text
        job = r.json()
        assert job['source_type'] == 'torrent_selected'
        assert (isolated / 'torrents' / (job['id'] + '.torrent')).read_bytes() == sample_torrent()
        assert json.loads(job['source'])['selection'] == '2'
        assert pid in stopped

        archive_job = main.new_job('archive_many', json.dumps(['https://archive.org/download/test/game.iso']), 'Demo', 'raw')
        uri = '/internal/archive/' + archive_job['id'] + '/0'
        assert client.get(uri).status_code == 403
        assert client.get(uri, headers={'X-X360-Proxy-Token': 'invalid'}).status_code == 403
        async def fake_stream(url, method, headers):
            assert url == 'https://archive.org/download/test/game.iso'
            return PlainTextResponse('proxy-ok')
        monkeypatch.setattr(archive_auth, 'stream_archive', fake_stream)
        token = archive_auth.signed_header(archive_job['id'], 0).split(': ', 1)[1]
        assert client.get(uri, headers={'X-X360-Proxy-Token': token}).text == 'proxy-ok'
        assert client.post('/api/jobs', auth=auth,
                           json={'url': magnet, 'format': 'raw'}).status_code == 400


def test_wrapped_and_html_escaped_magnet():
    hash_value = 'c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d'
    pasted = "'**magnet**:?xt=urn:btih:" + hash_value + "&dn=Minerva_Myrient\\\n&so=15&tr=udp%3A%2F%2Ftracker.ope\\\nnbittorrent.com%3A6969%2Fannounce'"
    kind, cleaned = common.validate_url(pasted)
    assert kind == 'magnet'
    assert cleaned.startswith('magnet:?xt=urn:btih:' + hash_value + '&dn=')
    assert '&so=15&tr=' in cleaned
    assert 'tracker.openbittorrent.com' in cleaned
    assert '\\' not in cleaned and '&' not in cleaned and '\n' not in cleaned
    assert torrent_meta.suggested_indices(cleaned, 20) == [16]
    with pytest.raises(ValueError):
        common.validate_url('magnet:?xt=urn:btih:abcdef')


def test_upload_torrent_and_queue_selection(isolated, monkeypatch):
    async def fake_rpc(method, *args):
        if method == 'addTorrent':
            return '123456'
        if method == 'tellStatus':
            return {'status': 'active', 'connections': '0'}
        return 'ok'
    monkeypatch.setattr(download, 'rpc', fake_rpc)
    with TestClient(main.app) as client:
        auth = ('admin', 'mysecretpassword')
        uploaded = client.post('/api/magnet/upload', auth=auth,
            content=sample_torrent(), headers={'Content-Type': 'application/x-bittorrent'})
        assert uploaded.status_code == 200, uploaded.text
        pid = uploaded.json()['preview_id']
        info = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
        assert info['status'] == 'ready'
        assert info['files'][0]['name'] == 'MyGame/Game.iso'
        assert info['suggested'] == []
        job = client.post('/api/magnet/jobs', auth=auth,
             json={'preview_id': pid, 'files': [1], 'format': 'xex'})
        assert job.status_code == 200, job.text
        assert (isolated / 'torrents' / (job.json()['id'] + '.torrent')).exists()
        assert client.post('/api/magnet/upload', auth=auth,
            content=b'garbage', headers={'Content-Type': 'application/x-bittorrent'}).status_code == 400
        assert client.post('/api/magnet/upload', auth=auth,
            content=sample_torrent(), headers={'Content-Type': 'application/json'}).status_code == 415


def test_metadata_status_includes_connections_and_timeout(isolated, monkeypatch):
    stopped = []
    async def fake_start(preview_id, magnet):
        pass
    async def fake_stop(preview_id):
        stopped.append(preview_id)
    monkeypatch.setattr(bt, 'start_inspection', fake_start)
    monkeypatch.setattr(bt, 'stop_inspection', fake_stop)
    monkeypatch.setattr(bt, 'inspect_state', lambda pid: {'connections': 0, 'known_peers': 0})
    magnet='magnet:?xt=urn:btih:c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d&so=15'
    with TestClient(main.app) as client:
        auth=('admin','mysecretpassword')
        resp=client.post('/api/magnet/inspect',auth=auth,json={'magnet':magnet})
        assert resp.status_code == 200
        pid=resp.json()['preview_id']
        status=client.get('/api/magnet/inspect/'+pid,auth=auth).json()
        assert status['status']=='loading' and status['connections']==0
        request_path=torrent_meta.inspect_dir(pid)/'request.json'
        state=json.loads(request_path.read_text())
        state['created'] -= 310
        request_path.write_text(json.dumps(state))
        expired=client.get('/api/magnet/inspect/'+pid,auth=auth).json()
        assert expired['status']=='error'
        assert pid in stopped