cms-SiteForge/tests/test_wp_migration_v81.py

"""Deterministic WP REST fixture tests, with no real external network."""
import json
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app import db, wp_migration_v81 as wp, v8_workflows as w, enterprise, migration_audit
from app.core import save_entry, uid
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','test-v8-admin-strong-password-12345')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','test-v8-super-secret-long-key-12345')
    db.init(); enterprise.setup(); migration_audit.setup(); w.setup(); wp.setup()
    from app import cms_pipeline
    cms_pipeline.setup()
    def site(name='Migration'):
        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,'','','automotive','#222222','','migration','{}',t,t))
        return i
    yield site


def item(wp_id,route,kind='pages'):
    return {'id':wp_id,'link':'https://old.example.org'+route,
       'title':{'rendered':f'WP Title {wp_id}'},
       'excerpt':{'rendered':'<p>Intro</p>'},'content':{'rendered':'<p>Legacy HTML</p>'},
       'yoast_head_json':{'title':f'Legacy SEO {wp_id}', 'description':f'Meta {wp_id}',
                           'canonical':'https://old.example.org'+route},
       '_embedded':{'wp:featuredmedia':[{'source_url':'https://old.example.org/image.jpg'}]}}


def mock_pages(monkeypatch, pages):
    def fetch(origin,phase,page):return pages.get((phase,page),[])
    monkeypatch.setattr(wp,'_fetch',fetch)


def test_complete_scan_stage_and_baseline(env,monkeypatch):
    sid=env(); mock_pages(monkeypatch,{('pages',1):[item(42,'/about/')],('posts',1):[item(84,'/news/item/')]})
    sess=wp.create(sid,'https://old.example.org',200)
    assert sess['state']=='created' and sess['next_page']==1
    scan=wp.advance(sid,sess['id']); assert scan['state']=='paused' and scan['phase']=='posts'
    assert not db.rows('SELECT * FROM entries WHERE site_id=?',(sid,))
    scan=wp.advance(sid,sess['id']); assert scan['state']=='ready' and scan['fetched']==2
    done=wp.finalize(sid,sess['id'])
    assert done['baseline_saved']==2 and done['session']['state']=='completed_with_warnings'
    assert {x['route']:x['type'] for x in db.rows('SELECT route,type FROM entries WHERE site_id=?',(sid,))}=={'/about/':'page','/news/item/':'post'}
    assert db.one('SELECT seo_title FROM migration_baseline WHERE site_id=? AND route=?',(sid,'/about/'))['seo_title']=='Legacy SEO 42'
    assert wp.finalize(sid,sess['id'])['state']=='completed_with_warnings'


def test_network_interruption_resumes_exact_page(env,monkeypatch):
    sid=env(); seen=[]
    def flakey(origin,phase,page):
        seen.append((phase,page))
        if len(seen)==1:raise OSError('simulated network outage')
        return [item(1,'/about/')] if phase=='pages' else []
    monkeypatch.setattr(wp,'_fetch',flakey)
    sess=wp.create(sid,'https://old.example.org/',100)
    with pytest.raises(HTTPException) as exc:wp.advance(sid,sess['id'])
    assert exc.value.status_code==502
    assert wp.details(sid,sess['id'])['next_page']==1
    wp.advance(sid,sess['id'])
    assert seen==[('pages',1),('pages',1)]
    assert len(wp.details(sid,sess['id'])['batches'])==1


def test_bad_offsite_url_requires_review(env,monkeypatch):
    sid=env(); wrong=item(1,'/');wrong['link']='https://other.example.org/not-ours/'
    mock_pages(monkeypatch,{('pages',1):[wrong,item(2,'/safe/')],('posts',1):[]})
    s=wp.create(sid,'https://old.example.org/',100)
    wp.advance(sid,s['id']); result=wp.advance(sid,s['id'])
    assert result['reviewed']>=1 and any('不同' in x['issue'] or '固定 URL' in x['issue'] for x in result['warnings'])
    done=wp.finalize(sid,s['id'])
    assert done['requires_review'] and done['session']['state']=='completed_with_warnings'
    from app import cms_pipeline
    assert cms_pipeline.open_reviews(sid)>0
    assert db.one('SELECT COUNT(*) n FROM migration_baseline WHERE site_id=?',(sid,))['n']==1


