"""Deterministic V8.5 API auth / MySQL staging and Worker regression tests.
No real production database or credentials are accessed.
"""
import json
from urllib.parse import urlsplit
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app import db, enterprise, jobs, v8_workflows, connectors_v84 as api, connectors_v85 as mysql
from app import safe_http, secret_refs
from app.core import uid
from app.main import app
from app.safe_http import Fetched, OutboundError

@pytest.fixture
def env(tmp_path,monkeypatch):
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
    monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','v85-admin-strong-for-test')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','v85-long-private-session-secret-test')
    monkeypatch.setenv('SITEFORGE_MYSQL_ALLOWED_HOSTS','mysql.example.org')
    monkeypatch.setenv('SITEFORGE_MYSQL_ALLOWED_TABLES','catalog.products')
    monkeypatch.setenv('SITEFORGE_API_TEST_TOKEN','unit-secret-never-return')
    db.init();enterprise.setup();v8_workflows.setup();api.setup();mysql.setup()
    def site():
        sid=uid();now=db.now()
        db.insert('INSERT INTO sites(id,name,theme,mode,created_at,updated_at) VALUES(?,?,?,?,?,?)',
                  (sid,'Connector V85','industrial','new',now,now))
        return sid
    return site


def mysql_config(**kw):
    return {'host':'mysql.example.org','database':'catalog','table':'products','key_column':'id',
        'columns':['id','name','brand'],'user_ref':'env:SITEFORGE_MYSQL_READ_USER',
        'password_ref':'file:mysql_read_password','mapping':{'title':'name','brand':'brand'},
        'page_size':2,'max_pages':4,**kw}


def test_existing_api_sources_upgraded_without_erasing(env):
    site=env();s=api.create_api_source(site,{'url':'https://api.example.org/data'})
    api.setup();row=db.one('SELECT auth_type FROM v84_api_sources WHERE id=?',(s['id'],))
    assert row['auth_type']=='none' and api.status(site,s['id'])['state']=='ready'


def test_api_secret_reference_redacted_and_authenticated_worker(env,monkeypatch):
    site=env()
    s=api.create_api_source(site,{'url':'https://api.example.org/products','items_path':'data.items',
        'page_size':2,'max_pages':2,'mapping':{'title':'name'},
        'auth_type':'bearer','secret_ref':'env:SITEFORGE_API_TEST_TOKEN'})
    assert s['auth_type']=='bearer' and 'secret_ref' not in s and 'unit-secret' not in str(s)
    requests=[]
    def fake_fetch(url,**kw):
        requests.append(kw)
        data={'data':{'items':[{'name':'Gamma','id':'3'}]}}
        return Fetched(url,200,{'content-type':'application/json'},json.dumps(data).encode())
    monkeypatch.setattr(api.safe_http,'fetch',fake_fetch)
    assert api.run_scan(site,s['id'])['staged_count']==1
    assert requests[0]['credential_headers']=={'Authorization':'Bearer unit-secret-never-return'}
    assert not 'unit-secret' in str(api.status(site,s['id']))
    assert 'unit-secret' not in json.dumps(db.rows('SELECT payload_json FROM build_jobs'))


def test_api_custom_header_and_credential_validation(env):
    site=env()
    for auth in ({'auth_type':'bearer','secret_ref':'my-token'},
                 {'auth_type':'custom','secret_ref':'file:partner','auth_header':'Cookie'},
                 {'auth_type':'none','secret_ref':'env:SITEFORGE_API_TEST_TOKEN'}):
        with pytest.raises(ValueError):api.create_api_source(site,{'url':'https://api.example.org/data',**auth})
    custom=api.create_api_source(site,{'url':'https://api.example.org/data','auth_type':'custom',
        'secret_ref':'file:partner_api','auth_header':'X-Partner-Key'})
    assert custom['auth_header']=='X-Partner-Key'


def test_http_rejects_credential_redirects_and_http():
    with pytest.raises(OutboundError):safe_http.fetch('http://example.org',credential_headers={'Authorization':'Bearer abc'},redirects=0)
    with pytest.raises(OutboundError):safe_http.fetch('https://example.org',credential_headers={'Authorization':'Bearer abc'},redirects=1)
    with pytest.raises(OutboundError):safe_http.fetch('https://example.org',credential_headers={'Cookie':'abc'},redirects=0)
    with pytest.raises(OutboundError):safe_http.fetch('https://example.org',credential_headers={'X-Api-Key':'abc\r\nHost: evil'},redirects=0)


