"""Local, deterministic SFTP + S3 publisher contract tests; no external servers.""" from __future__ import annotations from contextlib import contextmanager import hashlib, io, json, os, stat, zipfile from pathlib import Path from types import SimpleNamespace import pytest from fastapi import HTTPException from app import release_targets, release_api, db, enterprise def package(html='first version'): obj=io.BytesIO() with zipfile.ZipFile(obj,'w') as z: z.writestr('index.html',html) z.writestr('products/index.html','Products') return obj.getvalue() class FakeSFTP: def lstat(self,p):return os.lstat(p) def mkdir(self,p):Path(p).mkdir() def file(self,p,mode):return open(p,mode) def symlink(self,to,p):os.symlink(to,p) def readlink(self,p):return os.readlink(p) def remove(self,p):os.remove(p) def rename(self,a,b):os.rename(a,b) def posix_rename(self,a,b):os.replace(a,b) def test_sftp_immutable_release_atomic_switch_rollback_and_verification(monkeypatch,tmp_path): root=str(tmp_path/'safe-root') cfg={'type':'sftp','root':root,'host':'myhost','user':'publisher','port':22,'known_hosts':'mounted','ssh_key':'mounted'} @contextmanager def conn(_cfg):yield FakeSFTP() monkeypatch.setattr(release_targets,'_sftp_connection',conn) def probe(target,site,body): assert (Path(root)/site/'current'/'index.html').read_bytes()==body monkeypatch.setattr(release_targets,'_probe',probe) a=release_targets.deploy_sftp('staging','mysite',package('first'),cfg) b=release_targets.deploy_sftp('staging','mysite',package('second'),cfg) assert b['previous']==a['release'] assert (Path(root)/'mysite'/'current').is_symlink() assert (Path(root)/'mysite'/'current'/'index.html').read_text()=='second' # Immutable retry does not change release ID or create additional release. assert release_targets.deploy_sftp('staging','mysite',package('second'),cfg)['release']==b['release'] rolled=release_targets.rollback_sftp('staging','mysite',a['release'],cfg) assert rolled['rolled_back'] assert (Path(root)/'mysite'/'current'/'index.html').read_text()=='first' # Tampering with a staged/previous version causes verification to fail. (Path(root)/'mysite'/'releases'/b['release']/'index.html').write_text('tampered') with pytest.raises(ValueError,match='哈希'):release_targets.rollback_sftp('staging','mysite',b['release'],cfg) assert (Path(root)/'mysite'/'current'/'index.html').read_text()=='first' def test_sftp_probe_failure_restores_previous_release(monkeypatch,tmp_path): cfg={'type':'sftp','root':str(tmp_path/'release'),'host':'host','user':'user','port':22,'known_hosts':'kh','ssh_key':'key'} @contextmanager def fake(_):yield FakeSFTP() monkeypatch.setattr(release_targets,'_sftp_connection',fake) monkeypatch.setattr(release_targets,'_probe',lambda *x:True) old=release_targets.deploy_sftp('staging','abc',package('before'),cfg) def fail(*_):raise HTTPException(502,'probe fails') monkeypatch.setattr(release_targets,'_probe',fail) with pytest.raises(HTTPException):release_targets.deploy_sftp('staging','abc',package('after'),cfg) assert (Path(cfg['root'])/'abc'/'current'/'index.html').read_text()=='before' class MissingS3(Exception): def __init__(self):self.response={'Error':{'Code':'NoSuchKey'}} class FakeS3: def __init__(self):self.data={};self.etags={} def get_object(self,*,Bucket,Key): k=(Bucket,Key) if k not in self.data:raise MissingS3() return {'Body':io.BytesIO(self.data[k]),'ETag':self.etags[k]} def put_object(self,*,Bucket,Key,Body,IfMatch=None,IfNoneMatch=None,**kwargs): k=(Bucket,Key) if IfMatch and self.etags.get(k)!=IfMatch:raise ValueError('PreconditionFailed') if IfNoneMatch=='*' and k in self.data:raise ValueError('PreconditionFailed') self.data[k]=bytes(Body) self.etags[k]='"'+hashlib.md5(Body).hexdigest()+'"' return {'ETag':self.etags[k]} def test_s3_pointer_preflight_and_conditional_activation(monkeypatch): cfg={'type':'s3','bucket':'example-bucket','region':'us-east-1','prefix':'web','endpoint':''} s3=FakeS3() monkeypatch.setattr(release_targets,'_s3_client',lambda _:s3) def probe(target,site,body): cur=json.loads(s3.data[('example-bucket','web/sites/mysite/current.json')])['release'] assert s3.data[('example-bucket',f'web/sites/mysite/releases/{cur}/index.html')]==body monkeypatch.setattr(release_targets,'_probe',probe) a=release_targets.deploy_s3('staging','mysite',package('before'),cfg) b=release_targets.deploy_s3('staging','mysite',package('after'),cfg) assert b['previous']==a['release'] assert release_targets.deploy_s3('staging','mysite',package('after'),cfg)['release']==b['release'] assert release_targets.rollback_s3('staging','mysite',a['release'],cfg)['rolled_back'] curr,_=release_targets._s3_current(s3,cfg,'mysite') assert curr==a['release'] # Fail closed if no pointer-aware edge is configured. monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_TYPE','s3') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_BUCKET','example-bucket') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_REGION','us-east-1') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_PUBLIC_URL','https://preview.example.com/mysite') with pytest.raises(HTTPException,match='POINTER_AWARE'):release_targets.config('staging') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_PUBLIC_URL','https://preview.example.com/mysite') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_POINTER_AWARE','1') assert release_targets.config('staging')['type']=='s3' def test_s3_verification_failure_does_not_activate(monkeypatch): cfg={'type':'s3','bucket':'test-bucket','region':'us-east-1','prefix':'x','endpoint':''} obj=FakeS3();monkeypatch.setattr(release_targets,'_s3_client',lambda _:obj) monkeypatch.setattr(release_targets,'_probe',lambda *a:True) first=release_targets.deploy_s3('staging','one',package('ok'),cfg) raw=package('bad');release,digest,files,hashes=release_targets._archive(raw) manifest=release_targets._manifest(digest,hashes) key=f'x/sites/one/releases/{release}/' for path,body in files.items():obj.put_object(Bucket='test-bucket',Key=key+path,Body=body) obj.put_object(Bucket='test-bucket',Key=key+'.siteforge-manifest.json',Body=manifest) obj.put_object(Bucket='test-bucket',Key=key+'index.html',Body=b'corrupt') with pytest.raises(ValueError,match='校验'):release_targets.deploy_s3('staging','one',raw,cfg) current,_=release_targets._s3_current(obj,cfg,'one') assert current==first['release'] def test_config_fail_closed_and_no_url_secret(monkeypatch,tmp_path): monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_TYPE','sftp') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_PUBLIC_URL','https://preview.example.com/mysite') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_HOST','example.org') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_USERNAME','siteforge') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_REMOTE_ROOT','/srv/sites') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_KNOWN_HOSTS_FILE','/run/secrets/known_hosts') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_SSH_KEY_FILE','/run/secrets/publish_key') assert release_targets.config('staging')['type']=='sftp' with pytest.raises(HTTPException):release_targets.config('staging',check_files=True) monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_REMOTE_ROOT','/../etc') with pytest.raises(HTTPException):release_targets.config('staging') monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_TYPE','unsupported') with pytest.raises(HTTPException):release_targets.backend('staging') def test_s3_gateway_reads_atomic_pointer(monkeypatch): from app import s3_gateway from fastapi.testclient import TestClient cfg={'type':'s3','bucket':'test-bucket','region':'ap-southeast-1','prefix':'sitesdemo','endpoint':''} store=FakeS3() monkeypatch.setattr(release_targets,'_s3_client',lambda _:store) monkeypatch.setattr(release_targets,'_probe',lambda *a:True) one=release_targets.deploy_s3('staging','sample',package('hello'),cfg) monkeypatch.setenv('SITEFORGE_S3_GATEWAY_BUCKET','test-bucket') monkeypatch.setenv('SITEFORGE_S3_GATEWAY_REGION','ap-southeast-1') monkeypatch.setenv('SITEFORGE_S3_GATEWAY_PREFIX','sitesdemo') monkeypatch.setattr(s3_gateway,'client',lambda _:store) client=TestClient(s3_gateway.app) assert client.get('/sample/').status_code==200 assert client.get('/sample/').text=='hello' assert client.get('/sample/products/').text=='Products' assert client.get('/sample/').headers['cache-control'].startswith('no-cache') two=release_targets.deploy_s3('staging','sample',package('world'),cfg) assert one['release']!=two['release'] assert client.get('/sample/').text=='world' assert client.get('/sample/%2e%2e/private').status_code in (404,) assert client.get('/sample/no-such-file').status_code==404 def test_native_provider_keeps_existing_approval_queue_and_history(monkeypatch,tmp_path): from app import release_api from app import db,enterprise monkeypatch.setattr(db,'DATA',tmp_path) monkeypatch.setattr(db,'DB_FILE',tmp_path/'data.sqlite3') db.init();enterprise.setup();release_api.setup() db.insert('INSERT INTO sites(id,name,created_at,updated_at) VALUES(?,?,?,?)', ('proj','Project',db.now(),db.now())) archive_dir=tmp_path/'archives';archive_dir.mkdir() path=archive_dir/'proj-build1.zip';path.write_bytes(package('new')) monkeypatch.setattr(release_api,'approved',lambda site,build:{'zip_path':str(path)}) monkeypatch.setenv('SITEFORGE_DEPLOY_STAGING_TYPE','sftp') monkeypatch.setattr(release_targets,'deploy',lambda target,site,archive:{ 'published':True,'release':'a'*20,'sha256':'b'*64,'files':2,'target':target,'backend':'sftp'}) monkeypatch.setattr(release_targets,'rollback',lambda target,site,release:{ 'rolled_back':True,'release':release,'target':target,'backend':'sftp'}) assert release_api.run_delivery('proj','remote_deploy',{'target':'staging','build_id':'build1'},'task1')['published'] assert db.one('SELECT release_id FROM remote_releases WHERE site_id=?',('proj',))['release_id']=='a'*20 assert release_api.run_delivery('proj','remote_rollback',{'target':'staging','release_id':'a'*20},'task2')['rolled_back'] assert len(db.rows('SELECT * FROM release_history WHERE site_id=?',('proj',)))==2