def test_cancel_prevents_resume(env,monkeypatch):
    sid=env();mock_pages(monkeypatch,{('pages',1):[item(1,'/a/')]})
    s=wp.create(sid,'https://old.example.org/',100)
    assert wp.cancel(sid,s['id'])['state']=='cancelled'
    with pytest.raises(HTTPException) as exc:wp.advance(sid,s['id'])
    assert exc.value.status_code==409
    with pytest.raises(HTTPException):wp.finalize(sid,s['id'])


def test_limit_reached_not_marked_complete(env,monkeypatch):
    sid=env();mock_pages(monkeypatch,{('pages',1):[item(1,'/a/'),item(2,'/b/')]})
    s=wp.create(sid,'https://old.example.org/',1)
    result=wp.advance(sid,s['id'])
    assert result['state']=='limit_reached' and not result['batches']
    with pytest.raises(HTTPException):wp.finalize(sid,s['id'])


def test_cross_site_isolation(env,monkeypatch):
    sid=env('A');other=env('B');s=wp.create(sid,'https://old.example.org/',100)
    with pytest.raises(HTTPException) as exc:wp.details(other,s['id'])
    assert exc.value.status_code==404


def test_commit_preserves_existing_on_skip(env,monkeypatch):
    sid=env(); save_entry(sid,{'type':'page','title':'Custom About','route':'/about/','seo_title':'Our SEO'})
    mock_pages(monkeypatch,{('pages',1):[item(1,'/about/')],('posts',1):[]})
    s=wp.create(sid,'https://old.example.org/',100);wp.advance(sid,s['id']);wp.advance(sid,s['id'])
    wp.finalize(sid,s['id'])
    assert db.one('SELECT title FROM entries WHERE site_id=? AND route=?',(sid,'/about/'))['title']=='Custom About'
    assert db.one('SELECT seo_title FROM migration_baseline WHERE site_id=? AND route=?',(sid,'/about/'))['seo_title']=='Legacy SEO 1'


def test_authenticated_http_workflow(env,monkeypatch):
    sid=env();mock_pages(monkeypatch,{('pages',1):[item(3,'/test/')],('posts',1):[]})
    with TestClient(app) as c:
        assert c.post(f'/api/migrations/sites/{sid}/wordpress/sessions',json={'origin':'https://old.example.org'}).status_code==401
        assert c.post('/api/auth/login',json={'password':'test-v8-admin-strong-password-12345'}).status_code==200
        c.headers['X-SiteForge-CSRF']=c.get('/api/auth/me').json()['csrf']
        created=c.post(f'/api/migrations/sites/{sid}/wordpress/sessions',json={'origin':'https://old.example.org'})
        assert created.status_code==200,created.text
        session=created.json()['id']
        assert c.get('/production-v8').status_code==200
        for _ in range(2):
            r=c.post(f'/api/migrations/sites/{sid}/wordpress/sessions/{session}/advance')
            assert r.status_code==200,r.text
        fin=c.post(f'/api/migrations/sites/{sid}/wordpress/sessions/{session}/finalize')
        assert fin.status_code==200,fin.text
        assert fin.json()['baseline_saved']==1


def test_invalid_origin_or_duplicate_processing(env,monkeypatch):
    sid=env()
    for origin in ['file:///etc/passwd','https://user:pass@example.org','http://example.org?token=a',
                   'https://example.org:9999/path']:
        with pytest.raises(ValueError):wp.create(sid,origin)
    s=wp.create(sid,'https://old.example.org/',100)
    db.execute("UPDATE wp_migration_sessions SET state='running',lease_until='9999-01-01T00:00:00+00:00' WHERE id=?",(s['id'],))
    with pytest.raises(HTTPException) as exc:wp.advance(sid,s['id'])
    assert exc.value.status_code==409


def test_missing_seo_and_remote_media_create_release_review(env,monkeypatch):
    sid=env(); bad=item(10,'/product/example/')
    bad['yoast_head_json']={}
    mock_pages(monkeypatch,{('pages',1):[bad],('posts',1):[]})
    s=wp.create(sid,'https://old.example.org/',100)
    r=wp.advance(sid,s['id'])
    assert r['reviewed']>=2
    wp.advance(sid,s['id'])
    done=wp.finalize(sid,s['id'])
    assert done['session']['state']=='completed_with_warnings'
    from app.cms_pipeline import open_reviews
    assert open_reviews(sid)>=2
    assert wp.finalize(sid,s['id'])['state']=='completed_with_warnings'
    assert open_reviews(sid)>=2