Files

125 lines
7.0 KiB
Python
Raw Permalink Normal View History

import io,json,os,tempfile,time,zipfile
from pathlib import Path
import pytest
from app import cms_pipeline,pg_compat,deploy_agent,redis_signal
def sample_zip(files=None):
out=io.BytesIO()
with zipfile.ZipFile(out,'w') as archive:
for path,data in (files or {'index.html':'<title>Site</title>','products/index.html':'Item'}).items():
archive.writestr(path,data)
return out.getvalue()
def test_pg_placeholder_lexer():
sql="SELECT '?', \"?\", data FROM things WHERE id=? AND key='abc''?def' AND name=?"
assert pg_compat.placeholders(sql).count('%s')==2
assert "'abc''?def'" in pg_compat.placeholders(sql)
assert pg_compat.normalize('INSERT OR IGNORE INTO t (a) VALUES (?)')=='INSERT INTO t (a) VALUES (%s) ON CONFLICT DO NOTHING'
def test_release_sign_verify(monkeypatch):
secret='superlong-secret-32-characters-minimum'
raw=b'example'
stamp=str(int(time.time()))
from starlette.requests import Request
monkeypatch.setenv('SITEFORGE_DEPLOY_SECRET',secret)
sig=deploy_agent.signature(secret,'example',stamp,raw,'POST','/deploy/example')
req=Request({'type':'http','method':'POST','path':'/deploy/example','headers':[(b'x-siteforge-timestamp',stamp.encode()),(b'x-siteforge-signature',sig.encode())]})
deploy_agent.authorize('example',req,raw)
from fastapi import HTTPException
with pytest.raises(HTTPException):deploy_agent.authorize('example',req,b'tampered')
def test_release_archive_path_rejection():
assert deploy_agent.validate_archive(sample_zip())
with pytest.raises(ValueError):deploy_agent.validate_archive(sample_zip({'index.html':'ok','../secret.txt':'x'}))
with pytest.raises(ValueError):deploy_agent.validate_archive(sample_zip({'missing.txt':'oops'}))
def test_release_install_rollback(monkeypatch,tmp_path):
monkeypatch.setattr(deploy_agent,'ROOT',tmp_path)
data=sample_zip()
one=deploy_agent.install('cars',data)
two=deploy_agent.install('cars',data)
assert one['release']==two['release']
parent=tmp_path/'cars'
assert (parent/'current').read_text()==one['release']
assert (parent/'releases'/one['release']/'index.html').is_file()
assert not any(x.name.startswith('.unpacked-') for x in parent.iterdir())
def test_cms_wordpress_acf_elementor_warning():
rows={'items':[{'id':15,'type':'page','slug':'about','title':{'rendered':'About Us'},'link':'https://example.org/about/',
'content':{'rendered':'<p>Hello</p>'},'acf':{'dealer':'HESU','gallery':[1,2]},'meta':{'_elementor_data':'[]'},'yoast_head_json':{'title':'SEO About'}}]}
report=cms_pipeline.parse_export('wordpress-rest',json.dumps(rows).encode())
assert report['mapped']==1 and report['entries'][0]['route']=='/about/'
assert report['entries'][0]['seo_title']=='SEO About'
assert report['warnings']
def test_cms_woocommerce():
rows={'products':[{'id':5,'name':'New Pickup','permalink':'https://example.org/product/pickup/',
'sku':'ABC','price':'100','images':[{'src':'https://example.org/a.png'}],
'variants':[], 'variations':[1,2]}]}
report=cms_pipeline.parse_export('woocommerce',json.dumps(rows).encode())
assert report['entries'][0]['type']=='product'
assert report['entries'][0]['meta']['fields']['gallery']==['https://example.org/a.png']
assert report['warnings']
def test_cms_drupal_and_shopify():
drupal={'data':[{'type':'node--article','id':'abc','attributes':{'title':'News','path':{'alias':'/news/hello/'},
'body':{'processed':'<p>News</p>'}},'relationships':{'field_brand':{'data':{'id':'12'}}}}]}
report=cms_pipeline.parse_export('drupal-jsonapi',json.dumps(drupal).encode())
assert report['mapped']==1 and report['entries'][0]['route']=='/news/hello/'
assert report['warnings']
shopify={'products':[{'id':1,'title':'Shop Item','handle':'shop-item','vendor':'HESU',
'variants':[{'price':'100','sku':'Z9'}]}]}
report=cms_pipeline.parse_export('shopify',json.dumps(shopify).encode())
assert report['entries'][0]['meta']['fields']['brand']=='HESU'
def test_cms_invalid_routes_duplicates():
rec={'products':[{'title':'Foo','handle':'foo'},{'title':'Foo Again','handle':'foo'}]}
result=cms_pipeline.parse_export('shopify',json.dumps(rec).encode())
assert len(result['errors'])==1
def test_redis_down_is_nonfatal(monkeypatch):
monkeypatch.setattr(redis_signal,'URL','redis://127.0.0.1:1/0')
assert redis_signal.notify() is False
def test_release_agent_http_roundtrip(monkeypatch,tmp_path):
from fastapi.testclient import TestClient
monkeypatch.setattr(deploy_agent,'ROOT',tmp_path)
secret='test-release-secret-32-chars-safe-test'
monkeypatch.setenv('SITEFORGE_DEPLOY_SECRET',secret)
stamp=str(int(time.time()))
blob=sample_zip()
c=TestClient(deploy_agent.app)
headers={'X-SiteForge-Timestamp':stamp,'X-SiteForge-Signature':deploy_agent.signature(secret,'newsite',stamp,blob,'POST','/deploy/newsite')}
upload=c.post('/deploy/newsite',headers=headers,content=blob)
assert upload.status_code==200,upload.text
release=upload.json()['release']
assert c.get('/public/newsite/index.html').status_code==200
rollback_headers={'X-SiteForge-Timestamp':stamp,'X-SiteForge-Signature':deploy_agent.signature(secret,'newsite',stamp,b'','POST',f'/sites/newsite/rollback/{release}')}
assert c.post(f'/sites/newsite/rollback/{release}',headers=rollback_headers).status_code==200
assert c.get('/public/newsite/products/index.html').status_code==200
assert c.post('/deploy/newsite',headers={'X-SiteForge-Timestamp':stamp,'X-SiteForge-Signature':'bad'},content=blob).status_code==401
def test_release_endpoint_requires_approval(monkeypatch,tmp_path):
from fastapi.testclient import TestClient
from app import db,builder
from app.main import app
monkeypatch.setattr(db,'DATA',tmp_path)
monkeypatch.setattr(db,'DB_FILE',tmp_path/'db.sqlite3')
monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','test-admin-password-for-release')
monkeypatch.setenv('SITEFORGE_SESSION_SECRET','test-session-secret-for-release-app-2026')
with TestClient(app) as c:
assert c.get('/operations',follow_redirects=False).status_code in (302,303,307)
assert c.post('/api/auth/login',json={'password':'test-admin-password-for-release'}).status_code==200
c.headers['X-SiteForge-CSRF']=c.get('/api/auth/me').json()['csrf']
assert c.get('/operations').status_code==200
r=c.post('/api/sites',json={'name':'Release Review Site','theme':'corporate'})
assert r.status_code==201,r.text
sid=r.json()['id'];result=builder.build(sid)
res=c.post(f'/api/production/sites/{sid}/deploy',json={'build_id':result['version'],'target':'staging'})
assert res.status_code==409,res.text
# Migration upload requires same-site authentication and editor role
sample=c.post(f'/api/migrations/sites/{sid}/preview',files={'file':('sample.json',b'{"items":[]}','application/json')},data={'platform':'wordpress-rest'})
assert sample.status_code==200,sample.text