cms-SiteForge/tests/test_connectors_v84.py

"""Deterministic V8.4 XML and public API connector regression fixtures."""
import json
from urllib.parse import parse_qs,urlsplit
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app import db,enterprise,v8_workflows,jobs,connectors_v84 as conn
from app.core import uid
from app.main import app
from app.safe_http import Fetched


@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','v84-unit-password-strong')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','v84-unit-secret-long-and-private-test-only')
    db.init();enterprise.setup();v8_workflows.setup();conn.setup()
    def site():
        sid=uid();now=db.now()
        db.insert('INSERT INTO sites(id,name,theme,mode,created_at,updated_at) VALUES(?,?,?,?,?,?)',
                  (sid,'Connector Test','automotive','new',now,now))
        return sid
    return site


def product_xml():
    return b'<catalog><item><title>Pickup A</title><category><name>Trucks</name></category></item><item><title>Pickup B</title><category><name>SUV</name></category></item></catalog>'


def test_xml_flat_two_level_fields_and_mapping(env):
    site=env()
    rows=conn.parse_xml_records(product_xml(),'item')
    assert rows==[{'title':'Pickup A','category.name':'Trucks'},{'title':'Pickup B','category.name':'SUV'}]
    batch=v8_workflows.stage(site,rows,{'title':'title','category':'category.name'},'product','xml:catalog')
    assert batch['mapped']==2
    saved=v8_workflows.commit(site,batch['batch_id'])
    assert saved['created']==2
    meta=json.loads(db.one('SELECT meta_json FROM entries WHERE site_id=? AND title=?',(site,'Pickup A'))['meta_json'])
    assert meta['fields']['category']=='Trucks'
    assert v8_workflows.rollback(site,batch['batch_id'])['state']=='rolled_back'


def test_xml_blocks_external_entities_and_expansion():
    for sample in [b'<!DOCTYPE foo [<!ENTITY exploit SYSTEM "file:///etc/passwd">]><foo><item><title>&exploit;</title></item></foo>',
                   b'<!DOCTYPE lolz [<!ENTITY a "9999">]><foo><item><title>&a;</title></item></foo>']:
        with pytest.raises(ValueError):conn.parse_xml_records(sample,'item')


def test_xml_deep_large_duplicate_and_bad_shape():
    with pytest.raises(ValueError):conn.parse_xml_records(b'<a>'+b'<a>'*11+b'<item/>'+b'</a>'*11+b'</a>','item')
    with pytest.raises(ValueError):conn.parse_xml_records(b'<root><item><title>x</title><title>y</title></item></root>','item')
    with pytest.raises(ValueError):conn.parse_xml_records(b'<root><item><a><b><c>x</c></b></a></item></root>','item')
    with pytest.raises(ValueError):conn.parse_xml_records(b'<!DOCTYPE catalog><catalog><item><title>A</title></item></catalog>','item')
    with pytest.raises(ValueError):conn.parse_xml_records(b'<root/>','item')
    with pytest.raises(ValueError):conn.parse_xml_records(b' ', '../item')
    with pytest.raises(ValueError):conn.parse_xml_records(b'x'*(conn.MAX_XML_BYTES+1),'item')


def test_api_public_https_guard(env):
    site=env()
    for u in ['http://api.example.com/a','https://admin:pw@api.example.com/a',
              'https://127.0.0.1/items','https://api.example.com/a?access_token=foo',
              'https://api.example.com:8443/a','https://api.example.com/a#frag']:
        with pytest.raises(ValueError):conn.create_api_source(site,{'url':u})
    with pytest.raises(ValueError):conn.create_api_source(site,{'url':'https://api.example.com/items','page_size':500})
    with pytest.raises(ValueError):conn.create_api_source(site,{'url':'https://api.example.com/items','items_path':'data[0]'})


def fake_http(monkeypatch):
    requested=[]
    def fetch(url,**kwargs):
        requested.append((url,kwargs))
        q=parse_qs(urlsplit(url).query)
        page=int(q['page'][0]);assert q['per_page']==['2']
        if page==1:rows=[{'title':'Alpha','brand':'Brand A','id':100},{'title':'Beta','brand':'Brand B','id':101}]
        else:rows=[]
        return Fetched(url,200,{'content-type':'application/json; charset=utf-8'},
                       json.dumps({'data':{'items':rows}}).encode())
    monkeypatch.setattr(conn.safe_http,'fetch',fetch)
    return requested


def create(env):
    site=env()
    source=conn.create_api_source(site,{'url':'https://api.example.com/v1/products?lang=en',
            'items_path':'data.items','page_size':2,'max_pages':10,
            'mapping':{'title':'title','brand':'brand'}})
    return site,source['id']


def test_api_scan_page_cursor_and_reuse_existing_batch(env,monkeypatch):
    site,source=create(env);calls=fake_http(monkeypatch)
    first=conn.scan_page(site,source)
    assert first['count']==2 and first['state']=='ready'
    second=conn.scan_page(site,source)
    assert second['count']==0 and second['state']=='completed'
    assert len(calls)==2
    assert all(x[1]['redirects']==0 for x in calls)
    assert conn.run_scan(site,source)['staged_count']==2
    assert len(conn.status(site,source)['batches'])==2
    bid=first['batch_id'];res=v8_workflows.commit(site,bid)
    assert res['created']==2
    assert v8_workflows.commit(site,bid)['existing']


