import json
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, update
from app.config import settings
from app.db import Job, claim, engine, sessions
from app.main import app
from app.storage import storage
from app.worker import process

KEY = 'test-api-key-' + 'x'*40

@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setenv('STORAGE_BACKEND', 'local')
    monkeypatch.setenv('STORAGE_DIR', str(tmp_path))
    monkeypatch.setenv('DATABASE_URL', f'sqlite:///{tmp_path}/test.db')
    monkeypatch.setenv('API_KEY', KEY)
    monkeypatch.setenv('PUBLIC_BASE_URL', 'http://testserver')
    monkeypatch.setenv('MAX_UPLOAD_BYTES', '1048576')
    settings.cache_clear(); storage.cache_clear(); sessions.cache_clear(); engine.cache_clear()
    with TestClient(app) as client: yield client
    engine().dispose()
    settings.cache_clear(); storage.cache_clear(); sessions.cache_clear(); engine.cache_clear()


def upload(client, data, **fields):
    return client.post('/api/v1/schematics/render', files={'file': ('sample.SchDoc', data, 'application/octet-stream')},
                       data=fields, headers={'X-API-Key': KEY})


def test_private_upload_worker_viewer_and_expiry(client, ascii_schematic):
    response = upload(client, ascii_schematic)
    assert response.status_code == 202, response.text
    body = response.json(); id = body['id']; token = {'X-Viewer-Token': body['viewer_token']}
    assert len(id) == 48 and '#token=' in body['viewer_url']
    base = f'/api/v1/schematics/{id}'
    assert client.get(base+'/status').status_code == 404
    assert client.get(base+'/status',headers=token).json()['status'] == 'queued'
    assert client.get(base+'/data',headers=token).status_code == 409
    job = claim(); assert job['id'] == id
    process(job)
    response = client.get(base+'/status',headers=token)
    assert response.json()['status'] == 'completed', response.text
    assert response.json()['compatibility'] == 'partial'
    data = client.get(base+'/data',headers=token).json()
    assert data['components'][0]['designator'] == 'R1'
    drawing = client.get(base+'/svg',headers=token)
    assert drawing.status_code == 200 and '<svg' in drawing.text and 'R1' in drawing.text
    view = client.get('/view/'+id)
    assert view.status_code == 200 and 'viewer.js' in view.text
    assert 'no-referrer' == view.headers['referrer-policy']
    with sessions().begin() as db: db.execute(update(Job).where(Job.id==id).values(expires_at=time.time()-1))
    assert client.get(base+'/data',headers=token).status_code == 410
    assert client.get(base+'/status').status_code == 404


def test_unlisted_and_api_key(client, ascii_schematic):
    assert client.post('/api/v1/schematics/render').status_code == 401
    assert client.post('/api/v1/schematics/render',headers={'X-API-Key':'bad'}).status_code == 401
    body = upload(client,ascii_schematic,access='unlisted').json()
    assert 'viewer_token' not in body
    assert client.get(f"/api/v1/schematics/{body['id']}/status").status_code == 200


def test_upload_validation(client):
    assert upload(client,b'not a schematic').status_code == 422
    response=client.post('/api/v1/schematics/render',files={'file':('bad.PcbDoc',b'a')},headers={'X-API-Key':KEY})
    assert response.status_code == 415
    assert upload(client,b'x'*(1048576+1)).status_code == 413
    assert upload(client,b'|HEADER=x',expires_in=999999999).status_code == 422
    assert upload(client,b'|HEADER=x',access='public').status_code == 422


def test_multipart_limit_before_spooling(client):
    r = client.post('/api/v1/schematics/render',content=b'x'*(3*1048576),
                    headers={'X-API-Key':KEY,'Content-Type':'multipart/form-data; boundary=foo'})
    assert r.status_code == 413, r.text


def test_corruption_is_durable_failure(client):
    body = upload(client,b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1'+b'\0'*504).json()
    process(claim())
    r = client.get(f"/api/v1/schematics/{body['id']}/status",headers={'X-API-Key':KEY})
    assert r.json()['status'] == 'failed'
    assert r.json()['error']['code'] == 'invalid_schematic'


def test_queue_capacity(client, ascii_schematic, monkeypatch):
    monkeypatch.setenv('MAX_PENDING_JOBS','1'); settings.cache_clear()
    assert upload(client,ascii_schematic).status_code == 202
    assert upload(client,ascii_schematic).status_code == 503


def test_atomic_claim_and_stale_worker_fencing(client, ascii_schematic):
    body=upload(client,ascii_schematic).json()
    with ThreadPoolExecutor(max_workers=4) as pool: jobs=list(pool.map(lambda _:claim(),range(4)))
    found=[job for job in jobs if job]
    assert len(found)==1
    old=found[0]
    with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(lease_until=time.time()-1))
    new=claim(); assert new['attempt']==2
    process(new); process(old)
    with sessions()() as db:
        job=db.get(Job,body['id'])
        assert job.status=='completed' and job.output_prefix.endswith('attempt-2')


