144 lines
5.9 KiB
Python
144 lines
5.9 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "run-p1-07-6-isolated-uat.py"
|
|
SPEC = importlib.util.spec_from_file_location("p1_07_6_uat", SCRIPT)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
uat = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(uat)
|
|
|
|
|
|
def valid_env(tmp_path: Path) -> dict[str, str]:
|
|
env = {
|
|
"COMPOSE_PROJECT_NAME": "yuqei-p1-07-6-uat",
|
|
"POSTGRES_VOLUME_NAME": "yuqei-p1-07-6-uat-postgres",
|
|
"MINIO_VOLUME_NAME": "yuqei-p1-07-6-uat-rustfs-v1",
|
|
"CONTRACT_FILES_VOLUME_NAME": "yuqei-p1-07-6-uat-contract-files",
|
|
"MINIO_ROOT_USER": "synthetic-uat-access-key",
|
|
"MINIO_ROOT_PASSWORD": "synthetic-uat-secret-value",
|
|
"POSTGRES_IMAGE": "registry.local/postgres@sha256:" + "1" * 64,
|
|
"POSTGRES_CLIENT_IMAGE": "registry.local/postgres-client@sha256:" + "2" * 64,
|
|
"REDIS_IMAGE": "registry.local/redis@sha256:" + "3" * 64,
|
|
"MINIO_IMAGE": "registry.local/rustfs@sha256:" + "4" * 64,
|
|
"CONTRACT_API_IMAGE": "registry.local/contract-api@sha256:" + "5" * 64,
|
|
"CONTRACT_WEB_IMAGE": "registry.local/contract-web@sha256:" + "8" * 64,
|
|
"AI_PLATFORM_API_IMAGE": "registry.local/ai-platform-api@sha256:" + "6" * 64,
|
|
"AI_WORKER_IMAGE": "registry.local/ai-worker@sha256:" + "7" * 64,
|
|
"KMS_GATEWAY_IMAGE": "registry.local/kms@sha256:" + "a" * 64,
|
|
"EVIDENCE_WITNESS_GATEWAY_IMAGE": "registry.local/witness@sha256:" + "b" * 64,
|
|
"UAT_ALIYUN_KMS_REGION": "cn-hangzhou",
|
|
"UAT_ALIYUN_KMS_ENDPOINT": "kst-real.cryptoservice.kms.aliyuncs.com",
|
|
"UAT_ALIYUN_ENCRYPTION_KEY_ID": "aliyun-encryption",
|
|
"UAT_ALIYUN_EVIDENCE_KEY_ID": "aliyun-evidence",
|
|
"UAT_ALIYUN_EVIDENCE_KEY_VERSION": "v1",
|
|
"UAT_ALIYUN_WITNESS_KEY_ID": "aliyun-witness",
|
|
"UAT_ALIYUN_WITNESS_KEY_VERSION": "v1",
|
|
"UAT_ALIYUN_WITNESS_BUCKET": "aliyun-worm",
|
|
"UAT_ALIYUN_WITNESS_REGION": "cn-hangzhou",
|
|
"UAT_ALIYUN_WITNESS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
|
|
"UAT_TENCENT_KMS_REGION": "ap-guangzhou",
|
|
"UAT_TENCENT_ENCRYPTION_KEY_ID": "tencent-encryption",
|
|
"UAT_TENCENT_EVIDENCE_KEY_ID": "tencent-evidence",
|
|
"UAT_TENCENT_EVIDENCE_KEY_VERSION": "v1",
|
|
"UAT_TENCENT_WITNESS_KEY_ID": "tencent-witness",
|
|
"UAT_TENCENT_WITNESS_KEY_VERSION": "v1",
|
|
"UAT_TENCENT_WITNESS_BUCKET": "tencent-object-lock-123456",
|
|
"UAT_TENCENT_WITNESS_REGION": "ap-guangzhou",
|
|
"UAT_TENCENT_WITNESS_ENDPOINT": "https://cos.ap-guangzhou.myqcloud.com",
|
|
}
|
|
for key in (
|
|
"POSTGRES_BIND_ADDRESS", "REDIS_BIND_ADDRESS", "MINIO_API_BIND_ADDRESS",
|
|
"MINIO_CONSOLE_BIND_ADDRESS", "CONTRACT_API_BIND_ADDRESS",
|
|
"AI_PLATFORM_API_BIND_ADDRESS", "CONTRACT_WEB_BIND_ADDRESS",
|
|
):
|
|
env[key] = "127.0.0.1"
|
|
for key, port in zip(
|
|
("POSTGRES_HOST_PORT", "REDIS_HOST_PORT", "MINIO_API_HOST_PORT", "MINIO_CONSOLE_HOST_PORT", "CONTRACT_API_HOST_PORT", "AI_PLATFORM_API_HOST_PORT", "CONTRACT_WEB_HOST_PORT"),
|
|
("15432", "16379", "19000", "19001", "18100", "18101", "13000"),
|
|
strict=True,
|
|
):
|
|
env[key] = port
|
|
for index, key in enumerate(uat.SECRET_FILE_KEYS):
|
|
secret = tmp_path / f"secret-{index}"
|
|
secret.write_text(f"non-empty-{index}", encoding="utf-8")
|
|
env[key] = str(secret)
|
|
return env
|
|
|
|
|
|
def test_static_preflight_accepts_isolated_real_configuration(tmp_path: Path) -> None:
|
|
checks = uat.preflight(
|
|
valid_env(tmp_path),
|
|
admin_a="platform-admin-a",
|
|
admin_b="platform-admin-b",
|
|
static_only=True,
|
|
)
|
|
assert {item["name"] for item in checks if item["status"] == "failed"} == set()
|
|
|
|
|
|
def test_preflight_rejects_shared_admin_and_mutable_image(tmp_path: Path) -> None:
|
|
env = valid_env(tmp_path)
|
|
env["KMS_GATEWAY_IMAGE"] = "registry.local/kms:latest"
|
|
checks = uat.preflight(
|
|
env,
|
|
admin_a="shared-admin",
|
|
admin_b="shared-admin",
|
|
static_only=True,
|
|
)
|
|
failed = {item["name"] for item in checks if item["status"] == "failed"}
|
|
assert {"immutable_images", "two_platform_administrators"} <= failed
|
|
|
|
|
|
def _rendered_model(env: dict[str, str]) -> dict[str, object]:
|
|
services = {
|
|
service: {"image": env[env_key]}
|
|
for service, env_key in uat.UAT_SERVICE_IMAGE_ENV.items()
|
|
}
|
|
services["minio"].update(
|
|
{
|
|
"volumes": [
|
|
{"type": "volume", "source": "rustfs-data", "target": "/data"}
|
|
],
|
|
"entrypoint": ["/bin/sh", "-ec"],
|
|
"command": [
|
|
"set -eu\n"
|
|
"test ! -e /data/.minio.sys\n"
|
|
"marker=/data/.yuqei-rustfs-volume-v1\n"
|
|
"exec /entrypoint.sh rustfs\n"
|
|
],
|
|
"healthcheck": {
|
|
"test": ["CMD-SHELL", "curl -fsS http://127.0.0.1:9000/health/ready"]
|
|
},
|
|
"environment": {
|
|
"RUSTFS_ACCESS_KEY": env["MINIO_ROOT_USER"],
|
|
"RUSTFS_SECRET_KEY": env["MINIO_ROOT_PASSWORD"],
|
|
"RUSTFS_VOLUMES": "/data",
|
|
},
|
|
"user": "10001:10001",
|
|
"read_only": True,
|
|
"cap_drop": ["ALL"],
|
|
"security_opt": ["no-new-privileges:true"],
|
|
}
|
|
)
|
|
return {
|
|
"services": services,
|
|
"volumes": {"rustfs-data": {"name": env["MINIO_VOLUME_NAME"]}},
|
|
}
|
|
|
|
|
|
def test_rendered_uat_compose_is_bound_to_all_images_and_rustfs_volume(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env = valid_env(tmp_path)
|
|
model = _rendered_model(env)
|
|
uat.validate_rendered_compose(json.dumps(model), env)
|
|
|
|
model["services"]["minio"]["image"] = "registry.local/other@sha256:" + "f" * 64
|
|
with pytest.raises(uat.UatFailure, match="MINIO_IMAGE"):
|
|
uat.validate_rendered_compose(json.dumps(model), env)
|