cms-SiteForge/tests/test_wp_migration_v82.py

"""Deterministic tests for V8.2 sitemap, media, durable Worker, and release gates."""
import json
from pathlib import Path
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app import db, enterprise, migration_audit, v8_workflows as wf, wp_migration_v81 as wp
from app import wp_migration_v82 as ext, jobs, worker
from app.core import uid
from app.main import app
from app.safe_http import Fetched


@pytest.fixture
def sandbox(tmp_path,monkeypatch):
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
    monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','test-v8.2-admin-password-long')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','test-v8.2-session-secret-strong-and-extra-words')
    db.init();enterprise.setup();migration_audit.setup();wf.setup();wp.setup()
    from app import cms_pipeline
    cms_pipeline.setup()
    def create_site(mode='migration'):
        sid=uid();now=db.now()
        db.insert('INSERT INTO sites(id,name,theme,mode,created_at,updated_at) VALUES(?,?,?,?,?,?)',
                  (sid,'Test site','corporate',mode,now,now))
        return sid
    return create_site


def _record(item_id,path,with_media=False):
    url='https://example.org/wp-content/uploads/item.png'
    body='<p>Product</p>'+('<img src="'+url+'" srcset="'+url+' 1x">' if with_media else '')
    return {'id':item_id,'link':'https://example.org'+path,'title':{'rendered':f'Page {item_id}'},
            'excerpt':{'rendered':'Intro'},'content':{'rendered':body},
            'yoast_head_json':{'title':'SEO','description':'Meta','canonical':'https://example.org'+path},
            '_embedded':{'wp:featuredmedia':[{'source_url':url}] if with_media else []}}


def _scan(monkeypatch,session,site_id,media=False):
    monkeypatch.setattr(wp,'_fetch',lambda origin,phase,page: [_record(1,'/about/',media)] if phase=='pages' else [])
    wp.advance(site_id,session);wp.advance(site_id,session)


def test_sitemap_index_detects_missing_original_urls(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    files={
      'wp-sitemap.xml':b'<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"><sitemap><loc>https://example.org/child.xml</loc></sitemap></sitemapindex>',
      'child.xml':b'<urlset><url><loc>https://example.org/about/</loc></url><url><loc>https://example.org/missing/</loc></url></urlset>'}
    monkeypatch.setattr(ext.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{'content-type':'application/xml'},files[url.rsplit('/',1)[-1]]))
    result=ext.discover(site,ses['id']);assert result['discovered']==2
    _scan(monkeypatch,ses['id'],site)
    assert ext.sitemap_status(site,ses['id'])['missing_count']==1
    final=wp.finalize(site,ses['id'])
    assert final['session']['state']=='completed_with_warnings'
    from app.cms_pipeline import open_reviews
    assert open_reviews(site)>=1
    assert any('missing' in x['issue'] for x in db.rows('SELECT issue FROM cms_import_reviews WHERE site_id=?',(site,)))


def test_sitemap_rejects_xxe_and_external_host(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    monkeypatch.setattr(ext.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{},
          b'<!DOCTYPE x [<!ENTITY payload SYSTEM "file:///etc/passwd">]><urlset/>'))
    with pytest.raises(ValueError):ext.discover(site,ses['id'])
    with pytest.raises(ValueError):ext.discover(site,ses['id'],'https://attacker.invalid/sitemap.xml')
    with pytest.raises(ValueError):ext.discover(site,ses['id'],'http://example.org/wp-sitemap.xml')


def test_sitemap_fails_closed_on_decompression_bomb(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    import gzip
    compressed=gzip.compress(b'<urlset>'+b' '*3_000_200+b'</urlset>')
    monkeypatch.setattr(ext.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{},compressed))
    with pytest.raises(ValueError,match='解压'):ext.discover(site,ses['id'])


def test_media_localized_into_site_scoped_files(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200);_scan(monkeypatch,ses['id'],site,media=True)
    image=b'\x89PNG\r\n\x1a\n'+b'\x00'*80
    monkeypatch.setattr(ext.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{'content-type':'image/png'},image))
    report=ext.localize_media(site,ses['id'],20)
    assert report['localized']==1 and report['remaining']==0
    batch=wp.details(site,ses['id'])['batches'][0]['batch_id']
    row=db.one('SELECT content_json FROM v8_import_items WHERE batch_id=?',(batch,))
    content=json.loads(row['content_json']);assert content['cover'].startswith('/media/wp-')
    assert 'srcset' not in content['body']
    assert 'https://example.org/wp-content/uploads/item.png' not in content['body']
    assert (db.DATA/'media'/site/content['cover'].split('/')[-1]).is_file()
    assert ext.localize_media(site,ses['id'],20)['attempted']==0
    # Do not accidentally copy source media between different projects.
    other=sandbox();assert not (db.DATA/'media'/other).exists()


