"""游戏内容对象存储契约的离线测试。""" import hashlib import json import subprocess import sys from io import BytesIO from pathlib import Path import pytest from jsonschema import Draft202012Validator from worker import game_artifact_store as store from worker import game_source_archive_store as source_store TENANT_ID = "1" GAME_ID = "1024" VERSION_ID = "2048" def _make_game(root: Path, *, package_game_id: str = GAME_ID, package_version_id: str = VERSION_ID) -> None: """创建含完整 GamePackage、源工程、素材、runtime 和证据的最小游戏包。""" (root / "src").mkdir(parents=True) (root / "assets").mkdir(parents=True) (root / "dist").mkdir(parents=True) (root / "evidence").mkdir(parents=True) (root / "src" / "main.js").write_text("export const GAME = 1;\n", encoding="utf-8") (root / "assets" / "hero.webp").write_bytes(b"asset") (root / "dist" / "bundle.iife.js").write_bytes(b"bundle") (root / "dist" / "scan.json").write_text("{}", encoding="utf-8") (root / "evidence" / "qa.md").write_text("verified", encoding="utf-8") game_package = { "schemaVersion": "1.0", "gameId": package_game_id, "versionId": package_version_id, "templateId": "runner", "gameConfig": {}, "assets": [], "manifest": { "runtimeVersion": "1.0.0", "entry": "index.html", "preloadPolicy": "eager", "bundleSize": len(b"bundle"), "checksum": "0" * 64, }, "meta": { "title": "契约测试游戏", "summary": "只用于平台契约测试", "cover": "https://assets.invalid/cover.webp", "ageRating": "all", }, } (root / "game-package.json").write_text( json.dumps(game_package, ensure_ascii=False, separators=(",", ":")), encoding="utf-8", ) (root / "src" / "__pycache__").mkdir() (root / "src" / "__pycache__" / "ignored.pyc").write_bytes(b"cache") def _source_revision(root: Path, *, source_hash: str | None = None) -> dict: """构造指向现有 Tier2 SourceProjectStore 的不可变源修订身份。""" if source_hash is None: digest = hashlib.sha256() digest.update(b"src/main.js\0") digest.update("export const GAME = 1;\n\0".encode()) source_hash = digest.hexdigest() return { "provider": "tier2_source_project", "gameId": "night-market", "revisionId": "m4-r1", "manifestRef": "tier2-source-project:night-market:m4-r1", "sourceHash": source_hash, } def _build(root: Path) -> dict: return store.build_manifest( root, TENANT_ID, GAME_ID, VERSION_ID, package_type="phaser", runtime_manifest_path="game-package.json", source_revision=_source_revision(root), ) def test_build_and_verify_manifest(tmp_path: Path): _make_game(tmp_path) manifest = _build(tmp_path) store.verify_manifest(manifest) assert manifest["totals"]["objects"] == 4 assert manifest["totals"]["assetBytes"] == len(b"asset") assert manifest["bundleHash"] == store._hash_bytes(b"bundle") assert manifest["runtimeManifest"]["key"].endswith("/manifest.json") assert manifest["sourceRevision"]["revisionId"] == "m4-r1" assert not any(item["category"] == "source" for item in manifest["objects"]) assert not any("ignored.pyc" in item["path"] for item in manifest["objects"]) schema_path = Path(__file__).resolve().parents[3] / "contracts/game-artifact-manifest.schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) errors = sorted(Draft202012Validator(schema).iter_errors(manifest), key=lambda item: list(item.path)) assert not errors, [error.message for error in errors] assert _build(tmp_path)["manifestHash"] == manifest["manifestHash"] def test_manifest_hash_and_schema_detect_mutation(tmp_path: Path): _make_game(tmp_path) manifest = _build(tmp_path) manifest["objects"][0]["bytes"] += 1 with pytest.raises(store.ArtifactError, match="manifestHash"): store.verify_manifest(manifest) manifest = _build(tmp_path) manifest["status"] = "published" unsigned = dict(manifest) unsigned.pop("manifestHash") manifest["manifestHash"] = store._hash_bytes(store._MANIFEST_HASH_DOMAIN + store._canonical_json(unsigned)) with pytest.raises(store.ArtifactError, match="game-artifact-manifest.schema.json"): store.verify_manifest(manifest) def test_rejects_symlink_and_source_hash_drift(tmp_path: Path): _make_game(tmp_path) outside = tmp_path.parent / "outside-game-secret.txt" outside.write_text("secret", encoding="utf-8") (tmp_path / "src" / "linked.txt").symlink_to(outside) with pytest.raises(store.ArtifactError, match="符号链接"): _build(tmp_path) (tmp_path / "src" / "linked.txt").unlink() with pytest.raises(store.ArtifactError, match="源工程 hash"): store.build_manifest( tmp_path, TENANT_ID, GAME_ID, VERSION_ID, package_type="phaser", runtime_manifest_path="game-package.json", source_revision=_source_revision(tmp_path, source_hash="0" * 64), ) def test_game_package_contract_and_identity_are_enforced(tmp_path: Path): _make_game(tmp_path, package_version_id="9999") with pytest.raises(store.ArtifactError, match="对象版本不匹配"): _build(tmp_path) with pytest.raises(store.ArtifactError, match="数据库正整数"): store.build_manifest( tmp_path, TENANT_ID, "night-market", "m4-r1", package_type="phaser", runtime_manifest_path="game-package.json", source_revision=_source_revision(tmp_path), ) class _Response: """模拟 MinIO HTTPResponse，覆盖连接释放协议。""" def __init__(self, data: bytes): self._data = data self._offset = 0 self.read_sizes: list[int] = [] def read(self, size: int = -1) -> bytes: self.read_sizes.append(size) if size < 0: size = len(self._data) - self._offset chunk = self._data[self._offset:self._offset + size] self._offset += len(chunk) return chunk def close(self) -> None: return None def release_conn(self) -> None: return None class _NoSuchKey(Exception): """模拟 MinIO S3Error 的稳定 code。""" code = "NoSuchKey" class _NoSuchBucket(Exception): """模拟桶缺失，必须与对象缺失采用不同失败语义。""" code = "NoSuchBucket" class _FakeMinio: """内存对象存储，用于验证上传、幂等、可信下载和原子物化。""" def __init__(self): self.objects: dict[tuple[str, str], bytes] = {} self.bucket_exists_calls = 0 self.make_bucket_calls = 0 self.put_calls: list[tuple[str, str]] = [] self.last_response: _Response | None = None def bucket_exists(self, _bucket: str) -> bool: self.bucket_exists_calls += 1 raise AssertionError("上传器禁止探测/列出桶") def make_bucket(self, _bucket: str) -> None: self.make_bucket_calls += 1 def put_object(self, bucket: str, key: str, stream: BytesIO, *, length: int, content_type: str): data = stream.read() assert len(data) == length assert content_type self.put_calls.append((bucket, key)) self.objects[(bucket, key)] = data def get_object(self, bucket: str, key: str) -> _Response: identity = (bucket, key) if identity not in self.objects: raise _NoSuchKey(key) self.last_response = _Response(self.objects[identity]) return self.last_response def _trusted_fetch(store_config: dict, key: str, manifest: dict) -> dict: return store.fetch_manifest( store_config, key=key, expected_manifest_hash=manifest["manifestHash"], expected_tenant_id=TENANT_ID, expected_game_id=GAME_ID, expected_version_id=VERSION_ID, ) def test_upload_download_and_idempotency(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): game = tmp_path / "game" _make_game(game) manifest = _build(game) client = _FakeMinio() monkeypatch.setattr(store, "_minio_client", lambda _config: client) config = {"artifact_bucket": "game-artifacts", "evidence_bucket": "game-evidence"} first = store.upload_manifest(game, manifest, config, external_single_writer=True) assert first["idempotent"] is False assert first["status"] == "verified_pending" assert first["committed"] is False second = store.upload_manifest(game, manifest, config, external_single_writer=True) assert second["idempotent"] is True assert second["status"] == "verified_pending" assert second["committed"] is False prefix = f"tenants/{TENANT_ID}/games/{GAME_ID}/versions/{VERSION_ID}" assert ("game-artifacts", f"{prefix}/assets/hero.webp") in client.objects assert ("game-evidence", f"{prefix}/evidence/qa.md") in client.objects assert not any("source/src/main.js" in key for _, key in client.objects) remote = _trusted_fetch(config, first["artifactManifestKey"], manifest) target = tmp_path / "materialized" store.download_manifest(remote, config, target, expected_manifest_hash=manifest["manifestHash"]) assert not (target / "source").exists() assert (target / "assets/hero.webp").read_bytes() == b"asset" assert json.loads((target / "manifest.json").read_text(encoding="utf-8"))["schemaVersion"] == "1.0" changed = dict(manifest) changed["sourceRevision"] = dict(manifest["sourceRevision"]) changed["sourceRevision"]["revisionId"] = "m4-r2" changed["sourceRevision"]["manifestRef"] = "tier2-source-project:night-market:m4-r2" unsigned = dict(changed) unsigned.pop("manifestHash") changed["manifestHash"] = store._hash_bytes(store._MANIFEST_HASH_DOMAIN + store._canonical_json(unsigned)) puts_before = list(client.put_calls) marker_before = client.objects[("game-artifacts", first["artifactManifestKey"])] with pytest.raises(store.ArtifactError, match="拒绝覆盖"): store.upload_manifest(game, changed, config, external_single_writer=True) assert client.put_calls == puts_before assert client.objects[("game-artifacts", first["artifactManifestKey"])] == marker_before def test_partial_objects_require_external_single_writer(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """波次 1 可覆盖无 marker 的残留 partial，但必须由调用方持有外部单写锁。""" _make_game(tmp_path) manifest = _build(tmp_path) client = _FakeMinio() asset = next(item for item in manifest["objects"] if item["category"] == "asset") client.objects[("game-artifacts", asset["key"])] = b"stale-partial" monkeypatch.setattr(store, "_minio_client", lambda _config: client) config = {"artifact_bucket": "game-artifacts", "evidence_bucket": "game-evidence"} with pytest.raises(store.ArtifactError, match="外部单写"): store.upload_manifest(tmp_path, manifest, config) assert client.objects[("game-artifacts", asset["key"])] == b"stale-partial" receipt = store.upload_manifest(tmp_path, manifest, config, external_single_writer=True) assert receipt["status"] == "verified_pending" assert receipt["committed"] is False assert client.objects[("game-artifacts", asset["key"])] == b"asset" def test_artifact_upload_preflights_runtime_manifest_before_put(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): _make_game(tmp_path) manifest = _build(tmp_path) (tmp_path / "game-package.json").write_text("drift", encoding="utf-8") client = _FakeMinio() monkeypatch.setattr(store, "_minio_client", lambda _config: client) with pytest.raises(store.ArtifactError, match="GamePackage manifest 本地 hash 漂移"): store.upload_manifest(tmp_path, manifest, { "artifact_bucket": "game-artifacts", "evidence_bucket": "game-evidence", }, external_single_writer=True) assert client.put_calls == [] def test_trusted_anchor_and_atomic_materialization(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): game = tmp_path / "game" _make_game(game) manifest = _build(game) client = _FakeMinio() monkeypatch.setattr(store, "_minio_client", lambda _config: client) config = {"artifact_bucket": "game-artifacts", "evidence_bucket": "game-evidence"} receipt = store.upload_manifest(game, manifest, config, external_single_writer=True) with pytest.raises(store.ArtifactError, match="数据库期望 hash"): store.fetch_manifest( config, key=receipt["artifactManifestKey"], expected_manifest_hash="0" * 64, expected_tenant_id=TENANT_ID, expected_game_id=GAME_ID, expected_version_id=VERSION_ID, ) remote = _trusted_fetch(config, receipt["artifactManifestKey"], manifest) asset = next(item for item in manifest["objects"] if item["category"] == "asset") client.objects[("game-artifacts", asset["key"])] = b"wrong" target = tmp_path / "materialized" with pytest.raises(store.ArtifactError, match="下载对象 hash"): store.download_manifest(remote, config, target, expected_manifest_hash=manifest["manifestHash"]) assert not target.exists() assert not list(tmp_path.glob(".materialized.partial-*")) def test_fetch_rejects_untrusted_key(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(store, "_minio_client", lambda _config: _FakeMinio()) with pytest.raises(store.ArtifactError, match="key 非法"): store.fetch_manifest( {}, key="https://attacker.invalid/artifact-manifest.json", expected_manifest_hash="0" * 64, expected_tenant_id=TENANT_ID, expected_game_id=GAME_ID, expected_version_id=VERSION_ID, ) def test_upload_never_probes_or_creates_bucket(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): _make_game(tmp_path) client = _FakeMinio() monkeypatch.setattr(store, "_minio_client", lambda _config: client) store.upload_manifest(tmp_path, _build(tmp_path), {}, external_single_writer=True) assert client.bucket_exists_calls == 0 assert client.make_bucket_calls == 0 def test_missing_runtime_manifest_is_rejected(tmp_path: Path): _make_game(tmp_path) with pytest.raises(store.ArtifactError, match="GamePackage manifest 不存在"): store.build_manifest( tmp_path, TENANT_ID, GAME_ID, VERSION_ID, package_type="phaser", runtime_manifest_path="missing.json", source_revision=_source_revision(tmp_path), ) def test_cli_build_reports_source_revision_hash(tmp_path: Path): """CLI 构建结果必须从 sourceRevision 输出源 hash，不能读取不存在的顶层字段。""" _make_game(tmp_path) source_revision = _source_revision(tmp_path) script = Path(__file__).resolve().parents[1] / "scripts/game_artifact_sync.py" result = subprocess.run( [ sys.executable, str(script), "--root", str(tmp_path), "--tenant-id", TENANT_ID, "--game-id", GAME_ID, "--version-id", VERSION_ID, "--package-type", "phaser", "--runtime-manifest", "game-package.json", "--source-provider", source_revision["provider"], "--source-game-id", source_revision["gameId"], "--source-revision-id", source_revision["revisionId"], "--source-manifest-ref", source_revision["manifestRef"], "--source-hash", source_revision["sourceHash"], ], check=False, capture_output=True, text=True, ) assert result.returncode == 0, result.stderr receipt = json.loads(result.stdout) assert receipt["sourceHash"] == source_revision["sourceHash"] def test_only_tier2_source_provider_is_supported(tmp_path: Path): """Tier0/1 尚无统一 SourceProjectStore hash 口径，禁止提前声明 provider。""" _make_game(tmp_path) revision = _source_revision(tmp_path) revision.update({ "provider": "studio_source_project", "manifestRef": "studio-source-project:night-market:m4-r1", }) with pytest.raises(store.ArtifactError, match="sourceRevision.provider"): store.build_manifest( tmp_path, TENANT_ID, GAME_ID, VERSION_ID, package_type="phaser", runtime_manifest_path="game-package.json", source_revision=revision, ) def test_engine_agnostic_source_provider_binds_binary_safe_hash(tmp_path: Path): """LittleJS/Canvas 等游戏可绑定独立源归档，但不能复用 Tier2 文本 hash。""" _make_game(tmp_path) source_hash = source_store.source_hash_from_files([ ("src/main.js", tmp_path / "src/main.js"), ]) manifest = store.build_manifest( tmp_path, TENANT_ID, GAME_ID, VERSION_ID, package_type="littlejs", runtime_manifest_path="game-package.json", source_revision={ "provider": "game_source_archive", "gameId": GAME_ID, "revisionId": "shanhai-r1", "manifestRef": f"game-source-archive:{GAME_ID}:shanhai-r1", "sourceHash": source_hash, }, ) store.verify_manifest(manifest) assert manifest["sourceRevision"]["provider"] == "game_source_archive" assert manifest["sourceRevision"]["sourceHash"] == source_hash def test_tier2_source_provider_cannot_bind_littlejs_package(tmp_path: Path): _make_game(tmp_path) with pytest.raises(store.ArtifactError, match="packageType=phaser"): store.build_manifest( tmp_path, TENANT_ID, GAME_ID, VERSION_ID, package_type="littlejs", runtime_manifest_path="game-package.json", source_revision=_source_revision(tmp_path), ) def test_schema_rejects_tier2_source_with_littlejs_package(tmp_path: Path): _make_game(tmp_path) manifest = _build(tmp_path) manifest["packageType"] = "littlejs" schema_path = Path(__file__).resolve().parents[3] / "contracts/game-artifact-manifest.schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) errors = list(Draft202012Validator(schema).iter_errors(manifest)) assert any("phaser" in error.message for error in errors) def test_remote_read_stops_at_expected_bytes_plus_one(): """远端响应超限时只多读一个字节，不把任意大对象完整载入内存。""" client = _FakeMinio() key = "tenants/1/games/1024/versions/2048/runtime/bundle.js" client.objects[("game-artifacts", key)] = b"0123456789" with pytest.raises(store.ArtifactError, match="超过期望字节数"): store._read_remote(client, "game-artifacts", key, expected_bytes=4) assert client.last_response is not None assert client.last_response._offset == 5 assert client.last_response.read_sizes == [5] def test_fetch_manifest_rejects_oversized_remote_payload(monkeypatch: pytest.MonkeyPatch): """artifact manifest 也必须有固定读取上限，不能因伪造 marker 耗尽内存。""" client = _FakeMinio() key = "tenants/1/games/1024/versions/2048/artifact-manifest.json" client.objects[("game-artifacts", key)] = b"x" * (8 * 1024 * 1024 + 1) monkeypatch.setattr(store, "_minio_client", lambda _config: client) with pytest.raises(store.ArtifactError, match="超过期望字节数"): store.fetch_manifest( {}, key=key, expected_manifest_hash="0" * 64, expected_tenant_id=TENANT_ID, expected_game_id=GAME_ID, expected_version_id=VERSION_ID, ) def test_missing_bucket_is_not_treated_as_missing_marker(): """桶缺失表示部署错误，不能降级成尚未上传。""" class MissingBucketClient: def get_object(self, _bucket: str, _key: str): raise _NoSuchBucket("game-artifacts") with pytest.raises(store.ArtifactError, match="对象存储桶不存在"): store._existing_manifest( MissingBucketClient(), "game-artifacts", "tenants/1/games/1024/versions/2048/artifact-manifest.json", )