238 lines
8.5 KiB
Python
238 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from k1link.simulation.gaussian_pipeline_gateway import (
|
|
BUILD_REQUEST_SCHEMA,
|
|
GaussianPipelineGateway,
|
|
GaussianPipelineGatewayError,
|
|
GaussianPipelineIntegrityError,
|
|
)
|
|
|
|
|
|
def _token_file(tmp_path: Path) -> Path:
|
|
token = tmp_path / "gaussian.token"
|
|
token.write_text("t" * 64, encoding="utf-8")
|
|
return token
|
|
|
|
|
|
def test_gateway_uploads_with_tus_and_reads_provider_contract(tmp_path: Path) -> None:
|
|
uploaded = bytearray()
|
|
metadata: dict[str, str] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["authorization"] == f"Bearer {'t' * 64}"
|
|
if request.method == "GET" and request.url.path == "/v1/capabilities":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"schema_version": "gaussian-pipeline.capabilities/v1",
|
|
"service": "ndc-gaussian-pipeline",
|
|
"api_version": "gaussian-pipeline.api/v1",
|
|
"provider": {"id": "playcanvas-splat-transform", "version": "3.3.3"},
|
|
"runtime": {
|
|
"source_revision": "a" * 40,
|
|
"image_digest": f"sha256:{'b' * 64}",
|
|
},
|
|
},
|
|
)
|
|
if request.method == "POST" and request.url.path == "/v1/uploads":
|
|
for item in request.headers["upload-metadata"].split(","):
|
|
key, encoded = item.split(" ", 1)
|
|
metadata[key] = base64.b64decode(encoded).decode("utf-8")
|
|
return httpx.Response(201, headers={"Location": "/v1/uploads/upload-001"})
|
|
if request.method == "HEAD" and request.url.path == "/v1/uploads/upload-001":
|
|
return httpx.Response(200, headers={"Upload-Offset": str(len(uploaded))})
|
|
if request.method == "PATCH" and request.url.path == "/v1/uploads/upload-001":
|
|
assert int(request.headers["upload-offset"]) == len(uploaded)
|
|
uploaded.extend(request.content)
|
|
return httpx.Response(204, headers={"Upload-Offset": str(len(uploaded))})
|
|
return httpx.Response(404)
|
|
|
|
source = tmp_path / "yard.lcc2"
|
|
source.write_bytes(b"portable-gaussian-source")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
with GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
_token_file(tmp_path),
|
|
chunk_bytes=5,
|
|
transport=httpx.MockTransport(handler),
|
|
) as gateway:
|
|
assert gateway.capabilities()["service"] == "ndc-gaussian-pipeline"
|
|
descriptor = gateway.upload_source(
|
|
source,
|
|
filename="yard.lcc2",
|
|
source_format="lcc2",
|
|
sha256=digest,
|
|
)
|
|
|
|
assert bytes(uploaded) == source.read_bytes()
|
|
assert metadata == {"filename": "yard.lcc2", "format": "lcc2", "sha256": digest}
|
|
assert descriptor.to_dict() == {
|
|
"upload_id": "upload-001",
|
|
"filename": "yard.lcc2",
|
|
"format": "lcc2",
|
|
"sha256": digest,
|
|
"byte_length": len(uploaded),
|
|
}
|
|
|
|
|
|
def test_gateway_rejects_local_source_digest_mismatch(tmp_path: Path) -> None:
|
|
source = tmp_path / "yard.lcc"
|
|
source.write_bytes(b"source")
|
|
with (
|
|
GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
_token_file(tmp_path),
|
|
transport=httpx.MockTransport(lambda _request: httpx.Response(500)),
|
|
) as gateway,
|
|
pytest.raises(GaussianPipelineIntegrityError, match="digest does not match"),
|
|
):
|
|
gateway.upload_source(
|
|
source,
|
|
filename="yard.lcc",
|
|
source_format="lcc",
|
|
sha256="a" * 64,
|
|
)
|
|
|
|
|
|
def test_gateway_rejects_upload_location_outside_provider(tmp_path: Path) -> None:
|
|
source = tmp_path / "yard.lcc"
|
|
source.write_bytes(b"source")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
def handler(_request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
201,
|
|
headers={"Location": "https://attacker.invalid/v1/uploads/stolen"},
|
|
)
|
|
|
|
with (
|
|
GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
_token_file(tmp_path),
|
|
transport=httpx.MockTransport(handler),
|
|
) as gateway,
|
|
pytest.raises(GaussianPipelineGatewayError, match="escaped the provider"),
|
|
):
|
|
gateway.upload_source(
|
|
source,
|
|
filename="yard.lcc",
|
|
source_format="lcc",
|
|
sha256=digest,
|
|
)
|
|
|
|
|
|
def test_gateway_rejects_symlink_source_and_token(tmp_path: Path) -> None:
|
|
source = tmp_path / "source.lcc"
|
|
source.write_bytes(b"source")
|
|
source_link = tmp_path / "source-link.lcc"
|
|
source_link.symlink_to(source)
|
|
token = _token_file(tmp_path)
|
|
token_link = tmp_path / "token-link"
|
|
token_link.symlink_to(token)
|
|
|
|
with pytest.raises(GaussianPipelineGatewayError, match="token must be one regular file"):
|
|
GaussianPipelineGateway("http://gaussian.test", token_link)
|
|
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
with (
|
|
GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
token,
|
|
transport=httpx.MockTransport(lambda _request: httpx.Response(500)),
|
|
) as gateway,
|
|
pytest.raises(GaussianPipelineIntegrityError, match="one regular file"),
|
|
):
|
|
gateway.upload_source(
|
|
source_link,
|
|
filename="source.lcc",
|
|
source_format="lcc",
|
|
sha256=digest,
|
|
)
|
|
|
|
|
|
def test_gateway_submits_tracks_and_imports_digest_bound_artifact(tmp_path: Path) -> None:
|
|
artifact = b"preview-sog"
|
|
artifact_sha = hashlib.sha256(artifact).hexdigest()
|
|
job_id = "gsp-20260825200000-deadbeef"
|
|
request_document: dict[str, object] = {
|
|
"schema_version": BUILD_REQUEST_SCHEMA,
|
|
"idempotency_key": "missioncore-build-01",
|
|
}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.method == "POST" and request.url.path == "/v1/jobs":
|
|
assert json.loads(request.content) == request_document
|
|
return httpx.Response(
|
|
202,
|
|
json={"schema_version": "gaussian-pipeline.job/v1", "job_id": job_id},
|
|
)
|
|
if request.method == "GET" and request.url.path == f"/v1/jobs/{job_id}":
|
|
return httpx.Response(
|
|
200,
|
|
json={"schema_version": "gaussian-pipeline.job/v1", "job_id": job_id},
|
|
)
|
|
if request.method == "GET" and request.url.path == f"/v1/jobs/{job_id}/result":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"schema_version": "gaussian-pipeline.build-result/v1",
|
|
"job_id": job_id,
|
|
"runtime": {
|
|
"source_revision": "a" * 40,
|
|
"image_digest": f"sha256:{'b' * 64}",
|
|
},
|
|
},
|
|
)
|
|
if request.method == "GET" and request.url.path.endswith("/artifacts/preview.sog"):
|
|
return httpx.Response(200, content=artifact)
|
|
return httpx.Response(404)
|
|
|
|
with GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
_token_file(tmp_path),
|
|
transport=httpx.MockTransport(handler),
|
|
) as gateway:
|
|
assert gateway.submit_build(request_document)["job_id"] == job_id
|
|
assert gateway.get_job(job_id)["job_id"] == job_id
|
|
assert gateway.get_result(job_id)["job_id"] == job_id
|
|
destination = gateway.download_artifact(
|
|
job_id,
|
|
{
|
|
"logical_path": "preview.sog",
|
|
"sha256": artifact_sha,
|
|
"byte_length": len(artifact),
|
|
},
|
|
tmp_path / "import" / "preview.sog",
|
|
)
|
|
assert destination.read_bytes() == artifact
|
|
|
|
|
|
def test_gateway_rejects_capabilities_without_runtime_provenance(tmp_path: Path) -> None:
|
|
def handler(_request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"schema_version": "gaussian-pipeline.capabilities/v1",
|
|
"service": "ndc-gaussian-pipeline",
|
|
"api_version": "gaussian-pipeline.api/v1",
|
|
},
|
|
)
|
|
|
|
with (
|
|
GaussianPipelineGateway(
|
|
"http://gaussian.test",
|
|
_token_file(tmp_path),
|
|
transport=httpx.MockTransport(handler),
|
|
) as gateway,
|
|
pytest.raises(GaussianPipelineGatewayError, match="provenance is unavailable"),
|
|
):
|
|
gateway.capabilities()
|