浅念
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>
205 lines
11 KiB
Python
205 lines
11 KiB
Python
import json
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from app import db, v8_workflows as workflow, v8_templates as templating
|
|
from app.core import save_entry
|
|
from app.main import app
|
|
|
|
@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','dev-v8-admin-password-with-sufficient-entropy')
|
|
monkeypatch.setenv('SITEFORGE_SESSION_SECRET','v8-testing-session-secret-that-is-not-production')
|
|
db.init(); workflow.setup();
|
|
from app import cms_pipeline
|
|
cms_pipeline.setup();
|
|
from app import theme_store
|
|
theme_store.setup()
|
|
def site(name):
|
|
from app.core import uid
|
|
i=uid();t=db.now()
|
|
db.insert('''INSERT INTO sites(id,name,domain,tagline,theme,accent,logo,mode,settings_json,created_at,updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?)''',(i,name,'','', 'corporate','#2563eb','','new','{}',t,t))
|
|
save_entry(i,{'type':'page','title':'首页','route':'/','seo_title':'Example Home','meta':{'blocks':[]}})
|
|
return i
|
|
yield site
|
|
|
|
|
|
def rows():return [{'车型名称':'Model A','品牌':'Maker A'},{'车型名称':'Model B','品牌':'Maker B'}]
|
|
|
|
|
|
def test_stage_commit_and_rollback(env):
|
|
sid=env('Cars')
|
|
s=workflow.stage(sid,rows(),kind='vehicle')
|
|
assert s['mapped']==2 and not s['conflicts']
|
|
assert db.one('SELECT COUNT(*) n FROM entries WHERE site_id=?',(sid,))['n']==1
|
|
commit=workflow.commit(sid,s['batch_id'])
|
|
assert commit['created']==2 and commit['state']=='committed'
|
|
assert workflow.commit(sid,s['batch_id'])['existing']
|
|
rollback=workflow.rollback(sid,s['batch_id'])
|
|
assert rollback['restored']==2
|
|
assert workflow.rollback(sid,s['batch_id'])['existing']
|
|
assert db.one('SELECT COUNT(*) n FROM entries WHERE site_id=?',(sid,))['n']==1
|
|
|
|
|
|
def test_replace_restore_seo_and_body(env):
|
|
sid=env('Factory')
|
|
original={'type':'product','title':'Original','route':'/product/test/','seo_title':'Old SEO',
|
|
'body':'Old body'}
|
|
eid=save_entry(sid,original)
|
|
s=workflow.stage(sid,[{'title':'New title','route':'/product/test/','body':'New body','seo_title':'New SEO'}],
|
|
mapping={'title':'title','route':'route','body':'body','seo_title':'seo_title'})
|
|
assert s['conflicts']==['/product/test/']
|
|
assert workflow.commit(sid,s['batch_id'],'replace')['updated']==1
|
|
assert db.one('SELECT seo_title FROM entries WHERE id=?',(eid,))['seo_title']=='New SEO'
|
|
workflow.rollback(sid,s['batch_id'])
|
|
restored=db.one('SELECT title,seo_title,body FROM entries WHERE id=?',(eid,))
|
|
assert restored=={'title':'Original','seo_title':'Old SEO','body':'<p>Old body</p>'} or restored['seo_title']=='Old SEO' and restored['title']=='Original'
|
|
|
|
|
|
def test_rollback_refuses_clobbering_manual_edits(env):
|
|
sid=env('Brand'); s=workflow.stage(sid,rows()[:1],kind='vehicle')
|
|
workflow.commit(sid,s['batch_id'])
|
|
row=db.one("SELECT id FROM entries WHERE site_id=? AND type='vehicle'",(sid,))
|
|
db.execute('UPDATE entries SET title=? WHERE id=?',('MANUAL EDIT',row['id']))
|
|
with pytest.raises(HTTPException) as exc:workflow.rollback(sid,s['batch_id'])
|
|
assert exc.value.status_code==409
|
|
assert db.one('SELECT title FROM entries WHERE id=?',(row['id'],))['title']=='MANUAL EDIT'
|
|
|
|
|
|
def test_rollback_conflict_no_partial_side_effects(env):
|
|
sid=env('Batch'); s=workflow.stage(sid,rows(),kind='vehicle');workflow.commit(sid,s['batch_id'])
|
|
obj=db.rows("SELECT id FROM entries WHERE site_id=? AND type='vehicle' ORDER BY title",(sid,))
|
|
db.execute('UPDATE entries SET title=? WHERE id=?',('Changed',obj[1]['id']))
|
|
with pytest.raises(HTTPException):workflow.rollback(sid,s['batch_id'])
|
|
assert db.one("SELECT COUNT(*) n FROM entries WHERE site_id=? AND type='vehicle'",(sid,))['n']==2
|
|
|
|
|
|
def test_source_keys_must_be_unique(env):
|
|
sid=env('A')
|
|
with pytest.raises(ValueError,match='主键重复'):
|
|
workflow.stage(sid,rows(),kind='vehicle',source_keys=['same','same'])
|
|
|
|
|
|
def test_wxr_staging_and_xxe_rejection(env):
|
|
sid=env('WP')
|
|
raw=b'''<rss version="2.0" xmlns:wp="http://wordpress.org/export/1.2/"
|
|
xmlns:content="http://purl.org/rss/1.0/modules/content/"><channel><item>
|
|
<title>WordPress Page</title><link>https://old.example.org/about/</link>
|
|
<wp:post_id>42</wp:post_id><wp:post_type>page</wp:post_type><wp:status>publish</wp:status>
|
|
<content:encoded><![CDATA[<p>Hello</p>]]></content:encoded>
|
|
</item></channel></rss>'''
|
|
rows,ids,warnings=workflow.parse_wxr(raw)
|
|
assert rows[0]['路径']=='/about/' and ids==['wordpress:42'] and not warnings
|
|
payload=workflow.stage(sid,rows,{'title':'标题','route':'路径','body':'正文','seo_title':'SEO标题','seo_description':'SEO描述'},
|
|
kind='page',origin='wordpress-wxr',source_keys=ids)
|
|
assert workflow.commit(sid,payload['batch_id'])['created']==1
|
|
assert db.one('SELECT title FROM entries WHERE route=? AND site_id=?',('/about/',sid))['title']=='WordPress Page'
|
|
with pytest.raises(Exception):workflow.parse_wxr(b'<!DOCTYPE rss [<!ENTITY xxe SYSTEM "file:///etc/passwd">]><rss>&xxe;</rss>')
|
|
|
|
|
|
def test_whole_site_blueprint_no_business_copy(env):
|
|
source=env('Source'); dest=env('Destination')
|
|
from app import blueprints
|
|
page=blueprints.new_block('hero',{'title':'CONFIDENTIAL BRAND','description':'Secret marketing copy',
|
|
'image':'https://customer.example.org/private.png'})
|
|
save_entry(source,{'type':'page','title':'About Brand','route':'/about/',
|
|
'seo_title':'Confidential SEO','meta':{'blocks':[page]}})
|
|
document=templating.export_template(source,'Landing')
|
|
assert 'CONFIDENTIAL BRAND' not in json.dumps(document)
|
|
assert 'private.png' not in json.dumps(document)
|
|
assert 'Confidential SEO' not in json.dumps(document)
|
|
proposed=templating.preview_application(dest,document)
|
|
assert '/' in proposed['existing']
|
|
with pytest.raises(HTTPException):templating.apply_template(dest,document)
|
|
done=templating.apply_template(dest,document,overwrite_pages=True)
|
|
assert done['content_preserved'] is True
|
|
assert db.one('SELECT status,robots FROM entries WHERE site_id=? AND route=?',(dest,'/about/'))=={'status':'draft','robots':'noindex,follow'}
|
|
assert db.one('SELECT seo_title FROM entries WHERE site_id=? AND route=?',(dest,'/'))['seo_title']=='Example Home'
|
|
|
|
|
|
def test_template_invalid_and_duplicate_route(env):
|
|
sid=env('X')
|
|
doc=templating.export_template(sid,'Example')
|
|
doc['pages'].append(doc['pages'][0])
|
|
with pytest.raises(ValueError,match='重复'):templating.validate_template(doc)
|
|
doc['pages'].pop()
|
|
doc['theme']='untrusted-script-template'
|
|
with pytest.raises(ValueError,match='不可用'):templating.validate_template(doc)
|
|
|
|
|
|
def test_v8_api_authenticated_and_browser_page(env):
|
|
with TestClient(app) as client:
|
|
assert client.get('/production-v8',follow_redirects=False).status_code==303
|
|
assert client.post('/api/auth/login',json={'password':'dev-v8-admin-password-with-sufficient-entropy'}).status_code==200
|
|
client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf']
|
|
sid=env('API')
|
|
html=client.get('/production-v8')
|
|
assert html.status_code==200 and '网站生产控制台' in html.text
|
|
stage=client.post(f'/api/factory/sites/{sid}/staging',json={'rows':rows(),'content_type':'vehicle'})
|
|
assert stage.status_code==200,stage.text
|
|
bid=stage.json()['batch_id']
|
|
assert client.post(f'/api/factory/sites/{sid}/staging/{bid}/commit',json={}).status_code==200
|
|
assert client.get(f'/api/factory/sites/{sid}/staging').json()['items'][0]['state']=='committed'
|
|
snap=client.post(f'/api/factory/sites/{sid}/whole-site-templates',json={'name':'Test'})
|
|
assert snap.status_code==200,snap.text
|
|
assert client.get('/api/health').json()['version']=='8.10.0-dev.1'
|
|
|
|
|
|
def test_wxr_preserves_post_type(env):
|
|
sid=env('WXR post')
|
|
rows=[{'标题':'News','路径':'/news/news-a/','正文':'A','类型':'post'},
|
|
{'标题':'About','路径':'/about/','正文':'B','类型':'page'}]
|
|
s=workflow.stage(sid,rows,{'title':'标题','route':'路径','body':'正文'},'mixed','wordpress-wxr')
|
|
workflow.commit(sid,s['batch_id'])
|
|
assert db.one('SELECT type FROM entries WHERE site_id=? AND route=?',(sid,'/news/news-a/'))['type']=='post'
|
|
assert db.one('SELECT type FROM entries WHERE site_id=? AND route=?',(sid,'/about/'))['type']=='page'
|
|
|
|
|
|
def test_wordpress_complex_builder_creates_release_review(env):
|
|
sid=env('Complex WP')
|
|
raw=b'''<rss xmlns:wp="http://wordpress.org/export/1.2/">
|
|
<channel><item><title>Complex</title><link>https://example.org/a/</link>
|
|
<wp:post_type>page</wp:post_type><wp:status>publish</wp:status>
|
|
<wp:postmeta><wp:meta_key>_elementor_data</wp:meta_key>
|
|
<wp:meta_value><![CDATA[[{"elType":"section"}]]]></wp:meta_value></wp:postmeta>
|
|
</item></channel></rss>'''
|
|
rows,keys,warnings=workflow.parse_wxr(raw)
|
|
assert warnings
|
|
batch=workflow.stage(sid,rows,{'title':'标题','route':'路径','body':'正文'},'mixed','wordpress-wxr',keys)
|
|
db.execute('UPDATE v8_import_batches SET warnings_json=? WHERE id=?',
|
|
(json.dumps(warnings,ensure_ascii=False),batch['batch_id']))
|
|
workflow.commit(sid,batch['batch_id'])
|
|
assert db.one("SELECT COUNT(*) n FROM cms_import_reviews WHERE site_id=? AND state='pending'",(sid,))['n']==1
|
|
|
|
|
|
def test_csv_file_staged_then_committed_via_api(env):
|
|
sid=env('Excel Import')
|
|
with TestClient(app) as client:
|
|
assert client.post('/api/auth/login',json={'password':'dev-v8-admin-password-with-sufficient-entropy'}).status_code==200
|
|
client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf']
|
|
result=client.post(f'/api/factory/sites/{sid}/staging/file',
|
|
files={'file':('products.csv','车型名称,品牌\nMy New Vehicle,Acme\n'.encode(),'text/csv')},
|
|
data={'content_type':'vehicle'})
|
|
assert result.status_code==200,result.text
|
|
assert result.json()['mapped']==1
|
|
bid=result.json()['batch_id']
|
|
result=client.post(f'/api/factory/sites/{sid}/staging/{bid}/commit',json={})
|
|
assert result.status_code==200,result.text
|
|
assert db.one('SELECT title FROM entries WHERE site_id=? AND type=?',(sid,'vehicle'))['title']=='My New Vehicle'
|
|
|
|
|
|
def test_wordpress_rollback_resolves_its_own_warnings(env):
|
|
sid=env('Review rollback')
|
|
records=[{'标题':'Legacy','路径':'/legacy/','正文':'Text','类型':'page'}]
|
|
batch=workflow.stage(sid,records,{'title':'标题','route':'路径','body':'正文'},'mixed','wordpress-wxr')
|
|
db.execute('UPDATE v8_import_batches SET warnings_json=? WHERE id=?',
|
|
(json.dumps([{'row':1,'issue':'Unsupported component'}]),batch['batch_id']))
|
|
workflow.commit(sid,batch['batch_id'])
|
|
from app import cms_pipeline
|
|
assert cms_pipeline.open_reviews(sid)==1
|
|
workflow.rollback(sid,batch['batch_id'])
|
|
assert cms_pipeline.open_reviews(sid)==0
|