浅念
8cea7dee20
- SiteForge Enterprise V8.10.0-dev.1 upstream source (210 tests) - Fix: v8_workflows._mapped_input now reads both 类型/分类 columns so WXR imports keep post/product types instead of collapsing to page - Docs: trade-pro theme port spec (R3) and implementation plan Co-Authored-By: Claude Code <noreply@anthropic.com>
186 lines
10 KiB
Python
186 lines
10 KiB
Python
"""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':'<p>Safe source text</p>'},
|
|
'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)
|