110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
import io
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT))
|
|
TEST_ROOT=Path(tempfile.mkdtemp(prefix="sourcehub38_"))
|
|
os.environ["SOURCEHUB_DATABASE_URL"]=f"sqlite:///{(TEST_ROOT/'sourcehub.db').as_posix()}"
|
|
os.environ["SOURCEHUB_STORAGE_BACKEND"]="local"
|
|
os.environ["SOURCEHUB_STORAGE_DIR"]=str(TEST_ROOT/"storage")
|
|
os.environ["SOURCEHUB_AUTH_MODE"]="mock"
|
|
os.environ["SOURCEHUB_CSRF_ENABLED"]="false"
|
|
os.environ["SOURCEHUB_MALWARE_SCAN"]="disabled"
|
|
os.environ["SOURCEHUB_DEPLOY_MODE"]="disabled"
|
|
os.environ["SOURCEHUB_ALLOWED_HOSTS"]="testserver,localhost,127.0.0.1"
|
|
os.environ["SOURCEHUB_DOWNLOAD_SECRET"]="test-secret-abcdefghijklmnopqrstuvwxyz"
|
|
|
|
from fastapi.testclient import TestClient
|
|
import app as sourcehub
|
|
|
|
|
|
def teardown_module():
|
|
shutil.rmtree(TEST_ROOT,ignore_errors=True)
|
|
|
|
|
|
def create_transfer(c, project="sourcehub", version="3.8-test", receiver="dev3", role="dev2", content=b"NEXPA38"):
|
|
return c.post(
|
|
"/api/transfers",
|
|
data={
|
|
"project_code":project,"version":version,"receiver_role":receiver,
|
|
"file_type":"Source ZIP","summary":"3.8 upload","git_ref":"test","branch":"production-candidate",
|
|
},
|
|
files={"file":("source.zip",io.BytesIO(content),"application/zip")},
|
|
headers={"X-Role":role},
|
|
)
|
|
|
|
|
|
def test_health_system_and_baselines():
|
|
with TestClient(sourcehub.app) as c:
|
|
assert c.get("/health").json()["version"]=="3.8.0"
|
|
s=c.get("/api/system/status").json()
|
|
assert s["database"]["kind"]=="sqlite"
|
|
assert s["storage"]["backend"]=="local"
|
|
assert s["security"]["signed_download"] is True
|
|
assert s["security"]["malware_scan"]=="disabled"
|
|
assert len(c.get("/api/projects",headers={"X-Role":"dev1"}).json())==7
|
|
assert len(c.get("/api/baselines",headers={"X-Role":"dev1"}).json())>=7
|
|
|
|
|
|
def test_upload_hash_signed_download_and_revision():
|
|
with TestClient(sourcehub.app) as c:
|
|
r=create_transfer(c,content=b"NEXPA38-FILE")
|
|
assert r.status_code==200,r.text
|
|
t=r.json()
|
|
rev=t["revisions"][0]
|
|
assert len(rev["sha256"])==64
|
|
link=c.get(rev["download_link_api"],headers={"X-Role":"dev2"})
|
|
assert link.status_code==200
|
|
d=c.get(link.json()["url"])
|
|
assert d.status_code==200
|
|
assert d.content==b"NEXPA38-FILE"
|
|
|
|
r2=c.post(
|
|
f"/api/transfers/{t['id']}/revisions",
|
|
data={"summary":"R2"},
|
|
files={"file":("source-r2.zip",io.BytesIO(b"NEXPA38-R2"),"application/zip")},
|
|
headers={"X-Role":"dev2"},
|
|
)
|
|
assert r2.status_code==200
|
|
assert r2.json()["revisions"][0]["revision_no"]==2
|
|
|
|
|
|
def test_signed_download_rejects_bad_token():
|
|
with TestClient(sourcehub.app) as c:
|
|
t=create_transfer(c).json()
|
|
rid=t["revisions"][0]["id"]
|
|
bad=c.get(f"/download/{rid}?token=not-valid")
|
|
assert bad.status_code==403
|
|
|
|
|
|
def test_approval_permission_and_flow():
|
|
with TestClient(sourcehub.app) as c:
|
|
t=create_transfer(c,project="core",version="3.4.2-test",receiver="dev3",role="dev1").json()
|
|
denied=c.post(f"/api/transfers/{t['id']}/approve",json={"note":""},headers={"X-Role":"dev1"})
|
|
assert denied.status_code==403
|
|
ok=c.post(f"/api/transfers/{t['id']}/approve",json={"note":"received"},headers={"X-Role":"dev3"})
|
|
assert ok.status_code==200
|
|
assert ok.json()["state"]=="QA_REVIEW"
|
|
|
|
|
|
def test_git_binding_and_deployment_gate():
|
|
with TestClient(sourcehub.app) as c:
|
|
denied=c.post("/api/projects/core/git/bind",json={"owner":"nexpa","repo":"core","branch":"main"},headers={"X-Role":"planner"})
|
|
assert denied.status_code==403
|
|
ok=c.post("/api/projects/core/git/bind",json={"owner":"nexpa","repo":"core","branch":"main"},headers={"X-Role":"dev1"})
|
|
assert ok.status_code==200
|
|
assert ok.json()["status"]["configured"] is False
|
|
|
|
ds=c.get("/api/deployments",headers={"X-Role":"deploy"}).json()
|
|
core=next(x for x in ds if x["project_code"]=="core")
|
|
not_deploy=c.post(f"/api/deployments/{core['id']}/execute",json={},headers={"X-Role":"dev1"})
|
|
assert not_deploy.status_code==403
|
|
disabled=c.post(f"/api/deployments/{core['id']}/execute",json={},headers={"X-Role":"deploy"})
|
|
assert disabled.status_code==503
|
|
audit=c.get("/api/audit",headers={"X-Role":"dev1"}).json()
|
|
assert any(x["type"]=="배포" for x in audit)
|