def test_storage_path_traversal(client):
    for key in ['../etc/passwd','a'*48+'/../input.SchDoc','/tmp/evil','a'*48+'/input.SchDoc/evil']:
        with pytest.raises(ValueError): storage().get(key)


def test_api_docs_and_static(client):
    assert client.get('/openapi.json').json()['info']['version']=='1.0.0'
    assert client.get('/').status_code==200
    assert client.get('/assets/viewer.js').status_code==200
    assert client.get('/assets/not-allowed.js').status_code==404
    assert client.get('/view/nope').status_code==404


def test_production_fails_closed(client, monkeypatch):
    monkeypatch.setenv('APP_ENV','production'); monkeypatch.setenv('API_KEY',''); settings.cache_clear()
    with pytest.raises(RuntimeError,match='API_KEY'): settings()
    monkeypatch.setenv('API_KEY',KEY); settings.cache_clear()
    with pytest.raises(RuntimeError,match='HTTPS'): settings()


def test_expired_storage_cleanup(client, ascii_schematic):
    from app.cleanup import cleanup
    body=upload(client,ascii_schematic).json(); process(claim())
    identifier=body['id']
    with sessions().begin() as db: db.execute(update(Job).where(Job.id==identifier).values(expires_at=time.time()-1))
    assert cleanup()==1
    with sessions()() as db: assert db.get(Job,identifier) is None
    with pytest.raises(FileNotFoundError): storage().get(identifier+'/input.SchDoc')


def test_expired_worker_is_not_reclaimed(client, ascii_schematic):
    body=upload(client,ascii_schematic).json()
    with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(expires_at=time.time()-1))
    assert claim() is None


def test_repeated_worker_loss_fails_job(client, ascii_schematic):
    body=upload(client,ascii_schematic).json(); assert claim()['attempt']==1
    for attempt in (2,3):
        with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(lease_until=time.time()-1))
        result=claim()
        if attempt==2: assert result['attempt']==2
        else: assert result is None
    with sessions()() as db: assert db.get(Job,body['id']).error_code=='worker_lost'


def test_parser_wall_timeout_is_reported(client,ascii_schematic,monkeypatch):
    import subprocess
    from app import worker
    body=upload(client,ascii_schematic).json(); job=claim()
    def timeout(*args,**kw): raise subprocess.TimeoutExpired('parser',60)
    monkeypatch.setattr(worker.subprocess,'run',timeout)
    process(job)
    status=client.get(f"/api/v1/schematics/{body['id']}/status",headers={'X-API-Key':KEY}).json()
    assert status['status']=='failed' and status['error']['code']=='parse_timeout'


CORS_ORIGIN = 'https://git.sadentel.net'

@pytest.mark.parametrize('requested_headers', ['x-api-key', 'X-API-Key, Content-Type'])
def test_gitea_preflight_without_api_key(client, requested_headers):
    response = client.options('/api/v1/schematics/render', headers={
        'Origin': CORS_ORIGIN,
        'Access-Control-Request-Method': 'POST',
        'Access-Control-Request-Headers': requested_headers,
    })
    assert response.status_code == 200, response.text
    assert response.headers['access-control-allow-origin'] == CORS_ORIGIN
    assert {'POST', 'OPTIONS'} <= set(response.headers['access-control-allow-methods'].split(', '))
    allowed = {h.strip().lower() for h in response.headers['access-control-allow-headers'].split(',')}
    assert {'x-api-key', 'content-type'} <= allowed
    assert 'Origin' in response.headers['vary']
    assert 'access-control-allow-credentials' not in response.headers


@pytest.mark.parametrize('origin', ['https://evil.example', 'https://git.sadentel.net.evil.example', 'http://git.sadentel.net'])
def test_cors_rejects_other_origins(client, origin):
    response = client.options('/api/v1/schematics/render', headers={
        'Origin': origin,
        'Access-Control-Request-Method': 'POST',
        'Access-Control-Request-Headers': 'x-api-key',
    })
    assert response.status_code == 400
    assert 'access-control-allow-origin' not in response.headers


@pytest.mark.parametrize('key', [None, 'invalid-key', KEY])
def test_gitea_post_retains_authentication_and_cors(client, ascii_schematic, key):
    headers = {'Origin': CORS_ORIGIN}
    if key is not None: headers['X-API-Key'] = key
    response = client.post('/api/v1/schematics/render', headers=headers,
                           files={'file': ('design.SchDoc', ascii_schematic)}, data={'access': 'private'})
    assert response.status_code == (202 if key == KEY else 401)
    assert response.headers['access-control-allow-origin'] == CORS_ORIGIN


