"""V8.3 deterministic WordPress taxonomy and media catalog integration tests.""" import json import pytest from fastapi import HTTPException from fastapi.testclient import TestClient from app import db, enterprise, migration_audit, v8_workflows, wp_migration_v81 as wp from app import wp_migration_v83 as catalog from app.core import uid from app.main import app from app.safe_http import Fetched @pytest.fixture def site_env(tmp_path,monkeypatch): monkeypatch.setattr(db,'DATA',tmp_path) monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3') monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','v83-test-password-with-good-entropy') monkeypatch.setenv('SITEFORGE_SESSION_SECRET','v83-test-session-secret-long-enough-for-dev') db.init();enterprise.setup();migration_audit.setup();v8_workflows.setup();wp.setup() from app import cms_pipeline cms_pipeline.setup() def new(): sid=uid();now=db.now() db.insert('INSERT INTO sites(id,name,theme,mode,created_at,updated_at) VALUES(?,?,?,?,?,?)', (sid,'Migrating WordPress','automotive','migration',now,now)) return sid return new def post(): return {'id':101,'link':'https://source.example/blog/post/', 'title':{'rendered':'Source Post'},'content':{'rendered':'

Safe source text

'}, 'excerpt':{'rendered':'Excerpt'},'categories':[10,22],'tags':[35], 'featured_media':77, 'acf':{'specification':'example'}, 'yoast_head_json':{'title':'Title','description':'Description','canonical':'https://source.example/blog/post/'}, '_embedded':{'wp:featuredmedia':[{'source_url':'https://source.example/wp-content/uploads/a.png'}]}} def install_source(monkeypatch): monkeypatch.setattr(wp,'_fetch',lambda origin,phase,page:[post()] if phase=='posts' else []) def sync_fixtures(monkeypatch): from app.safe_http import Fetched def fake(url,**kwargs): if '/categories?' in url:body=[{'id':10,'name':'News','slug':'news','parent':0},{'id':22,'name':'Technology','slug':'technology','parent':10}] elif '/tags?' in url:body=[{'id':35,'name':'Electric','slug':'electric'}] elif '/media?' in url:body=[{'id':77,'title':{'rendered':'Image A'},'source_url':'https://source.example/wp-content/uploads/a.png','mime_type':'image/png','alt_text':'Cover','media_details':{'width':100,'height':200}}] else:raise AssertionError('Unexpected remote path: '+url) return Fetched(url,200,{'content-type':'application/json'},json.dumps(body).encode()) monkeypatch.setattr(catalog.safe_http,'fetch',fake) def test_taxonomy_catalog_staged_persisted_and_committed(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',300) install_source(monkeypatch) wp.advance(site,ses['id']);wp.advance(site,ses['id']) # Media, category and tag ID references are preserved before catalog sync. refs=db.rows('SELECT resource,target_id FROM wp_catalog_refs WHERE session_id=?',(ses['id'],)) assert {(r['resource'],r['target_id']) for r in refs}=={('categories',10),('categories',22),('tags',35),('media',77)} original=catalog.findings(site,ses['id']) assert any('缺少 WordPress categories' in msg for msg in original) sync_fixtures(monkeypatch) for kind in ('categories','tags','media'): result=catalog.sync_page(site,ses['id'],kind) assert result['resources'][kind]['done'] staged_id=db.one('SELECT batch_id FROM wp_migration_pages WHERE session_id=? AND phase=?',(ses['id'],'posts'))['batch_id'] row=db.one('SELECT content_json FROM v8_import_items WHERE batch_id=?',(staged_id,)) payload=json.loads(row['content_json']) assert payload['meta']['fields']['category']=='News' assert payload['meta']['fields']['tags']=='Electric' assert payload['meta']['wp_source']['category_ids']==[10,22] assert payload['meta']['wp_source']['acf_snapshot']=={'specification':'example'} assert payload['meta']['wp_source']['featured_media_original'].endswith('/a.png') warnings=catalog.findings(site,ses['id']) assert not any('缺少 WordPress' in issue for issue in warnings) assert any('ACF' in issue for issue in warnings) completed=wp.finalize(site,ses['id']) assert completed['requires_review'] and completed['baseline_saved']==1 entry=db.one('SELECT * FROM entries WHERE site_id=? AND route=?',(site,'/blog/post/')) meta=json.loads(entry['meta_json']) assert meta['fields']['tags']=='Electric' assert meta['wp_source']['wp_id']==101 def test_catalog_sync_idempotent_and_checkpointed(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) sync_fixtures(monkeypatch) first=catalog.sync_page(site,ses['id'],'tags') again=catalog.sync_page(site,ses['id'],'tags') assert first['resources']['tags']['fetched']==1==again['resources']['tags']['fetched'] assert db.one('SELECT COUNT(*) n FROM wp_catalog_items WHERE session_id=?',(ses['id'],))['n']==1 assert first['resources']['tags']['next_page']==2 def test_catalog_preserves_cursor_on_network_failure(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) def fail(url,**kwargs):raise RuntimeError('Network disconnected') monkeypatch.setattr(catalog.safe_http,'fetch',fail) with pytest.raises(RuntimeError):catalog.sync_page(site,ses['id'],'categories') assert catalog.catalog_status(site,ses['id'])['resources']['categories']['next_page']==1 assert db.one('SELECT COUNT(*) n FROM wp_catalog_items WHERE session_id=?',(ses['id'],))['n']==0 def test_catalog_does_not_allow_bad_resources_or_id_rows(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) with pytest.raises(ValueError):catalog.sync_page(site,ses['id'],'../wp-users') with pytest.raises(ValueError):catalog._record('categories',{'id':True,'name':'Wrong'}) with pytest.raises(ValueError):catalog._record('categories',{'id':'1','name':'Wrong'}) assert catalog._safe_id_list([0,True,12,12,'12',40])==[12,40] def test_missing_term_and_media_references_require_review(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) install_source(monkeypatch);wp.advance(site,ses['id']);wp.advance(site,ses['id']) findings=catalog.findings(site,ses['id']) assert any('media' not in f.lower() and '分类' not in f and 'categories' in f for f in findings) assert any('特色图片媒体记录' in f for f in findings) result=wp.finalize(site,ses['id']) assert result['requires_review'] reviews=db.rows('SELECT * FROM cms_import_reviews WHERE site_id=?',(site,)) assert any('特色图片' in review['issue'] for review in reviews) def test_access_control_and_ui_catalog_buttons(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) sync_fixtures(monkeypatch) base=f'/api/migrations/sites/{site}/wordpress/sessions/{ses["id"]}/catalog' with TestClient(app) as client: assert client.get(base).status_code==401 assert client.post(base+'/tags/next',json={}).status_code==401 assert client.post('/api/auth/login',json={'password':'v83-test-password-with-good-entropy'}).status_code==200 client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf'] assert client.get(base).status_code==200 result=client.post(base+'/tags/next',json={}) assert result.status_code==200,result.text assert result.json()['term_counts']['tags']==1 assert 'wpCatalogSync' in client.get('/production-v8').text def test_site_and_session_are_scoped(site_env,monkeypatch): one=site_env();two=site_env();ses=wp.create(one,'https://source.example/',200) with pytest.raises(HTTPException) as exc:catalog.catalog_status(two,ses['id']) assert exc.value.status_code==404 sync_fixtures(monkeypatch) catalog.sync_page(one,ses['id'],'categories') assert db.one('SELECT COUNT(*) n FROM wp_catalog_items WHERE session_id=?',(ses['id'],))['n']==2 def test_catalog_not_mutable_after_finalization(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) install_source(monkeypatch);wp.advance(site,ses['id']);wp.advance(site,ses['id']);wp.finalize(site,ses['id']) sync_fixtures(monkeypatch) with pytest.raises(HTTPException) as exc:catalog.sync_page(site,ses['id'],'tags') assert exc.value.status_code==409 def test_taxonomy_archives_preserved_in_original_url_audit(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) install_source(monkeypatch);wp.advance(site,ses['id']);wp.advance(site,ses['id']) from app.safe_http import Fetched def fetch(url,**kwargs): if '/categories?' in url: data=[{'id':10,'name':'News','slug':'news','parent':0,'link':'https://source.example/category/news/'}] elif '/tags?' in url: data=[{'id':35,'name':'Electric','slug':'electric','link':'https://source.example/tag/electric/'}] else:data=[] return Fetched(url,200,{'content-type':'application/json'},json.dumps(data).encode()) monkeypatch.setattr(catalog.safe_http,'fetch',fetch) catalog.sync_page(site,ses['id'],'categories') catalog.sync_page(site,ses['id'],'tags') result=wp.finalize(site,ses['id']) assert result['baseline_saved']==3 # article + both archives urls={r['route'] for r in db.rows('SELECT route FROM wp_migration_urls WHERE session_id=?',(ses['id'],))} assert urls=={'/blog/post/','/category/news/','/tag/electric/'} assert catalog.register_archive_baseline(site,ses['id'])['taxonomy_archives_added']==0 def test_missing_parent_and_empty_term_name_flagged(site_env,monkeypatch): site=site_env();ses=wp.create(site,'https://source.example/',200) db.insert('''INSERT INTO wp_catalog_items(session_id,resource,wp_id,name,slug,parent_id,source_url,meta_json) VALUES(?,?,?,?,?,?,?,?)''',(ses['id'],'categories',88,'','',300,'','{}')) notes=catalog.findings(site,ses['id']) assert any('上级分类' in note for note in notes) assert any('缺少名称' in note for note in notes)