cms-SiteForge/tests/test_v8_workflows.py

from app.version import VERSION
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']==VERSION


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