def test_gitea_private_status_read(client, ascii_schematic):
    created = upload(client, ascii_schematic).json()
    endpoint = f"/api/v1/schematics/{created['id']}/status"
    preflight = client.options(endpoint, headers={
        'Origin': CORS_ORIGIN, 'Access-Control-Request-Method': 'GET',
        'Access-Control-Request-Headers': 'x-viewer-token',
    })
    assert preflight.status_code == 200
    response = client.get(endpoint, headers={'Origin': CORS_ORIGIN, 'X-Viewer-Token': created['viewer_token']})
    assert response.status_code == 200
    assert response.headers['access-control-allow-origin'] == CORS_ORIGIN
    assert client.get(endpoint, headers={'Origin': CORS_ORIGIN}).status_code == 404


def test_cors_rejects_unneeded_methods_and_headers(client):
    for method, header in [('DELETE', 'x-api-key'), ('POST', 'x-unapproved-header')]:
        response = client.options('/api/v1/schematics/render', headers={
            'Origin': CORS_ORIGIN, 'Access-Control-Request-Method': method,
            'Access-Control-Request-Headers': header,
        })
        assert response.status_code == 400


DEFAULT_CSP = "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; font-src 'self'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'"
VIEWER_CSP = DEFAULT_CSP.replace("frame-ancestors 'none'", "frame-ancestors 'self' https://git.sadentel.net")

@pytest.mark.parametrize('method,path,expected_status', [
    ('GET', '/view/' + '0'*48, 200),
    ('GET', '/view/invalid-id', 404),
    ('GET', '/view/nested/missing', 404),
    ('POST', '/view/' + '0'*48, 405),
    ('GET', '/compare/' + '0'*48, 200),
    ('GET', '/compare/invalid-id', 404),
    ('GET', '/compare/nested/missing', 404),
    ('POST', '/compare/' + '0'*48, 405),
])
def test_viewer_only_allows_self_and_gitea_frames(client, method, path, expected_status):
    response = client.request(method, path)
    assert response.status_code == expected_status
    assert response.headers.get_list('content-security-policy') == [VIEWER_CSP]
    assert 'x-frame-options' not in response.headers
    assert response.headers['x-content-type-options'] == 'nosniff'


@pytest.mark.parametrize('path', ['/', '/docs', '/healthz', '/assets/viewer.js', '/view', '/compare', '/api/compare/'+'0'*48, '/viewer/invalid', '/api/v1/schematics/'+'0'*48+'/status'])
def test_non_viewer_keeps_original_csp(client, path):
    response = client.get(path)
    assert response.headers.get_list('content-security-policy') == [DEFAULT_CSP]


@pytest.mark.parametrize('frame_option', ['DENY', 'SAMEORIGIN'])
@pytest.mark.parametrize('prefix', ['/view/', '/compare/'])
def test_conflicting_headers_removed_only_for_viewer(client, monkeypatch, frame_option, prefix):
    from fastapi.responses import Response
    def page(*args, **kwargs):
        response = Response('page', media_type='text/html', headers={
            'X-Frame-Options': frame_option, 'Content-Security-Policy': DEFAULT_CSP,
        })
        response.headers.append('Content-Security-Policy', "frame-ancestors 'none'")
        return response
    monkeypatch.setattr('app.main.FileResponse', page)
    monkeypatch.setattr('app.compare_api.FileResponse', page)
    viewer = client.get(prefix+'0'*48)
    assert 'x-frame-options' not in viewer.headers
    assert viewer.headers.get_list('content-security-policy') == [VIEWER_CSP]
    other = client.get('/docs')
    assert other.headers['x-frame-options'] == frame_option
    assert other.headers.get_list('content-security-policy') == [DEFAULT_CSP]


def test_framed_viewer_does_not_bypass_private_token(client, ascii_schematic):
    created = upload(client, ascii_schematic).json()
    process(claim())
    identifier = created['id']
    assert client.get('/view/'+identifier).status_code == 200
    assert client.get('/view/'+identifier, headers={'Sec-Fetch-Dest':'iframe', 'Referer': CORS_ORIGIN+'/'}).status_code == 200
    for endpoint in ('status','data','svg'):
        path=f'/api/v1/schematics/{identifier}/{endpoint}'
        assert client.get(path).status_code == 404
        assert client.get(path, headers={'X-Viewer-Token':'invalid-token'}).status_code == 404
        response=client.get(path, headers={'X-Viewer-Token':created['viewer_token']})
        assert response.status_code == 200
        assert response.headers['content-security-policy'] == DEFAULT_CSP