def test_mysql_host_table_identifier_and_secret_validation(env):
    site=env()
    for patch in [{'host':'internal-unauthorized'},{'database':'other'},
                  {'table':'products;DROP TABLE users'},{'columns':['id','bad-name']},
                  {'key_column':'unknown'},{'password_ref':'literal-password'},
                  {'max_pages':51},{'page_size':1000}]:
        with pytest.raises(ValueError):mysql.create_source(site,mysql_config(**patch))
    source=mysql.create_source(site,mysql_config());record=mysql.status(site,source['id'])
    assert 'password_ref' not in record and 'user_ref' not in record
    assert record['host']=='mysql.example.org'


def test_mysql_query_is_fixed_and_parameterized(env):
    site=env();created=mysql.create_source(site,mysql_config())
    src=mysql._one(site,created['id']);sql,params=mysql.query_for(src,'12')
    assert sql=='SELECT `id`, `name`, `brand` FROM `catalog`.`products` WHERE `id` > %s ORDER BY `id` ASC LIMIT %s'
    assert params==('12',2)
    assert '12' not in sql


def test_mysql_scan_stages_without_auto_committing_and_can_rollback(env,monkeypatch):
    site=env();created=mysql.create_source(site,mysql_config());sid=created['id']
    calls=[]
    def fake_db_read(source,cursor):
        calls.append(cursor)
        if cursor is None:return [{'id':1,'name':'Engine A','brand':'Brand A'},{'id':2,'name':'Engine B','brand':'Brand B'}]
        return []
    monkeypatch.setattr(mysql,'_fetch_page',fake_db_read)
    report=mysql.run_scan(site,sid)
    assert report['staged_count']==2 and report['state']=='completed' and calls==[None,'2']
    batches=mysql.status(site,sid)['pages']
    assert len(batches)==2 and batches[0]['batch_id']
    assert db.rows('SELECT id FROM entries WHERE site_id=?',(site,))==[]
    committed=v8_workflows.commit(site,batches[0]['batch_id'])
    assert committed['created']==2
    assert v8_workflows.rollback(site,batches[0]['batch_id'])['state']=='rolled_back'
    monkeypatch.setattr(mysql,'_fetch_page',lambda *a: (_ for _ in ()).throw(AssertionError('must not re-run')))
    assert mysql.run_scan(site,sid)['staged_count']==2


def test_mysql_retry_after_staging_before_checkpoint(env,monkeypatch):
    site=env();source=mysql.create_source(site,mysql_config())['id']
    batch=v8_workflows.stage(site,[{'id':'4','name':'Item Four','brand':'Maker'}],
        {'title':'name','brand':'brand'},'product',f'mysql:{source}:page:1',[json.dumps('4')])
    monkeypatch.setattr(mysql,'_fetch_page',lambda *a:(_ for _ in ()).throw(AssertionError('should reuse batch')))
    row=mysql.scan_page(site,source)
    assert row['batch_id']==batch['batch_id'] and row['count']==1
    assert len(db.rows('SELECT id FROM v8_import_batches WHERE site_id=?',(site,)))==1


def test_mysql_worker_job_dedup_site_isolation_cancel(env):
    site=env();source=mysql.create_source(site,mysql_config())['id']
    a=mysql.submit_scan(site,source,'tester')
    b=mysql.submit_scan(site,source,'tester')
    assert a['id']==b['id'] and b['existing']
    assert db.one('SELECT kind FROM build_jobs WHERE id=?',(a['id'],))['kind']=='mysql_scan'
    other=env()
    with pytest.raises(HTTPException):mysql.status(other,source)
    jobs.cancel(site,a['id'])
    assert db.one('SELECT state FROM build_jobs WHERE id=?',(a['id'],))['state']=='cancelled'