def test_media_cdn_requires_explicit_review(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    _scan(monkeypatch,ses['id'],site,media=True)
    # A cross-site URL is rejected without a network request.
    monkeypatch.setattr(ext.safe_http,'fetch',lambda *a,**kw:pytest.fail('unexpected outbound connection'))
    batch=wp.details(site,ses['id'])['batches'][0]['batch_id']
    obj=json.loads(db.one('SELECT content_json FROM v8_import_items WHERE batch_id=?',(batch,))['content_json'])
    obj['cover']='https://cdn.example.net/photo.png'
    obj['body']='<p>No remote image</p>'
    db.execute('UPDATE v8_import_items SET content_json=? WHERE batch_id=?',(json.dumps(obj),batch))
    report=ext.localize_media(site,ses['id'],20)
    assert report['failed'] and '单独授权' in report['failed'][0]['reason']
    assert report['requires_review']


def test_migration_job_idempotency_and_capacity(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    first=ext.submit_scan(site,ses['id'],'test','wp-migration:unique-id')
    duplicate=ext.submit_scan(site,ses['id'],'test','wp-migration:unique-id')
    assert first['id']==duplicate['id'] and duplicate['existing']
    other=sandbox(); other_session=wp.create(other,'https://example.org/',200)
    monkeypatch.setenv('SITEFORGE_MAX_PENDING_JOBS','1')
    with pytest.raises(jobs.JobCapacityError):ext.submit_scan(other,other_session['id'],'test')


def test_worker_dispatch_and_partial_scan_resume(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    monkeypatch.setattr(wp,'_fetch',lambda origin,phase,page: [_record(3,'/services/')] if phase=='pages' else [])
    job=ext.submit_scan(site,ses['id'],'test')
    # Independent process uses the same dispatch path; invoke the scanner directly with mocked fixture.
    out=ext.run_scan(site,ses['id'],job['id'])
    assert out['state']=='ready'
    assert wp.details(site,ses['id'])['fetched']==1
    assert wp.finalize(site,ses['id'])['baseline_saved']==1


def test_authenticated_sitemap_and_worker_endpoints(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    xml=b'<urlset><url><loc>https://example.org/about/</loc></url></urlset>'
    monkeypatch.setattr(ext.safe_http,'fetch',lambda url,**kw:Fetched(url,200,{},xml))
    endpoint=f'/api/migrations/sites/{site}/wordpress/sessions/{ses["id"]}'
    with TestClient(app) as client:
        assert client.post(endpoint+'/scan',json={}).status_code==401
        assert client.post('/api/auth/login',json={'password':'test-v8.2-admin-password-long'}).status_code==200
        client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf']
        assert client.post(endpoint+'/sitemap',json={}).json()['discovered']==1
        job=client.post(endpoint+'/scan',json={})
        assert job.status_code==200,job.text
        assert client.post(endpoint+'/scan',json={}).json()['id']==job.json()['id']
        assert client.get(endpoint+'/sitemap').json()['discovered']==1
        assert 'wpWorker' in client.get('/production-v8').text


def test_worker_subprocess_executes_wordpress_job_kind(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    job=ext.submit_scan(site,ses['id'],'test')
    # Simulate another process finishing the scan after enqueue. The spawned
    # subprocess must dispatch `wordpress_scan`, not accidentally build HTML.
    db.execute("UPDATE wp_migration_sessions SET state='ready' WHERE id=?",(ses['id'],))
    monkeypatch.setattr(worker,'running',True)
    monkeypatch.setattr(worker,'WORKER_ID','testworker')
    assert worker.run_once()==job['id']
    status=db.one('SELECT state,result_json,kind FROM build_jobs WHERE id=?',(job['id'],))
    assert status['kind']=='wordpress_scan' and status['state']=='done'
    assert json.loads(status['result_json'])['state']=='ready'


def test_remote_media_stays_in_manual_review_if_not_localized(sandbox,monkeypatch):
    site=sandbox();ses=wp.create(site,'https://example.org/',200)
    _scan(monkeypatch,ses['id'],site,media=True)
    issues=ext._findings(site,ses['id'])
    assert any('外部图片' in x for x in issues)
    assert wp.finalize(site,ses['id'])['requires_review']