def test_api_worker_enqueue_dedup_and_cancel(env):
    site,source=create(env)
    first=conn.submit_scan(site,source,'test-worker')
    again=conn.submit_scan(site,source,'test-worker')
    assert again['existing'] and again['id']==first['id']
    assert db.one('SELECT kind FROM build_jobs WHERE id=?',(first['id'],))['kind']=='api_scan'
    jobs.cancel(site,first['id'])
    assert db.one('SELECT state FROM build_jobs WHERE id=?',(first['id'],))['state']=='cancelled'


def test_api_retry_safe_after_partial_stage(env,monkeypatch):
    site,source=create(env)
    origin=f'api:{source}:page:1'
    pending=v8_workflows.stage(site,[{'title':'Alpha','brand':'Brand A'}],{'title':'title','brand':'brand'},'product',origin)
    def should_not_request(*a,**k):raise AssertionError('Duplicate HTTP download')
    monkeypatch.setattr(conn.safe_http,'fetch',should_not_request)
    first=conn.scan_page(site,source)
    assert first['batch_id']==pending['batch_id']
    assert conn.status(site,source)['staged_count']==1
    assert len(db.rows('SELECT id FROM v8_import_batches WHERE origin=?',(origin,)))==1


def test_api_limit_reached_requires_manual_attention(env,monkeypatch):
    site=env()
    s=conn.create_api_source(site,{'url':'https://api.example.com/products','items_path':'data.items','page_size':2,'max_pages':1,'mapping':{'title':'title'}})['id']
    fake_http(monkeypatch)
    state=conn.run_scan(site,s)
    assert state['state']=='limit_reached'
    assert conn.submit_scan(site,s,'test')['state']=='limit_reached'


def test_api_network_failure_does_not_move_cursor(env,monkeypatch):
    site,source=create(env)
    monkeypatch.setattr(conn.safe_http,'fetch',lambda *a,**k:(_ for _ in ()).throw(ConnectionError('network lost')))
    with pytest.raises(ConnectionError):conn.scan_page(site,source)
    assert conn.status(site,source)['next_page']==1
    assert not conn.status(site,source)['batches']


def test_api_json_schema_and_bad_mime(env,monkeypatch):
    site,source=create(env)
    monkeypatch.setattr(conn.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{'content-type':'text/html'},b'hi'))
    with pytest.raises(ValueError):conn.scan_page(site,source)
    monkeypatch.setattr(conn.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{'content-type':'application/json'},b'{"data":{"items":{}}}'))
    with pytest.raises(ValueError):conn.scan_page(site,source)


def test_api_site_isolation(env,monkeypatch):
    site,source=create(env)
    other=env()
    with pytest.raises(HTTPException) as err:conn.status(other,source)
    assert err.value.status_code==404
    with pytest.raises(HTTPException):conn.submit_scan(other,source,'someone')


def test_xml_and_api_authenticated_ui(env):
    site=env()
    with TestClient(app) as c:
        route=f'/api/connectors/sites/{site}/api-sources'
        assert c.get(route).status_code==401
        assert c.post(route,json={'url':'https://api.example.com/items'}).status_code==401
        assert c.post('/api/auth/login',json={'password':'v84-unit-password-strong'}).status_code==200
        c.headers['X-SiteForge-CSRF']=c.get('/api/auth/me').json()['csrf']
        source=c.post(route,json={'url':'https://api.example.com/items','items_path':'items','page_size':2})
        assert source.status_code==200,source.text
        assert c.get(route).json()['items'][0]['id']==source.json()['id']
        xml=b'<catalog><item><title>ABC Model</title></item></catalog>'
        posted=c.post(f'/api/connectors/sites/{site}/xml/staging',
                      files={'file':('products.xml',xml,'application/xml')},data={'record_tag':'item','content_type':'product'})
        assert posted.status_code==200,posted.text
        assert posted.json()['mapped']==1
        assert 'apiCreate' in c.get('/production-v8').text
        assert c.post(route,json={'url':'http://localhost/test'}).status_code==400


def test_worker_api_task_finish_and_cancel(env,monkeypatch):
    from app import worker
    site,source=create(env);fake_http(monkeypatch)
    queued=conn.submit_scan(site,source,'tester')
    def execute(s,opts):
        assert s==site
        return conn.run_scan(s,opts['source_id'])
    assert worker.run_once(executor=execute)==queued['id']
    job=db.one('SELECT state,result_json FROM build_jobs WHERE id=?',(queued['id'],))
    assert job['state']=='done'
    assert json.loads(job['result_json'])['staged_count']==2
    assert conn.status(site,source)['state']=='completed'


def test_api_job_runner_dispatch_to_api_scan(env,monkeypatch):
    from app import job_runner
    from sys import argv
    site,source=create(env);fake_http(monkeypatch)
    queued=conn.submit_scan(site,source,'tester')
    monkeypatch.setattr(job_runner.sys,'argv',['worker',queued['id']])
    assert job_runner.main()==0
    result=json.loads((db.DATA/'job_results'/(queued['id']+'.json')).read_text())
    assert result['ok'] and result['result']['staged_count']==2