def test_mysql_auth_required_and_redaction(env):
    site=env()
    with TestClient(app) as client:
        route=f'/api/connectors/sites/{site}/mysql-sources'
        assert client.get(route).status_code==401
        assert client.post(route,json=mysql_config()).status_code==401
        assert client.post('/api/auth/login',json={'password':'v85-admin-strong-for-test'}).status_code==200
        client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf']
        c=client.post(route,json=mysql_config())
        assert c.status_code==201,c.text
        assert 'file:mysql_read_password' not in c.text
        assert 'env:SITEFORGE_MYSQL_READ_USER' not in c.text
        assert len(client.get(route).json()['items'])==1
        assert client.get('/production-v8').status_code==200
        assert 'myCreate' in client.get('/production-v8').text


def test_mysql_worker_runner_dispatch(env,monkeypatch):
    import sys
    from app import job_runner
    site=env();source=mysql.create_source(site,mysql_config())['id']
    jid=mysql.submit_scan(site,source,'tester')['id']
    monkeypatch.setattr(mysql,'run_scan',lambda sid,src,jid=None:{'state':'completed','source':src})
    monkeypatch.setattr(sys,'argv',['job_runner',jid])
    assert job_runner.main()==0
    result=json.loads((db.DATA/'job_results'/f'{jid}.json').read_text())
    assert result['ok'] and result['result']['state']=='completed'


def test_relationship_preflight_detects_brand_references(env):
    site=env()
    batch=v8_workflows.stage(site,[
        {'title':'SUV Alpha','brand':'Brand A','category':'SUV'},
        {'title':'SUV Beta','brand':'Brand A','category':'SUV'},
    ],{'title':'title','brand':'brand','category':'category'},'product','xml:relationship')
    r=mysql.relationship_preview(site,batch['batch_id'])
    assert r['brands']=={'Brand A':2} and r['categories']=={'SUV':2}
    assert r['unlinked_brand_total']==2
    v8_workflows.stage(site,[{'title':'Brand A'}],{'title':'title'},'brand','manual')
    # Merely staging a separate batch is not equivalent to committing a brand page.
    assert mysql.relationship_preview(site,batch['batch_id'])['unlinked_brand_total']==2


def test_relationship_preflight_site_scope(env):
    one=env();two=env()
    batch=v8_workflows.stage(one,[{'title':'Product'}],{'title':'title'},'product','manual')
    with pytest.raises(HTTPException) as exc:mysql.relationship_preview(two,batch['batch_id'])
    assert exc.value.status_code==404


def test_mysql_staged_products_can_generate_real_static_site(env,monkeypatch):
    from app import builder
    site=env()
    src=mysql.create_source(site,mysql_config(page_size=2,max_pages=3))['id']
    monkeypatch.setattr(mysql,'_fetch_page',lambda cfg,cursor: (
        [{'id':1,'name':'Pump A','brand':'Brand A'},{'id':2,'name':'Pump B','brand':'Brand B'}]
        if cursor is None else []))
    outcome=mysql.run_scan(site,src)
    assert outcome['staged_count']==2
    batch=outcome['pages'][0]['batch_id']
    assert v8_workflows.commit(site,batch)['created']==2
    with TestClient(app):
        built=builder.build(site)
    assert built['pages']>=2
    assert built['version']


def test_pinned_https_places_auth_only_in_explicit_secret_headers(monkeypatch):
    captured=[]
    class Response:
        status=200
        def getheaders(self):return [('Content-Type','application/json')]
        def read(self,maxlen):return b'{}'
    class Connection:
        def __init__(self,host,ip,port,timeout):
            captured.append((host,ip,port,timeout))
        def request(self,method,path,body=None,headers=None):
            captured.append((method,path,headers))
        def getresponse(self):return Response()
        def close(self):pass
    monkeypatch.setattr(safe_http,'resolve_target',lambda url:(urlsplit(url),'api.example.org',443,'198.51.100.20'))
    monkeypatch.setattr(safe_http,'_PinnedHTTPS',Connection)
    r=safe_http.fetch('https://api.example.org/items',headers={'Authorization':'SHOULD-BE-DROPPED'},
                      credential_headers={'Authorization':'Bearer real-worker-token'},redirects=0)
    assert r.status_code==200
    assert captured[0][1]=='198.51.100.20'
    assert captured[1][2]['Authorization']=='Bearer real-worker-token'
    assert 'SHOULD-BE-DROPPED' not in str(captured)