yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
bf6beb30f67459f9f0c9f3e22e3a454dd7a9deef
ai-contract-platform/yuqei-infra/tests/test_production_release_tools.py
T

2525 lines
99 KiB
Python
Raw Blame History

from __future__ import annotations
import base64
import hashlib
import importlib.util
import io
import json
import signal
import shutil
import subprocess
import sys
import tarfile
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from threading import Thread
from types import ModuleType, SimpleNamespace
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from rc06_environment_test_support import rc06_environment_values
REPO_ROOT = Path(__file__).resolve().parents[2]
def load_script(name: str, relative_path: str) -> ModuleType:
path = REPO_ROOT / relative_path
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
preflight = load_script("yuqei_production_preflight", "yuqei-infra/deploy/preflight-production.py")
backup_run = load_script("yuqei_backup_run", "yuqei-infra/backup/run-production-backup.py")
backup_verify = load_script("yuqei_backup_verify", "yuqei-infra/backup/verify-production-backup.py")
snapshot_restore_planner = load_script(
"yuqei_snapshot_restore_planner_for_backup",
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
)
provider_acceptance = load_script("yuqei_provider_acceptance", "yuqei-infra/scripts/run-ai-provider-acceptance.py")
contract_product_golden = load_script(
"yuqei_contract_product_golden",
"yuqei-infra/scripts/run-contract-product-golden.py",
)
release_quality_gate = load_script(
"yuqei_release_quality_gate",
"yuqei-infra/scripts/run-release-quality-gate.py",
)
from yuqei_ai_platform_api.provider_acceptance_receipt import ( # noqa: E402
RECEIPT_DOMAIN,
ProviderAcceptanceReceiptPayload,
SignedProviderAcceptanceReceipt,
canonical_receipt_payload,
)
def production_values() -> dict[str, str]:
values = preflight.parse_env_file(REPO_ROOT / "yuqei-infra/compose/.env.production.example")
replacements = {
"RELEASE_ID": "v2.0.0-rc.1",
"YUQEI_RELEASE_ENVIRONMENT_ID": "production-primary",
"YUQEI_RELEASE_HOST_ID": "production-host-01",
"CONTRACT_API_IMAGE": f"registry.internal/contract-api@sha256:{'1' * 64}",
"CONTRACT_WEB_IMAGE": f"registry.internal/contract-web@sha256:{'c' * 64}",
"AI_PLATFORM_API_IMAGE": f"registry.internal/ai-platform-api@sha256:{'2' * 64}",
"AI_PLATFORM_RELEASE_IMAGE_DIGEST": f"sha256:{'2' * 64}",
"CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST": f"sha256:{'1' * 64}",
"AI_WORKER_IMAGE": f"registry.internal/ai-worker@sha256:{'3' * 64}",
"POSTGRES_IMAGE": f"registry.internal/pgvector@sha256:{'4' * 64}",
"POSTGRES_CLIENT_IMAGE": f"registry.internal/postgres@sha256:{'5' * 64}",
"REDIS_IMAGE": f"registry.internal/redis@sha256:{'6' * 64}",
"MINIO_IMAGE": f"registry.internal/rustfs@sha256:{'7' * 64}",
"MINIO_MC_IMAGE": f"registry.internal/rustfs-rc@sha256:{'8' * 64}",
"NGINX_IMAGE": f"registry.internal/nginx@sha256:{'9' * 64}",
"BACKUP_UTILITY_IMAGE": f"registry.internal/alpine@sha256:{'a' * 64}",
"BACKEND_SERVER_NAME": "api.contracts.acme.cn",
"BACKEND_PUBLIC_BASE_URL": "https://api.contracts.acme.cn",
"POSTGRES_USER": "yuqei_prod",
"POSTGRES_PASSWORD": "postgres-secret-that-is-long-and-unique-001",
"MINIO_ROOT_USER": "yuqei_prod_minio",
"MINIO_ROOT_PASSWORD": "minio-secret-that-is-long-and-unique-002",
"CONTRACT_API_WEB_PROXY_SHARED_SECRET": "proxy-secret-that-is-long-and-unique-003",
"CONTRACT_API_IDENTITY_RATE_LIMIT_SECRET": "identity-secret-that-is-long-and-unique-004",
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET": "contract-anti-abuse-secret-that-is-long-004a", # pragma: allowlist secret
"CONTRACT_WEB_SESSION_SECRET": "session-secret-that-is-long-and-unique-005",
"AI_PLATFORM_SERVICE_SHARED_SECRET": "ai-service-secret-that-is-long-and-unique-006",
"KMS_GATEWAY_IMAGE": f"registry.internal/kms-gateway@sha256:{'b' * 64}",
"KMS_GATEWAY_ALLOWED_KEY_IDS": "calendar-envelope-key-v1",
"KMS_GATEWAY_PRIMARY_KEY_ID": "calendar-envelope-key-v1",
"KMS_GATEWAY_ALLOWED_SIGNING_KEY_IDS": "reliable-evidence-signing-key-v1",
"KMS_GATEWAY_SIGNING_KEY_ID": "reliable-evidence-signing-key-v1",
"KMS_GATEWAY_SIGNING_KEY_VERSION": "version-2026-08-25",
"ALIYUN_KMS_ENDPOINT": "kst-production.cryptoservice.kms.aliyuncs.com",
"AI_PLATFORM_GENERATIVE_AI_FILING_NUMBER": "内网信算备-TEST-001",
"AI_PLATFORM_GENERATIVE_AI_SERVICE_REGISTRATION_NUMBER": "内网信生测评-TEST-001",
"AI_PLATFORM_GENERATIVE_AI_SERVICE_PROVIDER_CODE": "yuqei-test-provider-code",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256": hashlib.sha256(
b"pinned-production-provider-acceptance-receipt"
).hexdigest(),
"PRODUCTION_ALERT_WEBHOOK_URL_FILE": "/etc/yuqei/monitoring/alert-webhook-url",
"AI_PLATFORM_ANTI_ABUSE_HASH_SECRET": "ai-anti-abuse-secret-that-is-long-and-unique-006a", # pragma: allowlist secret
"CONTRACT_API_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_contract",
"AI_PLATFORM_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_ai",
"YUQEI_WORKER_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_ai",
}
values.update(replacements)
values.update(
rc06_environment_values(
contract_digest="sha256:" + "1" * 64,
ai_digest=values["AI_PLATFORM_RELEASE_IMAGE_DIGEST"],
)
)
return values
def test_object_storage_backup_injects_rc_config_via_stdin_without_secret_argv(
tmp_path: Path,
) -> None:
partial_dir = tmp_path / ".backup-20260811T120000Z.partial"
values = {
"COMPOSE_PROJECT_NAME": "yuqei-v2-production",
"AI_PLATFORM_OBJECT_STORAGE_BUCKET": "ai-platform-uploads",
"MINIO_ROOT_USER": "backup-access-key",
"MINIO_ROOT_PASSWORD": "backup-secret-value",
"MINIO_MC_IMAGE": f"rustfs/rc@sha256:{'a' * 64}",
}
command = backup_run.object_storage_mirror_command(
values=values,
partial_dir=partial_dir,
)
config = backup_run._rc_config_bytes(values).decode("utf-8")
assert command[:2] == ["docker", "run"]
assert command[command.index("--pull") + 1] == "never"
assert "--interactive" in command
assert command[command.index("--entrypoint") + 1] == "/bin/sh"
assert "--env-file" not in command
assert "HOME=/tmp/rc-home" in command
assert command[-3:] == [
"--",
"source/ai-platform-uploads/",
"/backup/minio/",
]
read_only_index = command.index("--read-only")
assert ["--read-only", "--cap-drop", "ALL"] == command[
read_only_index : read_only_index + 3
]
assert "no-new-privileges:true" in command
assert "cat > \"$HOME/.config/rc/config.toml\"" in command[-4]
assert "exec rc --no-color --no-progress mirror --overwrite" in command[-4]
assert all("backup-secret-value" not in token for token in command)
assert all("backup-access-key" not in token for token in command)
assert 'name = "source"' in config
assert 'endpoint = "http://minio:9000"' in config
assert 'access_key = "backup-access-key"' in config
assert 'secret_key = "backup-secret-value"' in config
def test_object_storage_backup_rejects_unsafe_bind_source(tmp_path: Path) -> None:
with pytest.raises(RuntimeError, match="unsafe character"):
backup_run._docker_bind_source(tmp_path / "comma,path")
def test_backup_quiesces_every_long_running_application_writer() -> None:
assert backup_run.WRITER_SERVICES == (
"contract-api",
"contract-review-worker",
"contract-playbook-governance-worker",
"ai-platform-api",
"worker",
)
@pytest.mark.parametrize(
("label", "validation_name", "table_name"),
[
("contract", "contract_api_alembic_version", "contract_api_alembic_version"),
("ai_platform", "ai_platform_alembic_version", "ai_platform_alembic_version"),
("operations", "public_schema_fingerprint", "information_schema.columns"),
],
)
def test_restore_check_uses_each_database_owned_schema_marker(
label: str,
validation_name: str,
table_name: str,
) -> None:
actual_name, query = backup_verify.restore_validation_query(label)
assert actual_name == validation_name
assert table_name in query
def test_production_preflight_accepts_strict_configuration() -> None:
checks = preflight.validate_environment(production_values())
assert [check.key for check in checks if check.status == "fail"] == []
def test_production_launch_scope_excludes_support_access_and_identity_execution() -> None:
manifest = json.loads(preflight.LAUNCH_SCOPE_MANIFEST_PATH.read_text(encoding="utf-8"))
assert manifest == preflight.EXPECTED_LAUNCH_SCOPE
assert manifest["decision"] == "excluded"
assert manifest["support_access"] == {
"included": False,
"enabled": False,
"kill_switch_engaged": True,
"executor_enabled": False,
}
assert manifest["identity_executor"] == {"enabled": False}
assert manifest["destructive_identity_write_api"]["included"] is False
assert set(manifest["destructive_identity_write_api"]["forbidden_operations"]) == {
"execute",
"grant",
"revoke",
"impersonate",
"impersonation",
}
@pytest.mark.parametrize(
"activation_key",
sorted(preflight.FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS),
)
def test_production_preflight_rejects_support_access_or_identity_write_activation(
activation_key: str,
) -> None:
values = production_values()
values[activation_key] = "true"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["launch_scope_activation_flags"].status == "fail"
assert activation_key in checks["launch_scope_activation_flags"].message
@pytest.mark.parametrize("artifact_state", ["missing", "tampered", "invalid"])
def test_production_preflight_fails_closed_on_launch_scope_artifact_drift(
tmp_path: Path,
artifact_state: str,
) -> None:
manifest_path = tmp_path / "launch-scope.json"
if artifact_state == "tampered":
tampered = dict(preflight.EXPECTED_LAUNCH_SCOPE)
tampered["support_access"] = {
**tampered["support_access"],
"enabled": True,
}
manifest_path.write_text(json.dumps(tampered), encoding="utf-8")
elif artifact_state == "invalid":
manifest_path.write_text("not-json", encoding="utf-8")
checks = {
check.key: check
for check in preflight._validate_launch_scope_exclusion(
production_values(),
manifest_path=manifest_path,
)
}
assert checks["launch_scope_support_access_exclusion"].status == "fail"
@pytest.mark.parametrize(
("contract_timeout", "provider_timeout"),
[("30", "30"), ("0", "30"), ("601", "30"), ("NaN", "30"), ("60", "inf")],
)
def test_production_preflight_rejects_invalid_ai_timeout_budgets(
contract_timeout: str,
provider_timeout: str,
) -> None:
values = production_values()
values["CONTRACT_API_AI_PLATFORM_TIMEOUT_SECONDS"] = contract_timeout
values["AI_PLATFORM_PROVIDER_REQUEST_TIMEOUT_SECONDS"] = provider_timeout
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["ai_timeout_budget"].status == "fail"
def test_production_preflight_rejects_promotion_image_digest_drift() -> None:
values = production_values()
values["AI_PLATFORM_RELEASE_IMAGE_DIGEST"] = f"sha256:{'f' * 64}"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["provider_promotion_image_binding"].status == "fail"
@pytest.mark.parametrize(
"forbidden_key",
[
"AI_PLATFORM_PROVIDER_LEGAL_ATTESTATION_SECRET",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_SIGNING_PRIVATE_KEY_FILE",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_SIGNING_KEY",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_STAGING_DATABASE_URL",
"SHARED_STAGING_POSTGRES_PASSWORD",
],
)
def test_production_preflight_rejects_promotion_private_or_staging_credentials(
forbidden_key: str,
) -> None:
values = production_values()
values[forbidden_key] = "must-not-enter-production"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["provider_acceptance_credential_isolation"].status == "fail"
assert forbidden_key in checks["provider_acceptance_credential_isolation"].message
@pytest.mark.parametrize(
("key", "value"),
[
("RELEASE_ID", "release id with spaces"),
("RELEASE_ID", "../release"),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256", "0" * 64),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256", "A" * 64),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_MAX_TTL_SECONDS", "604801"),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_MAX_TTL_SECONDS", "not-a-number"),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_CLOCK_SKEW_SECONDS", "301"),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_CLOCK_SKEW_SECONDS", "-1"),
("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST", "relative/receipt.json"),
(
"AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST",
"/etc/yuqei/provider-acceptance/receipt.json",
),
],
)
def test_production_preflight_rejects_unpinned_or_ambiguous_promotion_config(
key: str,
value: str,
) -> None:
values = production_values()
values[key] = value
failed = {check.key for check in preflight.validate_environment(values) if check.status == "fail"}
assert failed & {"release_id", "provider_acceptance_deployment"}
def test_provider_acceptance_promotion_files_are_present_and_hash_pinned(tmp_path: Path) -> None:
receipt_path = tmp_path / "receipt.json"
receipt_path.write_bytes(b'{"receipt":"signed"}\n')
trusted_keys_path = tmp_path / "trusted-keys.json"
trusted_keys_path.write_bytes(b'{"version":1,"keys":[]}\n')
values = production_values()
values.update(
{
"AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST": str(receipt_path.resolve()),
"AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST": str(
trusted_keys_path.resolve()
),
"AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256": hashlib.sha256(
receipt_path.read_bytes()
).hexdigest(),
}
)
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["provider_acceptance_receipt_file"].status == "pass"
assert checks["provider_acceptance_receipt_hash"].status == "pass"
assert checks["provider_acceptance_trusted_keys_file"].status == "pass"
receipt_path.write_bytes(b'{"receipt":"tampered"}\n')
tampered = {check.key: check for check in preflight.validate_files(values)}
assert tampered["provider_acceptance_receipt_hash"].status == "fail"
def test_production_preflight_requires_owner_only_https_alert_webhook(tmp_path: Path) -> None:
values = production_values()
webhook = tmp_path / "alert-webhook-url"
webhook.write_text("https://alerts.acme.cn/hooks/yuqei\n", encoding="utf-8")
webhook.chmod(0o600)
values["PRODUCTION_ALERT_WEBHOOK_URL_FILE"] = str(webhook.resolve())
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["alert_webhook_file"].status == "pass"
if "alert_webhook_permissions" in checks:
assert checks["alert_webhook_permissions"].status == "pass"
webhook.write_text("http://alerts.acme.cn/hooks/yuqei\n", encoding="utf-8")
assert {check.key: check for check in preflight.validate_files(values)}["alert_webhook_file"].status == "fail"
def test_production_preflight_rejects_missing_alert_webhook_file(tmp_path: Path) -> None:
values = production_values()
values["PRODUCTION_ALERT_WEBHOOK_URL_FILE"] = str((tmp_path / "missing").resolve())
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["alert_webhook_file"].status == "fail"
def test_production_compose_mounts_only_promotion_verification_material_read_only() -> None:
override = (
REPO_ROOT / "yuqei-infra/compose/docker-compose.production.yml"
).read_text(encoding="utf-8")
assert "AI_PLATFORM_RELEASE_ID: ${RELEASE_ID:?" in override
assert "AI_PLATFORM_RELEASE_IMAGE_DIGEST: ${AI_PLATFORM_RELEASE_IMAGE_DIGEST:?" in override
assert "AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256: ${" in override
assert ":/run/yuqei/provider-acceptance/receipt.json:ro" in override
assert ":/run/yuqei/provider-acceptance/trusted-keys.json:ro" in override
assert "CONTRACT_API_PUBLIC_POLICY_CANDIDATE_DIR: /run/yuqei/public-policy/candidate" in override
assert "${PUBLIC_POLICY_CANDIDATE_DIR_HOST:?PUBLIC_POLICY_CANDIDATE_DIR_HOST is required}:/run/yuqei/public-policy/candidate:ro" in override
assert "${PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST:?PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST is required}:/run/yuqei/public-policy/named-policy-legal-approval.json:ro" in override
assert "${PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST:?PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST is required}:/run/yuqei/public-policy/policy-legal-keyring.trusted.json:ro" in override
assert "CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED: ${CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED:-false}" in override
assert 'AI_PLATFORM_PROVIDER_LEGAL_ATTESTATION_SECRET: ""' in override
assert "SIGNING_PRIVATE_KEY" not in override
assert "STAGING_DATABASE_URL" not in override
def test_production_prompt_asset_import_is_dry_run_by_default() -> None:
override = (
REPO_ROOT / "yuqei-infra/compose/docker-compose.production.yml"
).read_text(encoding="utf-8")
service = override.split(" ai-platform-prompt-assets-import:", 1)[1].split(
"\n ai-platform-api:", 1
)[0]
command_line = next(line.strip() for line in service.splitlines() if line.strip().startswith("command:"))
assert command_line == 'command: ["python", "-m", "yuqei_ai_platform_api.packaged_prompt_assets_import"]'
def _write_public_policy_evidence(tmp_path: Path) -> dict[str, str]:
candidate = tmp_path / "public-policy"
candidate.mkdir()
documents = []
for policy_id, source_file in (
("user_terms", "user-terms.md"),
("privacy_policy", "privacy-policy.md"),
("service_agreement", "service-agreement.md"),
("ai_use_notice", "ai-use-notice.md"),
):
(candidate / source_file).write_text(f"# {policy_id}\n", encoding="utf-8")
documents.append({"policy_id": policy_id, "source_file": source_file})
(candidate / "manifest.json").write_text(
json.dumps(
{
"schema_version": "yuqei-public-policy-candidate-v1",
"existing_user_reconsent": "required_on_next_login",
"documents": documents,
},
ensure_ascii=False,
),
encoding="utf-8",
)
approval = tmp_path / "policy-approval.json"
approval.write_text('{"approval_id":"policy-approval"}', encoding="utf-8")
keyring = tmp_path / "policy-keyring.json"
keyring.write_text(
json.dumps({"trust_basis": {"policy_activation_authorized": False}}),
encoding="utf-8",
)
return {
"PUBLIC_POLICY_CANDIDATE_DIR_HOST": str(candidate.resolve()),
"PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST": str(approval.resolve()),
"PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST": str(keyring.resolve()),
}
def test_production_preflight_rejects_invalid_public_policy_deployment() -> None:
values = production_values()
values["CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST"] = f"sha256:{'f' * 64}"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_deployment"].status == "fail"
@pytest.mark.parametrize(
("key", "value"),
[
("PUBLIC_POLICY_CANDIDATE_DIR_HOST", "relative/public-policy"),
("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", "/etc/yuqei/public-policy/2026-08-12/policy-legal-keyring.trusted.json"),
("CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED", "true"),
],
)
def test_production_preflight_rejects_unsafe_public_policy_paths_or_executor(
key: str,
value: str,
) -> None:
values = production_values()
values[key] = value
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_deployment"].status == "fail"
def test_production_preflight_rejects_public_policy_private_material() -> None:
values = production_values()
values["CONTRACT_API_PUBLIC_POLICY_PRIVATE_KEY_FILE"] = "/etc/yuqei/private.pem"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_credential_isolation"].status == "fail"
def test_public_policy_evidence_files_are_present_and_safe(tmp_path: Path) -> None:
values = production_values()
values.update(_write_public_policy_evidence(tmp_path))
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["public_policy_candidate_dir"].status == "pass"
assert checks["public_policy_manifest"].status == "pass"
assert checks["public_policy_legal_approval_file"].status == "pass"
assert checks["public_policy_trusted_keyring_file"].status == "pass"
def test_public_policy_preflight_fails_on_manifest_or_keyring_drift(tmp_path: Path) -> None:
values = production_values()
values.update(_write_public_policy_evidence(tmp_path))
candidate = Path(values["PUBLIC_POLICY_CANDIDATE_DIR_HOST"])
(candidate / "manifest.json").write_text('{"schema_version":"wrong"}', encoding="utf-8")
keyring = Path(values["PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST"])
keyring.write_text(
json.dumps({"trust_basis": {"policy_activation_authorized": True}}),
encoding="utf-8",
)
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["public_policy_manifest"].status == "fail"
assert checks["public_policy_trusted_keyring_file"].status == "fail"
def test_shared_compose_wires_contract_anti_abuse_controls() -> None:
compose = (REPO_ROOT / "yuqei-infra/compose/docker-compose.yml").read_text(
encoding="utf-8"
)
for variable in (
"CONTRACT_API_ANTI_ABUSE_MODE",
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET",
"CONTRACT_API_ANTI_ABUSE_S3_RECORDS_DAILY_LIMIT",
"CONTRACT_API_ANTI_ABUSE_S4_RECORDS_DAILY_LIMIT",
"CONTRACT_API_ANTI_ABUSE_FILES_DAILY_LIMIT",
"CONTRACT_API_ANTI_ABUSE_BYTES_DAILY_LIMIT",
):
assert f"{variable}: ${{{variable}:-" in compose
def test_shared_compose_and_smoke_deploy_wire_contract_ai_timeout() -> None:
compose = (REPO_ROOT / "yuqei-infra/compose/docker-compose.yml").read_text(
encoding="utf-8"
)
deploy_script = (
REPO_ROOT / "yuqei-infra/scripts/deploy-v2-smoke-docker.ps1"
).read_text(encoding="utf-8")
assert compose.count("CONTRACT_API_AI_PLATFORM_TIMEOUT_SECONDS: ${") == 3
assert "CONTRACT_API_AI_PLATFORM_TIMEOUT_SECONDS=240" in deploy_script
assert "AI_PLATFORM_PROVIDER_REQUEST_TIMEOUT_SECONDS=180" in deploy_script
def test_release_gate_benchmark_defaults_fit_enforced_anti_abuse_window(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(sys, "argv", ["run-release-quality-gate.py"])
args = release_quality_gate.parse_args()
assert args.benchmark_requests_per_scenario + args.benchmark_warmup + 1 < 20
assert 1 <= args.benchmark_concurrency <= args.benchmark_requests_per_scenario
def test_legal_qa_edge_timeout_covers_contract_client_maximum() -> None:
nginx = (
REPO_ROOT / "yuqei-infra/nginx/templates/default.conf.template"
).read_text(encoding="utf-8")
location = nginx.split("location = /contract/api/v1/legal/qa", 1)[1].split("}", 1)[0]
assert "proxy_read_timeout 660s;" in location
assert "proxy_send_timeout 660s;" in location
assert "proxy_pass http://contract_api_upstream/api/v1/legal/qa;" in location
def test_smoke_review_requests_include_server_authorized_identity_context() -> None:
deploy_script = (
REPO_ROOT / "yuqei-infra/scripts/deploy-v2-smoke-docker.ps1"
).read_text(encoding="utf-8")
assert deploy_script.count("-H 'X-Tenant-Id: default'") >= 3
assert deploy_script.count("-H 'X-User-Id: user-current-admin'") >= 3
def test_disabled_identity_email_does_not_require_smtp_secret_file() -> None:
values = production_values()
values.pop("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST", None)
environment_checks = {check.key: check for check in preflight.validate_environment(values)}
file_checks = {check.key: check for check in preflight.validate_files(values)}
assert environment_checks["identity_email_delivery"].status == "pass"
assert file_checks["smtp_password_file"].status == "pass"
def test_smtp_identity_email_requires_and_validates_opt_in_secret_file(tmp_path: Path) -> None:
password_file = tmp_path / "contract-smtp-password"
password_file.write_text("test-smtp-password\n", encoding="utf-8")
password_file.chmod(0o600)
values = production_values()
values.update(
{
"CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE": "smtp",
"CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED": "true",
"CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED": "false",
"CONTRACT_API_IDENTITY_SMTP_HOST": "smtp.mail.acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USERNAME": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD": "",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE": "/run/secrets/contract_smtp_password",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST": str(password_file.resolve()),
"CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS": "true",
"CONTRACT_API_IDENTITY_SMTP_USE_SSL": "false",
"CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL": "https://contracts.acme.cn",
}
)
environment_checks = {check.key: check for check in preflight.validate_environment(values)}
file_checks = {check.key: check for check in preflight.validate_files(values)}
assert environment_checks["identity_email_delivery"].status == "pass"
assert file_checks["smtp_password_file"].status == "pass"
if "smtp_password_permissions" in file_checks:
assert file_checks["smtp_password_permissions"].status == "pass"
unencrypted = dict(values)
unencrypted["CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS"] = "false"
unencrypted["CONTRACT_API_IDENTITY_SMTP_USE_SSL"] = "false"
unencrypted_checks = {
check.key: check for check in preflight.validate_environment(unencrypted)
}
assert unencrypted_checks["identity_email_delivery"].status == "fail"
password_file.unlink()
missing_file_checks = {check.key: check for check in preflight.validate_files(values)}
assert missing_file_checks["smtp_password_file"].status == "fail"
@pytest.mark.parametrize(
("key", "value"),
[
("CONTRACT_API_IDENTITY_SMTP_PORT", "0"),
("CONTRACT_API_IDENTITY_SMTP_PORT", "not-a-port"),
("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS", "0"),
("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS", "121"),
("CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL", "not-an-email"),
("CONTRACT_API_IDENTITY_SMTP_HOST", " smtp.mail.acme.cn"),
("CONTRACT_API_IDENTITY_SMTP_FROM_NAME", "AI Contract\r\nBcc: hidden@example.com"),
("CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS", "sometimes"),
],
)
def test_smtp_preflight_rejects_runtime_invalid_values(
tmp_path: Path,
key: str,
value: str,
) -> None:
password_file = tmp_path / "contract-smtp-password"
password_file.write_text("test-smtp-password\n", encoding="utf-8")
values = production_values()
values.update(
{
"CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE": "smtp",
"CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED": "true",
"CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED": "false",
"CONTRACT_API_IDENTITY_SMTP_HOST": "smtp.mail.acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USERNAME": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD": "",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE": "/run/secrets/contract_smtp_password",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST": str(password_file.resolve()),
"CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS": "true",
"CONTRACT_API_IDENTITY_SMTP_USE_SSL": "false",
"CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS": "10",
"CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL": "https://contracts.acme.cn",
key: value,
}
)
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["identity_email_delivery"].status == "fail"
def test_smtp_production_override_mounts_only_the_password_file_secret() -> None:
override = (
REPO_ROOT / "yuqei-infra/compose/docker-compose.production.smtp.yml"
).read_text(encoding="utf-8")
assert "CONTRACT_API_IDENTITY_SMTP_PASSWORD: \"\"" in override
assert "CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE: /run/secrets/contract_smtp_password" in override
assert "target: contract_smtp_password" in override
assert "file: ${CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST:?" in override
def test_production_preflight_rejects_smoke_database_public_bind_and_fallback() -> None:
values = production_values()
values["CONTRACT_POSTGRES_DB"] = "yuqei_contract_smoke"
values["CONTRACT_API_DATABASE_URL"] = (
"postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_contract_smoke"
)
values["POSTGRES_BIND_ADDRESS"] = "0.0.0.0"
values["AI_PLATFORM_PROVIDER_FALLBACK_ENABLED"] = "true"
failed = {check.key for check in preflight.validate_environment(values) if check.status == "fail"}
assert {"production_modes", "internal_port_binding", "database_boundaries"} <= failed
def _complete_backup_fixture(tmp_path: Path) -> tuple[Path, dict[str, object]]:
backup_dir = tmp_path / "backup-20260716T120000Z"
backup_dir.mkdir(parents=True)
archive_path = backup_dir / "contract-files.tar.gz"
with tarfile.open(archive_path, mode="w:gz") as archive:
content = b"contract"
info = tarfile.TarInfo("contract-1/document.txt")
info.size = len(content)
archive.addfile(info, io.BytesIO(content))
for label in ("contract", "ai_platform", "operations"):
(backup_dir / f"postgres-{label}.dump").write_bytes(
f"postgres-{label}-dump".encode("utf-8")
)
minio_object = backup_dir / "minio" / "tenant-a" / "object.bin"
minio_object.parent.mkdir(parents=True)
minio_object.write_bytes(b"minio-object")
artifacts = []
for path in sorted(backup_dir.rglob("*")):
if not path.is_file():
continue
artifacts.append(
{
"path": path.relative_to(backup_dir).as_posix(),
"size_bytes": path.stat().st_size,
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
}
)
minio_records = [
item for item in artifacts if str(item["path"]).startswith("minio/")
]
manifest: dict[str, object] = {
"schema_version": "yuqei-production-backup-artifacts.v2",
"backup_id": backup_dir.name,
"created_at": "2026-07-16T12:00:00+00:00",
"release_id": "release-2026-07-16",
"compose_project_name": "yuqei-production",
"consistency_mode": "quiesced",
"database_names": {
"contract": "yuqei_contract",
"ai_platform": "yuqei_ai",
"operations": "yuqei_ops",
},
"images": {
"contract_api": f"registry.internal/contract-api@sha256:{'1' * 64}",
"ai_platform_api": f"registry.internal/ai-platform-api@sha256:{'2' * 64}",
"ai_worker": f"registry.internal/ai-worker@sha256:{'3' * 64}",
},
"object_storage_mirror": {
"engine": "rustfs",
"engine_version": "1.0.0",
"engine_image": f"registry.internal/rustfs@sha256:{'7' * 64}",
"client": "rustfs-rc",
"client_version": "0.1.36",
"client_image": f"registry.internal/rustfs-rc@sha256:{'8' * 64}",
"engine_runtime_image": f"registry.internal/rustfs@sha256:{'7' * 64}",
"engine_resolved_image_id": f"sha256:{'9' * 64}",
"client_resolved_image_id": f"sha256:{'a' * 64}",
"bucket": "ai-platform-uploads",
"path": "minio",
"object_count": len(minio_records),
"total_size_bytes": sum(int(item["size_bytes"]) for item in minio_records),
"content_index_sha256": backup_run._minio_content_index(minio_records),
},
"artifacts": artifacts,
}
(backup_dir / "manifest.json").write_text(
json.dumps(manifest),
encoding="utf-8",
)
return backup_dir, manifest
def _restore_check_fixture(source: dict[str, object]) -> list[dict[str, str]]:
database_names = source["database_names"]
assert isinstance(database_names, dict)
return [
{
"label": "contract",
"database": database_names["contract"],
"restore_database": "yuqei_restore_check_contract_20260927010000",
"validation": "contract_api_alembic_version",
"validation_value": "20260926_0081",
"schema_revision": "20260926_0081",
},
{
"label": "ai_platform",
"database": database_names["ai_platform"],
"restore_database": "yuqei_restore_check_ai_platform_20260927010000",
"validation": "ai_platform_alembic_version",
"validation_value": "20260925_0042",
"schema_revision": "20260925_0042",
},
{
"label": "operations",
"database": database_names["operations"],
"restore_database": "yuqei_restore_check_operations_20260927010000",
"validation": "public_schema_fingerprint",
"validation_value": "a" * 32,
"schema_revision": "a" * 32,
},
]
def test_backup_manifest_and_contract_archive_are_verified(tmp_path: Path) -> None:
backup_dir, _ = _complete_backup_fixture(tmp_path)
generated_manifest = backup_run.verify_manifest(backup_dir)
manifest, verified = backup_verify.validate_backup(backup_dir)
assert generated_manifest["backup_id"] == backup_dir.name
assert manifest["backup_id"] == backup_dir.name
assert len(verified) == 5
def test_artifacts_v2_adapter_is_consumed_by_first_release_recovery_planner(
tmp_path: Path,
) -> None:
backup_dir, source = _complete_backup_fixture(tmp_path)
source, verified = backup_verify.validate_backup(backup_dir)
restored = _restore_check_fixture(source)
adapted = backup_verify.build_planner_backup_manifest(
backup_dir=backup_dir,
source_manifest=source,
restore_checks=restored,
environment_id="isolated-uat-01",
consistency_set_id="consistency-backup-20260716T120000Z",
)
adapted_path = tmp_path / "consistent-backup-manifest.json"
adapted_path.write_text(json.dumps(adapted), encoding="utf-8")
record = {
"path": str(adapted_path.resolve()),
"sha256": hashlib.sha256(adapted_path.read_bytes()).hexdigest(),
}
application_images = {
"CONTRACT_API_IMAGE": source["images"]["contract_api"],
"AI_PLATFORM_API_IMAGE": source["images"]["ai_platform_api"],
"AI_WORKER_IMAGE": source["images"]["ai_worker"],
}
object_storage = source["object_storage_mirror"]
infrastructure_images = {
"MINIO_IMAGE": object_storage["engine_image"],
"MINIO_MC_IMAGE": object_storage["client_image"],
}
package_binding = {
"release_id": source["release_id"],
"compose_project_name": source["compose_project_name"],
"application_images": {
key: {"reference": reference, "digest": reference.rsplit("@", 1)[1]}
for key, reference in application_images.items()
},
"infrastructure_images": {
key: {"reference": reference, "digest": reference.rsplit("@", 1)[1]}
for key, reference in infrastructure_images.items()
},
"offline_image_ids": {
"CONTRACT_API_IMAGE": f"sha256:{'4' * 64}",
"AI_PLATFORM_API_IMAGE": f"sha256:{'5' * 64}",
"AI_WORKER_IMAGE": f"sha256:{'6' * 64}",
"MINIO_IMAGE": object_storage["engine_resolved_image_id"],
"MINIO_MC_IMAGE": object_storage["client_resolved_image_id"],
},
}
validated, digest, _completed = snapshot_restore_planner._validate_backup_components(
{
"backup_manifest": record,
"consistency_mode": "quiesced",
"databases": list(snapshot_restore_planner.REQUIRED_DATABASES),
},
{
"backup_manifest": record,
"stores": list(snapshot_restore_planner.REQUIRED_STORAGE_TARGETS),
},
package_binding=package_binding,
environment_id="isolated-uat-01",
consistency_set_id="consistency-backup-20260716T120000Z",
require_formal_backup_manifest=True,
)
assert validated == adapted
assert digest == record["sha256"]
assert adapted["schema_version"] == backup_verify.PLANNER_BACKUP_MANIFEST_SCHEMA
assert adapted["source_backup_manifest_sha256"] == hashlib.sha256(
(backup_dir / "manifest.json").read_bytes()
).hexdigest()
assert adapted["application_runtime_images"] == application_images
assert adapted["object_storage_runtime_images"]["MINIO_IMAGE"] == {
"candidate_reference": object_storage["engine_image"],
"runtime_reference": object_storage["engine_runtime_image"],
"resolved_image_id": object_storage["engine_resolved_image_id"],
}
assert adapted["databases"]["ai"]["schema_revision"] == "20260925_0042"
summary = backup_verify.build_restore_check_summary(
checked_at=datetime(2026, 9, 27, 1, 0, tzinfo=timezone.utc),
source_manifest=source,
verified=verified,
restored=restored,
source_manifest_sha256=hashlib.sha256(
(backup_dir / "manifest.json").read_bytes()
).hexdigest(),
planner_manifest_record={
"schema_version": adapted["schema_version"],
"path": str(adapted_path.resolve()),
"sha256": record["sha256"],
"size_bytes": adapted_path.stat().st_size,
},
)
unsigned_summary = {
key: value for key, value in summary.items() if key != "summary_payload_sha256"
}
assert summary["schema_version"] == backup_verify.RESTORE_CHECK_SUMMARY_SCHEMA
assert summary["summary_payload_sha256"] == backup_verify.canonical_sha256(
unsigned_summary
)
assert summary["independent_restore_test_claimed"] is False
def test_recovery_adapter_rejects_restore_check_for_another_source_database(
tmp_path: Path,
) -> None:
backup_dir, source = _complete_backup_fixture(tmp_path)
source, _verified = backup_verify.validate_backup(backup_dir)
restored = _restore_check_fixture(source)
restored[0]["database"] = "another_contract_database"
with pytest.raises(RuntimeError, match="does not bind the source database"):
backup_verify.build_planner_backup_manifest(
backup_dir=backup_dir,
source_manifest=source,
restore_checks=restored,
environment_id="isolated-uat-01",
consistency_set_id="consistency-backup-20260716T120000Z",
)
def test_recovery_adapter_rejects_legacy_v1_instead_of_relabeling_it(
tmp_path: Path,
) -> None:
backup_dir, source = _complete_backup_fixture(tmp_path)
source["schema_version"] = "yuqei-production-backup.v1"
with pytest.raises(RuntimeError, match="Only artifacts.v2"):
backup_verify.build_planner_backup_manifest(
backup_dir=backup_dir,
source_manifest=source,
restore_checks=[],
environment_id="isolated-uat-01",
consistency_set_id="consistency-1",
)
@pytest.mark.parametrize(
("field", "value", "message"),
(
("engine", "minio", "engine or backup client"),
("client", "minio-mc", "engine or backup client"),
("engine_image", "rustfs/rustfs:1.0.0", "image identity is not immutable"),
("client_image", "rustfs/rc:0.1.36", "image identity is not immutable"),
("engine_version", "1.0.1", "version metadata is invalid"),
(
"engine_runtime_image",
f"registry.internal/other@sha256:{'b' * 64}",
"running image does not match",
),
),
)
def test_v2_backup_manifest_rejects_unbound_object_storage_identity(
tmp_path: Path,
field: str,
value: str,
message: str,
) -> None:
backup_dir, manifest = _complete_backup_fixture(tmp_path)
manifest["object_storage_mirror"][field] = value
(backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
with pytest.raises(RuntimeError, match=message):
backup_run.verify_manifest(backup_dir)
with pytest.raises(RuntimeError, match=message):
backup_verify.validate_backup(backup_dir)
def test_object_storage_runtime_identity_is_bound_to_running_and_client_images(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
outputs = iter(
(
b"a" * 64 + b"\n",
values["MINIO_IMAGE"].encode("utf-8") + b"\n",
b"sha256:" + b"b" * 64 + b"\n",
b"sha256:" + b"b" * 64 + b"\n",
b"sha256:" + b"c" * 64 + b"\n",
)
)
commands: list[list[str]] = []
def fake_run(command, **_kwargs):
commands.append(command)
return subprocess.CompletedProcess(command, 0, stdout=next(outputs), stderr=b"")
monkeypatch.setattr(backup_run, "_run", fake_run)
identity = backup_run._inspect_object_storage_runtime(
env_file=tmp_path / "production.env",
values=values,
)
assert identity == {
"engine_runtime_image": values["MINIO_IMAGE"],
"engine_resolved_image_id": "sha256:" + "b" * 64,
"client_runtime_image": values["MINIO_MC_IMAGE"],
"client_resolved_image_id": "sha256:" + "c" * 64,
}
assert commands[1][:3] == ["docker", "inspect", "--format"]
assert commands[2][:3] == ["docker", "inspect", "--format"]
assert commands[3][:3] == ["docker", "image", "inspect"]
assert commands[4][:3] == ["docker", "image", "inspect"]
def test_offline_object_storage_runtime_uses_receipt_bound_config_digests(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
engine_id = "sha256:" + "d" * 64
client_id = "sha256:" + "e" * 64
outputs = iter(
(
b"a" * 64 + b"\n",
engine_id.encode("ascii") + b"\n",
engine_id.encode("ascii") + b"\n",
engine_id.encode("ascii") + b"\n",
client_id.encode("ascii") + b"\n",
)
)
def fake_run(command, **_kwargs):
return subprocess.CompletedProcess(command, 0, stdout=next(outputs), stderr=b"")
monkeypatch.setattr(backup_run, "_run", fake_run)
identity = backup_run._inspect_object_storage_runtime(
env_file=tmp_path / "production.env",
values=values,
runtime_image_overrides={
"MINIO_IMAGE": engine_id,
"MINIO_MC_IMAGE": client_id,
},
)
assert identity["engine_runtime_image"] == engine_id
assert identity["engine_resolved_image_id"] == engine_id
assert identity["client_runtime_image"] == client_id
assert identity["client_resolved_image_id"] == client_id
def test_offline_config_digest_without_candidate_bound_receipt_fails_closed(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
outputs = iter(
(
b"a" * 64 + b"\n",
("sha256:" + "d" * 64).encode("ascii") + b"\n",
)
)
def fake_run(command, **_kwargs):
return subprocess.CompletedProcess(command, 0, stdout=next(outputs), stderr=b"")
monkeypatch.setattr(backup_run, "_run", fake_run)
with pytest.raises(RuntimeError, match="selected runtime identity"):
backup_run._inspect_object_storage_runtime(
env_file=tmp_path / "production.env",
values=values,
)
def test_release_receipt_binds_offline_ids_to_frozen_candidate_environment(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
state = tmp_path / "release-state"
receipts = state / "receipts"
receipts.mkdir(parents=True)
idempotency_key = "install-identity-001"
snapshot = (
state
/ "snapshots"
/ hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest()
/ "production.env"
)
snapshot.parent.mkdir(parents=True)
snapshot.write_text(
"\n".join(
f"{key}={values[key]}"
for key in ("RELEASE_ID", "COMPOSE_PROJECT_NAME", *backup_run.RUNTIME_IMAGE_KEYS)
)
+ "\n",
encoding="utf-8",
)
compose_files = []
for relative in (
"yuqei-infra/compose/docker-compose.yml",
"yuqei-infra/compose/docker-compose.production.yml",
"yuqei-infra/compose/docker-compose.production.offline.yml",
):
target = snapshot.parent.joinpath(*Path(relative).parts)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text("services: {}\n", encoding="utf-8")
compose_files.append(target)
def canonical_bytes(payload):
return json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
snapshot_manifest = {
"schema_version": "yuqei.release-input-snapshot.v2",
"plan_sha256": "1" * 64,
"tls_material_binding_sha256": "2" * 64,
"execution_env_sha256": hashlib.sha256(snapshot.read_bytes()).hexdigest(),
"files": [
{
"path": path.as_posix(),
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
"size_bytes": path.stat().st_size,
}
for path in (snapshot, *compose_files)
],
}
snapshot_manifest["snapshot_payload_sha256"] = hashlib.sha256(
canonical_bytes(snapshot_manifest)
).hexdigest()
snapshot_manifest_path = snapshot.parent / "snapshot-manifest.json"
snapshot_manifest_path.write_bytes(canonical_bytes(snapshot_manifest) + b"\n")
overrides = {
key: "sha256:" + f"{index:x}"[-1] * 64
for index, key in enumerate(backup_run.RUNTIME_IMAGE_KEYS, start=1)
}
overrides["AI_PLATFORM_RELEASE_IMAGE_DIGEST"] = "sha256:" + "f" * 64
receipt = {
"execution_mode": "builtin_fixed_subprocess_runner",
"status": "backend_verified",
"result": {"returncode": 0},
"target_release_id": values["RELEASE_ID"],
"compose_project_name": values["COMPOSE_PROJECT_NAME"],
"idempotency_key": idempotency_key,
"execution_env_sha256": hashlib.sha256(snapshot.read_bytes()).hexdigest(),
"snapshot_payload_sha256": snapshot_manifest["snapshot_payload_sha256"],
"environment_overrides": overrides,
"post_apply_check": {
"services": [{"service": "minio", "image_id": overrides["MINIO_IMAGE"]}],
},
}
receipt_path = receipts / "receipt.json"
receipt_path.write_text("{}\n", encoding="utf-8")
monkeypatch.setattr(
backup_run,
"_load_release_package_module",
lambda: SimpleNamespace(
_read_existing_receipt=lambda _path: receipt,
_read_json=lambda path, **_kwargs: json.loads(path.read_text(encoding="utf-8")),
canonical_json_bytes=canonical_bytes,
canonical_sha256=lambda payload: hashlib.sha256(canonical_bytes(payload)).hexdigest(),
),
)
actual_overrides, actual_compose = backup_run._release_runtime_overrides(
receipt_path=receipt_path,
values=values,
)
assert actual_overrides == overrides
assert actual_compose == tuple(path.resolve() for path in compose_files)
discovered = backup_run._discover_runtime_identity(
state_root=state,
values=values,
)
assert discovered is not None
assert discovered[0] == receipt_path
second_receipt = receipts / "second.json"
second_receipt.write_text("{}\n", encoding="utf-8")
with pytest.raises(RuntimeError, match="exactly one fully validated receipt"):
backup_run._discover_runtime_identity(
state_root=state,
values=values,
)
second_receipt.unlink()
values["MINIO_IMAGE"] = f"registry.internal/other@sha256:{'9' * 64}"
with pytest.raises(RuntimeError, match="candidate binding differs for MINIO_IMAGE"):
backup_run._release_runtime_overrides(
receipt_path=receipt_path,
values=values,
)
def test_offline_writer_restore_uses_frozen_three_layer_compose_and_never_pulls(
tmp_path: Path,
monkeypatch,
) -> None:
compose_files = (
tmp_path / "docker-compose.yml",
tmp_path / "docker-compose.production.yml",
tmp_path / "docker-compose.production.offline.yml",
)
calls: list[list[str]] = []
def fake_run(command, **_kwargs):
calls.append(command)
stdout = b"contract-api\n" if "ps" in command else b""
return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=b"")
monkeypatch.setattr(backup_run, "_run", fake_run)
attempts = backup_run._restore_stopped_writers(
env_file=tmp_path / "production.env",
project_name="yuqei-v2-production",
stopped_writers=("contract-api",),
runtime_env={"MINIO_IMAGE": "sha256:" + "1" * 64},
compose_files=compose_files,
)
assert attempts[0]["command_succeeded"] is True
assert attempts[0]["running_verified"] is True
for command in calls:
assert command.count("-f") == 3
assert str(compose_files[2]) in command
restore_command = calls[0]
assert restore_command[restore_command.index("--pull") + 1] == "never"
@pytest.mark.parametrize("field", ("backup_id", "created_at", "release_id", "images"))
def test_v2_backup_manifest_requires_complete_envelope(
tmp_path: Path,
field: str,
) -> None:
backup_dir, manifest = _complete_backup_fixture(tmp_path)
manifest.pop(field)
(backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
with pytest.raises(RuntimeError, match="manifest fields"):
backup_run.verify_manifest(backup_dir)
with pytest.raises(RuntimeError, match="manifest fields"):
backup_verify.validate_backup(backup_dir)
def test_create_backup_verifies_partial_before_publishing_completed_name(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
values.update(
{
"BACKUP_HOST_DIR": str(tmp_path),
"OBJECT_STORAGE_ENGINE": "rustfs",
"OBJECT_STORAGE_ENGINE_VERSION": "1.0.0",
"OBJECT_STORAGE_BACKUP_CLIENT": "rustfs-rc",
"OBJECT_STORAGE_BACKUP_CLIENT_VERSION": "0.1.36",
}
)
def fake_run(command, **_kwargs):
return subprocess.CompletedProcess(command, 0, stdout=b"", stderr=b"")
def fake_run_to_file(_command, target, **_kwargs):
target.write_bytes(b"database-dump")
def reject_partial(path):
assert path.name.startswith(".backup-") and path.name.endswith(".partial")
raise RuntimeError("synthetic manifest rejection")
monkeypatch.setattr(backup_run, "_run", fake_run)
monkeypatch.setattr(backup_run, "_run_to_file", fake_run_to_file)
monkeypatch.setattr(backup_run, "_verify_contract_archive", lambda _path: None)
monkeypatch.setattr(
backup_run,
"_inspect_object_storage_runtime",
lambda **_kwargs: {
"engine_runtime_image": values["MINIO_IMAGE"],
"engine_resolved_image_id": f"sha256:{'b' * 64}",
"client_runtime_image": values["MINIO_MC_IMAGE"],
"client_resolved_image_id": f"sha256:{'c' * 64}",
},
)
monkeypatch.setattr(
backup_run,
"_object_storage_mirror_metadata",
lambda *_args, **_kwargs: {},
)
monkeypatch.setattr(backup_run, "verify_manifest", reject_partial)
with pytest.raises(RuntimeError, match="synthetic manifest rejection"):
backup_run.create_backup(
env_file=tmp_path / "production.env",
values=values,
backup_root=tmp_path,
quiesce_writers=False,
)
assert not list(tmp_path.glob("backup-*"))
partials = list(tmp_path.glob(".backup-*.partial"))
assert len(partials) == 1
assert (partials[0] / "FAILED").is_file()
def test_writer_restore_failure_quarantines_completed_backup_and_attempts_all(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
values.update(
{
"BACKUP_HOST_DIR": str(tmp_path),
"OBJECT_STORAGE_ENGINE": "rustfs",
"OBJECT_STORAGE_ENGINE_VERSION": "1.0.0",
"OBJECT_STORAGE_BACKUP_CLIENT": "rustfs-rc",
"OBJECT_STORAGE_BACKUP_CLIENT_VERSION": "0.1.36",
}
)
restored: list[str] = []
compose_calls: list[list[str]] = []
ps_calls = 0
frozen_compose = (
tmp_path / "docker-compose.yml",
tmp_path / "docker-compose.production.yml",
tmp_path / "docker-compose.production.offline.yml",
)
def fake_run(command, **_kwargs):
nonlocal ps_calls
if command[:2] == ["docker", "compose"]:
compose_calls.append(command)
if "ps" in command and "--status" in command:
ps_calls += 1
stdout = ("\n".join(backup_run.WRITER_SERVICES) + "\n").encode("utf-8")
return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=b"")
if "up" in command and "--no-build" in command:
service = command[-1]
restored.append(service)
return subprocess.CompletedProcess(
command,
19 if service == backup_run.WRITER_SERVICES[0] else 0,
stdout=b"",
stderr=b"synthetic restore failure" if not restored[:-1] else b"",
)
return subprocess.CompletedProcess(command, 0, stdout=b"", stderr=b"")
def fake_run_to_file(_command, target, **_kwargs):
compose_calls.append(_command)
target.write_bytes(b"database-dump")
monkeypatch.setattr(
backup_run,
"_release_runtime_overrides",
lambda **_kwargs: (
{
**{
key: "sha256:" + "d" * 64
for key in backup_run.RUNTIME_IMAGE_KEYS
},
"AI_PLATFORM_RELEASE_IMAGE_DIGEST": "sha256:" + "e" * 64,
},
frozen_compose,
),
)
monkeypatch.setattr(backup_run, "_run", fake_run)
monkeypatch.setattr(backup_run, "_run_to_file", fake_run_to_file)
monkeypatch.setattr(backup_run, "_verify_contract_archive", lambda _path: None)
monkeypatch.setattr(
backup_run,
"_inspect_object_storage_runtime",
lambda **_kwargs: {
"engine_runtime_image": values["MINIO_IMAGE"],
"engine_resolved_image_id": f"sha256:{'b' * 64}",
"client_runtime_image": values["MINIO_MC_IMAGE"],
"client_resolved_image_id": f"sha256:{'c' * 64}",
},
)
monkeypatch.setattr(backup_run, "_object_storage_mirror_metadata", lambda *_a, **_k: {})
monkeypatch.setattr(backup_run, "verify_manifest", lambda _path: {})
with pytest.raises(RuntimeError, match="could not be restored and verified"):
backup_run.create_backup(
env_file=tmp_path / "production.env",
values=values,
backup_root=tmp_path,
quiesce_writers=True,
runtime_receipt=tmp_path / "receipt.json",
)
assert restored == list(backup_run.WRITER_SERVICES)
assert ps_calls == 2
assert not list(tmp_path.glob("backup-*"))
partials = list(tmp_path.glob(".backup-*.partial"))
assert len(partials) == 1
failure = json.loads((partials[0] / "failure.json").read_text(encoding="utf-8"))
backup_manifest = json.loads(
(partials[0] / "manifest.json").read_text(encoding="utf-8")
)
assert failure["status"] == "failed"
assert set(backup_manifest["images"].values()) == {f"sha256:{'d' * 64}"}
assert failure["writer_restore_attempts"][0]["command_succeeded"] is False
assert len(failure["writer_restore_attempts"]) == len(backup_run.WRITER_SERVICES)
assert compose_calls
for command in compose_calls:
assert command.count("-f") == 3
assert str(frozen_compose[2]) in command
def test_sigterm_enters_writer_restore_path_and_cannot_publish_success(
tmp_path: Path,
monkeypatch,
) -> None:
values = production_values()
values.update(
{
"BACKUP_HOST_DIR": str(tmp_path),
"OBJECT_STORAGE_ENGINE": "rustfs",
"OBJECT_STORAGE_ENGINE_VERSION": "1.0.0",
"OBJECT_STORAGE_BACKUP_CLIENT": "rustfs-rc",
"OBJECT_STORAGE_BACKUP_CLIENT_VERSION": "0.1.36",
}
)
installed_handlers: dict[int, object] = {}
restored: list[str] = []
def fake_signal(signum, handler):
if callable(handler):
installed_handlers[signum] = handler
def fake_run(command, **_kwargs):
if "ps" in command and "--status" in command:
stdout = ("\n".join(backup_run.WRITER_SERVICES) + "\n").encode("utf-8")
return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=b"")
if "up" in command and "--no-build" in command:
restored.append(command[-1])
return subprocess.CompletedProcess(command, 0, stdout=b"", stderr=b"")
def interrupting_dump(_command, _target, **_kwargs):
handler = installed_handlers[signal.SIGTERM]
assert callable(handler)
handler(signal.SIGTERM, None)
monkeypatch.setattr(backup_run.signal, "signal", fake_signal)
monkeypatch.setattr(backup_run, "_run", fake_run)
monkeypatch.setattr(backup_run, "_run_to_file", interrupting_dump)
with pytest.raises(backup_run.BackupInterrupted, match="SIGTERM"):
backup_run.create_backup(
env_file=tmp_path / "production.env",
values=values,
backup_root=tmp_path,
quiesce_writers=True,
)
assert restored == list(backup_run.WRITER_SERVICES)
assert not list(tmp_path.glob("backup-*"))
partials = list(tmp_path.glob(".backup-*.partial"))
assert len(partials) == 1
failure = json.loads((partials[0] / "failure.json").read_text(encoding="utf-8"))
assert failure["signal"] == "SIGTERM"
assert all(item["running_verified"] for item in failure["writer_restore_attempts"])
def test_backup_manifest_requires_all_databases_contract_files_and_minio(
tmp_path: Path,
) -> None:
backup_dir, manifest = _complete_backup_fixture(tmp_path)
manifest["artifacts"] = [
item
for item in manifest["artifacts"]
if item["path"] != "postgres-operations.dump"
]
(backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
with pytest.raises(RuntimeError, match="missing required artifacts"):
backup_verify.validate_backup(backup_dir)
legacy_dir, legacy_manifest = _complete_backup_fixture(tmp_path / "legacy")
legacy_manifest["schema_version"] = "yuqei-production-backup.v1"
legacy_manifest.pop("object_storage_mirror")
legacy_manifest["artifacts"] = [
item
for item in legacy_manifest["artifacts"]
if not str(item["path"]).startswith("minio/")
]
shutil.rmtree(legacy_dir / "minio")
(legacy_dir / "minio").mkdir()
(legacy_dir / "manifest.json").write_text(
json.dumps(legacy_manifest), encoding="utf-8"
)
with pytest.raises(RuntimeError, match="no MinIO mirror metadata or recorded MinIO content"):
backup_verify.validate_backup(legacy_dir)
def test_backup_manifest_supports_explicit_empty_minio_and_legacy_content(
tmp_path: Path,
) -> None:
empty_dir, empty_manifest = _complete_backup_fixture(tmp_path / "empty")
minio_path = empty_dir / "minio" / "tenant-a" / "object.bin"
minio_path.unlink()
empty_manifest["artifacts"] = [
item
for item in empty_manifest["artifacts"]
if not str(item["path"]).startswith("minio/")
]
empty_manifest["object_storage_mirror"].update({
"bucket": "ai-platform-uploads",
"path": "minio",
"object_count": 0,
"total_size_bytes": 0,
"content_index_sha256": backup_run._minio_content_index([]),
})
(empty_dir / "manifest.json").write_text(
json.dumps(empty_manifest), encoding="utf-8"
)
backup_verify.validate_backup(empty_dir)
legacy_dir, legacy_manifest = _complete_backup_fixture(tmp_path / "legacy-content")
legacy_manifest["schema_version"] = "yuqei-production-backup.v1"
legacy_manifest.pop("object_storage_mirror")
(legacy_dir / "manifest.json").write_text(
json.dumps(legacy_manifest), encoding="utf-8"
)
backup_verify.validate_backup(legacy_dir)
@pytest.mark.parametrize("member_type", (tarfile.SYMTYPE, tarfile.LNKTYPE, tarfile.CHRTYPE))
def test_contract_archive_rejects_links_and_special_entries(
tmp_path: Path,
member_type: bytes,
) -> None:
backup_dir, manifest = _complete_backup_fixture(tmp_path)
archive_path = backup_dir / "contract-files.tar.gz"
with tarfile.open(archive_path, mode="w:gz") as archive:
info = tarfile.TarInfo("unsafe-entry")
info.type = member_type
if member_type in {tarfile.SYMTYPE, tarfile.LNKTYPE}:
info.linkname = "outside"
archive.addfile(info)
for item in manifest["artifacts"]:
if item["path"] == "contract-files.tar.gz":
item["size_bytes"] = archive_path.stat().st_size
item["sha256"] = hashlib.sha256(archive_path.read_bytes()).hexdigest()
(backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
with pytest.raises(RuntimeError, match="link or special entry"):
backup_run.verify_manifest(backup_dir)
with pytest.raises(RuntimeError, match="link or special entry"):
backup_verify.validate_backup(backup_dir)
def test_provider_acceptance_separates_technical_success_from_legal_approval() -> None:
result = provider_acceptance.evaluate_case(
row={
"sample_id": "KBQA-1",
"expected_citation_required": "true",
"expected_source_ids": "LAW-CIVIL-001",
"production_ready": "false",
"status": "LEGAL_REVIEW",
"legal_reviewer": "待法务复核",
},
suite="legal_qa",
response={
"answer": "应结合合同和证据进一步判断。",
"citations": [{"document_id": "LAW-CIVIL-001"}],
},
audit={
"id": "run-1",
"status": "succeeded",
"provider_status": "succeeded",
"provider": "deepseek",
"model": "deepseek-chat",
"quality_labels": [],
},
latency_ms=100,
expected_boundary="不得直接判断个案效力。",
expected_severity=None,
)
assert result.technical_status == "passed"
assert result.legal_status == "needs_legal_review"
assert result.matched_source_ids == ["LAW-CIVIL-001"]
def test_provider_acceptance_rejects_fallback_audit() -> None:
result = provider_acceptance.evaluate_case(
row={"sample_id": "KBQA-2", "expected_citation_required": "false"},
suite="legal_qa",
response={"answer": "fallback answer", "citations": []},
audit={
"id": "run-2",
"status": "fallback",
"provider_status": "failed",
"provider_error": "provider unavailable",
"quality_labels": [{"code": "fallback_used"}],
},
latency_ms=50,
expected_boundary="",
expected_severity=None,
)
assert result.technical_status == "failed"
assert "provider_not_succeeded" in result.problems
assert "fallback_or_provider_failure_label" in result.problems
def test_provider_acceptance_requires_explicit_stable_source_ids() -> None:
observed = provider_acceptance._response_source_ids(
[
{"document_id": "doc-LAW-CIVIL-001-random"},
{"document_id": "doc-random", "reference": "LAW-CONTRACT-001 article 1"},
{"document_id": "doc-random-2", "source_ids": ["LAW-CIVIL-001", "not-stable"]},
]
)
assert observed == {"LAW-CIVIL-001", "LAW-CONTRACT-001"}
def test_provider_acceptance_loads_phase13_knowledge_cards_idempotently() -> None:
documents: list[dict[str, object]] = []
class Client:
def get(self, path: str):
assert path.startswith("/api/v1/knowledge-documents/search?")
return documents, 1
def post(self, path: str, payload: dict[str, object]):
assert path == "/api/v1/knowledge-documents"
documents.append({"id": f"doc-{len(documents)}", **payload})
return documents[-1], 1
client = Client()
assert provider_acceptance.load_phase13_knowledge_assets(client) == 10
assert provider_acceptance.load_phase13_knowledge_assets(client) == 0
assert len(documents) == 10
assert all(str(item["reference"]).startswith("LAW-") for item in documents)
assert all("不代表法务批准" in str(item["content"]) for item in documents)
def test_provider_acceptance_requires_complete_named_legal_approval(tmp_path: Path) -> None:
qa_rows = provider_acceptance.load_rows(provider_acceptance.QA_INVENTORY)
review_rows = provider_acceptance.load_rows(provider_acceptance.REVIEW_INVENTORY)
card_rows = provider_acceptance.load_rows(provider_acceptance.KNOWLEDGE_CARD_INVENTORY)
approval_file = tmp_path / "legal-approval.json"
approval_file.write_text(
json.dumps(
{
"status": "approved",
"reviewer": "Named legal reviewer",
"approved_at": "2026-08-10T08:00:00+08:00",
"evidence_ref": "https://legal.example.test/approvals/provider-phase13",
"evidence_sha256": "a" * 64,
"approved_qa_sample_ids": [row["sample_id"] for row in qa_rows],
"approved_review_sample_ids": [row["sample_id"] for row in review_rows],
"approved_knowledge_card_ids": [row["knowledge_card_id"] for row in card_rows],
},
ensure_ascii=False,
),
encoding="utf-8",
)
approval = provider_acceptance.load_legal_approval(
approval_file,
qa_sample_ids=[row["sample_id"] for row in qa_rows],
review_sample_ids=[row["sample_id"] for row in review_rows],
)
assert approval["legal_approval_status"] == "approved"
assert approval["legal_reviewer"] == "Named legal reviewer"
assert len(approval["approved_knowledge_card_ids"]) == 10
tampered = json.loads(approval_file.read_text(encoding="utf-8"))
tampered["approved_review_sample_ids"] = tampered["approved_review_sample_ids"][:-1]
approval_file.write_text(json.dumps(tampered), encoding="utf-8")
with pytest.raises(ValueError, match="all 8 contract review samples"):
provider_acceptance.load_legal_approval(
approval_file,
qa_sample_ids=[row["sample_id"] for row in qa_rows],
review_sample_ids=[row["sample_id"] for row in review_rows],
)
def test_provider_acceptance_registers_only_complete_10_plus_8_manifest() -> None:
qa_rows = provider_acceptance.load_rows(provider_acceptance.QA_INVENTORY)
review_rows = provider_acceptance.load_rows(provider_acceptance.REVIEW_INVENTORY)
card_rows = provider_acceptance.load_rows(provider_acceptance.KNOWLEDGE_CARD_INVENTORY)
cases = []
for suite, rows in (("legal_qa", qa_rows), ("contract_review", review_rows)):
for row in rows:
cases.append(
provider_acceptance.CaseResult(
sample_id=row["sample_id"],
suite=suite,
technical_status="passed",
legal_status="needs_legal_review",
latency_ms=1,
provider="openai-compatible",
model="runtime-model",
audit_run_id=f"run-{row['sample_id']}",
answer_non_empty=True,
citation_count=2,
expected_source_ids=[],
matched_source_ids=[],
expected_boundary="",
expected_severity=None,
observed_severity=None,
production_ready=False,
asset_status="LEGAL_REVIEW",
legal_reviewer="",
problems=[],
)
)
class Client:
payload: dict[str, object] | None = None
tenant_id = "default"
def post(
self,
path: str,
payload: dict[str, object],
*,
extra_headers: dict[str, str] | None = None,
):
assert path == "/api/v1/quality/provider-acceptance-runs"
assert extra_headers
assert extra_headers["X-AI-Platform-Legal-Attestation"] == payload["legal_approval_attestation"]
self.payload = payload
return {"id": "provider-acceptance-complete"}, 1
client = Client()
run_id = provider_acceptance.register_acceptance_manifest(
client,
provider_config={
"id": "provider-config-1",
"provider": "openai-compatible",
"model": "runtime-model",
},
qa_rows=qa_rows,
review_rows=review_rows,
cases=cases,
approval={
"legal_approval_status": "approved",
"legal_reviewer": "Named legal reviewer",
"legal_approved_at": "2026-08-10T08:00:00+08:00",
"legal_approval_evidence_ref": "https://legal.example.test/approval",
"legal_approval_evidence_sha256": "b" * 64,
"approved_knowledge_card_ids": [row["knowledge_card_id"] for row in card_rows],
},
legal_attestation_secret="independent-legal-attestation-secret-for-tests", # pragma: allowlist secret
)
assert run_id == "provider-acceptance-complete"
assert client.payload is not None
assert len(client.payload["qa_sample_ids"]) == 10
assert len(client.payload["review_sample_ids"]) == 8
assert len(client.payload["knowledge_card_ids"]) == 10
assert len(client.payload["audit_bindings"]) == 18
with pytest.raises(ValueError, match="exactly 10 QA and 8 review"):
provider_acceptance.register_acceptance_manifest(
client,
provider_config={"id": "p", "provider": "p", "model": "m"},
qa_rows=qa_rows[:-1],
review_rows=review_rows,
cases=cases[:-1],
approval={},
legal_attestation_secret="independent-legal-attestation-secret-for-tests", # pragma: allowlist secret
)
def _valid_provider_receipt_payload(
*,
run_id: str = "provider-acceptance-complete",
key_id: str = "staging-receipt-key-1",
target_release_id: str = "v2.0.0-rc.1",
target_image_digest: str = "sha256:" + "2" * 64,
) -> dict[str, object]:
qa_rows = provider_acceptance.load_rows(provider_acceptance.QA_INVENTORY)
review_rows = provider_acceptance.load_rows(provider_acceptance.REVIEW_INVENTORY)
card_rows = provider_acceptance.load_rows(provider_acceptance.KNOWLEDGE_CARD_INVENTORY)
audit_bindings = [
{
"suite": suite,
"sample_id": row["sample_id"],
"audit_run_id": f"audit-{index}",
"audit_snapshot_sha256": f"{index:064x}",
}
for index, (suite, row) in enumerate(
[*(('legal_qa', row) for row in qa_rows), *(('contract_review', row) for row in review_rows)],
start=1,
)
]
return {
"receipt_version": "provider-acceptance-receipt-v2",
"algorithm": "Ed25519",
"key_id": key_id,
"issuer": "shared_staging",
"audience": "yuqei-ai-platform-production",
"source_environment": "shared_staging",
"target_environment": "production",
"target_release_id": target_release_id,
"target_image_digest": target_image_digest,
"tenant_id": "default",
"acceptance_run": {
"source_environment": "shared_staging",
"acceptance_policy_version": provider_acceptance.ACCEPTANCE_POLICY_VERSION,
"provider_config_id": "provider-config-1",
"provider": "openai-compatible",
"model": "runtime-model",
"qa_sample_ids": [row["sample_id"] for row in qa_rows],
"review_sample_ids": [row["sample_id"] for row in review_rows],
"knowledge_card_ids": [row["knowledge_card_id"] for row in card_rows],
"audit_bindings": audit_bindings,
"suite_manifest_sha256": provider_acceptance._manifest_hash(
[provider_acceptance.QA_INVENTORY, provider_acceptance.REVIEW_INVENTORY]
),
"knowledge_manifest_sha256": provider_acceptance._manifest_hash(
[provider_acceptance.KNOWLEDGE_CARD_INVENTORY]
),
"legal_approval_status": "approved",
"legal_reviewer": "Named legal reviewer",
"legal_approved_at": "2026-08-10T08:00:00+08:00",
"legal_approval_evidence_ref": "https://legal.example.test/approval",
"legal_approval_evidence_sha256": "a" * 64,
"legal_approval_attestation": "b" * 64,
"id": run_id,
"tenant_id": "default",
"acceptance_manifest_sha256": "c" * 64,
"created_at": "2026-08-10T00:01:00+00:00",
},
"provider_runtime": {
"provider": "openai-compatible",
"model": "runtime-model",
"base_url": "https://provider.example.test/v1",
"wire_api": "chat_completions",
"temperature": 0.1,
"max_tokens": 4096,
"prompt_templates_sha256": "d" * 64,
"fingerprint_sha256": "e" * 64,
},
"quality_policy_sha256": "f" * 64,
"issued_at": "2026-08-10T00:02:00+00:00",
"not_before": "2026-08-10T00:01:30+00:00",
"expires_at": "2026-08-11T00:02:00+00:00",
}
def test_provider_acceptance_requires_all_receipt_options_and_legal_approval(tmp_path: Path) -> None:
values = {
"receipt_private_key_file": tmp_path / "receipt-key.pem",
"receipt_key_id": "staging-receipt-key-1",
"target_release_id": "v2.0.0-rc.1",
"target_image_digest": "sha256:" + "2" * 64,
"receipt_output": tmp_path / "receipt.json",
"legal_approval_file": tmp_path / "legal-approval.json",
}
assert provider_acceptance.validate_receipt_options(SimpleNamespace(**values)) is True
assert provider_acceptance.validate_receipt_options(
SimpleNamespace(**{key: None for key in values})
) is False
for missing in (
"receipt_private_key_file",
"receipt_key_id",
"target_release_id",
"target_image_digest",
"receipt_output",
):
with pytest.raises(ValueError, match="all receipt options"):
provider_acceptance.validate_receipt_options(
SimpleNamespace(**{**values, missing: None})
)
with pytest.raises(ValueError, match="named legal approval"):
provider_acceptance.validate_receipt_options(
SimpleNamespace(**{**values, "legal_approval_file": None})
)
with pytest.raises(ValueError, match="separate JSON"):
provider_acceptance.validate_receipt_output_path(
values["receipt_private_key_file"],
[values["receipt_private_key_file"]],
)
with pytest.raises(ValueError, match="separate JSON"):
provider_acceptance.validate_receipt_output_path(tmp_path / "receipt.txt", [])
def test_provider_acceptance_fetches_signs_and_writes_separate_receipt(tmp_path: Path) -> None:
payload = _valid_provider_receipt_payload()
requested_paths: list[str] = []
class Client:
tenant_id = "default"
def get(self, path: str):
requested_paths.append(path)
return payload, 1
private_key = Ed25519PrivateKey.generate()
key_file = tmp_path / "receipt-key.pem"
key_file.write_bytes(
private_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
)
receipt_path = tmp_path / "provider-acceptance-receipt.json"
metadata = provider_acceptance.create_signed_acceptance_receipt(
Client(),
acceptance_run_id="provider-acceptance-complete",
key_id="staging-receipt-key-1",
target_release_id="v2.0.0-rc.1",
target_image_digest="sha256:" + "2" * 64,
private_key_file=key_file,
output_path=receipt_path,
)
assert requested_paths == [
"/api/v1/quality/provider-acceptance-runs/provider-acceptance-complete/receipt-payload"
"?key_id=staging-receipt-key-1&target_release_id=v2.0.0-rc.1"
f"&target_image_digest=sha256%3A{'2' * 64}"
]
receipt = SignedProviderAcceptanceReceipt.model_validate_json(receipt_path.read_bytes())
assert receipt.payload == ProviderAcceptanceReceiptPayload.model_validate(payload)
signature = base64.urlsafe_b64decode(receipt.signature + "==")
private_key.public_key().verify(
signature,
RECEIPT_DOMAIN + canonical_receipt_payload(receipt.payload),
)
assert metadata == {
"receipt_id": receipt.receipt_id,
"sha256": hashlib.sha256(receipt_path.read_bytes()).hexdigest(),
}
provider_acceptance.write_evidence(
tmp_path,
provider_acceptance.build_evidence_summary(
status="passed",
provider="openai-compatible",
model="runtime-model",
cases=[],
),
)
assert receipt_path.exists()
assert receipt_path.name not in {
"ai-provider-acceptance-summary.json",
"ai-provider-acceptance-summary.md",
}
def test_provider_acceptance_rejects_unsafe_receipt_keys_without_leaking_material(tmp_path: Path) -> None:
payload = _valid_provider_receipt_payload()
class Client:
tenant_id = "default"
def get(self, _path: str):
return payload, 1
rsa_key = rsa.generate_private_key(public_exponent=65537, key_size=2048).private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
key_inputs = {
"invalid.pem": b"private-secret-marker-not-a-key",
"rsa.pem": rsa_key,
"oversized.pem": b"private-secret-marker" * 4096,
}
key_paths: list[Path] = [tmp_path / "missing-private-secret-marker.pem"]
for name, content in key_inputs.items():
path = tmp_path / name
path.write_bytes(content)
key_paths.append(path)
for index, key_path in enumerate(key_paths):
with pytest.raises(ValueError, match="unavailable or invalid") as raised:
provider_acceptance.create_signed_acceptance_receipt(
Client(),
acceptance_run_id="provider-acceptance-complete",
key_id="staging-receipt-key-1",
target_release_id="v2.0.0-rc.1",
target_image_digest="sha256:" + "2" * 64,
private_key_file=key_path,
output_path=tmp_path / f"receipt-{index}.json",
)
assert "private-secret-marker" not in str(raised.value)
assert not (tmp_path / f"receipt-{index}.json").exists()
def test_provider_acceptance_never_fetches_or_writes_receipt_without_acceptance_id(tmp_path: Path) -> None:
class Client:
tenant_id = "default"
def get(self, _path: str):
raise AssertionError("receipt payload must not be fetched")
receipt_path = tmp_path / "receipt.json"
with pytest.raises(ValueError, match="acceptance_run_id"):
provider_acceptance.create_signed_acceptance_receipt(
Client(),
acceptance_run_id="",
key_id="staging-receipt-key-1",
target_release_id="v2.0.0-rc.1",
target_image_digest="sha256:" + "2" * 64,
private_key_file=tmp_path / "missing.pem",
output_path=receipt_path,
)
assert not receipt_path.exists()
def test_provider_acceptance_excludes_deterministic_product_suites() -> None:
assert not hasattr(provider_acceptance, "GENERATION_INVENTORY")
assert not hasattr(provider_acceptance, "run_generation_case")
def test_contract_product_golden_runs_real_deterministic_product_paths() -> None:
results = contract_product_golden.run_product_golden()
summary = contract_product_golden.build_summary(results)
assert len(results) == 31
assert sum(item.suite == "contract_generation" for item in results) == 23
assert sum(item.suite == "clause_recommendation" for item in results) == 8
assert all(item.technical_status == "passed" for item in results)
assert all(item.legal_status == "needs_legal_review" for item in results)
assert summary["status"] == "needs_legal_review"
assert summary["counts"]["technical_passed"] == 31
assert summary["counts"]["legal_approved"] == 0
assert summary["counts"]["preserved_required_placeholders"] == 0
assert summary["problems"] == []
def test_contract_product_golden_rejects_inconsistent_expected_boundary() -> None:
row = contract_product_golden.load_rows()[0]
tampered = dict(
row,
expected_source_citations=row["expected_source_citations"] + ";LAW-NOT-IN-BOUNDARY-001",
)
result = contract_product_golden.evaluate_product_case(tampered)
assert result.technical_status == "failed"
assert "expected_boundary_missing_identifier" in result.problems
def test_contract_product_golden_does_not_approve_names_without_evidence() -> None:
row = contract_product_golden.load_rows()[0]
incomplete_approval = dict(
row,
status="APPROVED",
production_ready="true",
legal_reviewer="Named legal reviewer",
business_reviewer="Named business reviewer",
)
result = contract_product_golden.evaluate_product_case(incomplete_approval)
assert result.technical_status == "passed"
assert result.legal_status == "needs_legal_review"
def test_contract_product_golden_evidence_does_not_claim_legal_approval(
tmp_path: Path,
) -> None:
summary = contract_product_golden.build_summary(contract_product_golden.run_product_golden())
contract_product_golden.write_evidence(tmp_path, summary)
assert summary["status"] == "needs_legal_review"
assert "acceptance_run_id" not in summary
assert set(path.name for path in tmp_path.iterdir()) == {
"contract-product-golden-summary.json",
"contract-product-golden-summary.md",
}
def test_provider_acceptance_evidence_excludes_responses_and_identifiers(tmp_path: Path) -> None:
result = provider_acceptance.evaluate_case(
row={
"sample_id": "sensitive-sample-id",
"expected_citation_required": "true",
"expected_source_ids": "LAW-SECRET-001",
"production_ready": "false",
},
suite="legal_qa",
response={
"answer": "sensitive-provider-answer",
"citations": [{"document_id": "LAW-SECRET-001"}],
},
audit={
"id": "sensitive-audit-run-id",
"status": "succeeded",
"provider_status": "succeeded",
"provider": "deepseek",
"model": "deepseek-chat",
"quality_labels": [],
},
latency_ms=100,
expected_boundary="sensitive-legal-boundary",
expected_severity=None,
)
summary = provider_acceptance.build_evidence_summary(
status="needs_legal_review",
provider="deepseek",
model="deepseek-chat",
cases=[result],
)
provider_acceptance.write_evidence(tmp_path, summary)
assert set(summary) == {"provider", "model", "status", "counts", "latency_ms", "problems"}
serialized = json.dumps(summary, ensure_ascii=False)
for sensitive_value in (
"sensitive-sample-id",
"sensitive-provider-answer",
"LAW-SECRET-001",
"sensitive-audit-run-id",
"sensitive-legal-boundary",
):
assert sensitive_value not in serialized
assert not (tmp_path / "ai-provider-acceptance-responses.json").exists()
assert set(path.name for path in tmp_path.iterdir()) == {
"ai-provider-acceptance-summary.json",
"ai-provider-acceptance-summary.md",
}
def test_provider_acceptance_validates_safe_runtime_parameters() -> None:
valid_key = "service-key-that-is-long-enough-for-testing" # pragma: allowlist secret
with pytest.raises(ValueError, match="query/fragment"):
provider_acceptance.ApiClient(
"https://api.example.test/ai?key=unsafe",
service_key=valid_key,
tenant_id="default",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="service key"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=f"{valid_key}\r\nInjected: true",
tenant_id="default",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="Tenant ID"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=valid_key,
tenant_id="default\r\nInjected: true",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="Timeout"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=valid_key,
tenant_id="default",
timeout_seconds=0,
)
def test_provider_acceptance_allows_explicit_isolated_memory_schema_warning() -> None:
class Client:
def get(self, path: str):
if path == "/api/v1/runtime/schema-check":
return {
"status": "warning",
"checks": [
{
"name": "repository_backend",
"status": "warning",
"details": {"repository_backend": "memory"},
},
{"name": "anti_abuse_runtime", "status": "warning", "details": {}},
],
}, 1
if path == "/api/v1/model-configs":
return [
{
"provider": "isolated-provider",
"model": "isolated-model",
"enabled": True,
"is_default": True,
"base_url": "https://provider.example.test",
"api_key_set": True,
}
], 1
raise AssertionError(path)
schema, provider_config = provider_acceptance.validate_runtime(Client())
assert schema["status"] == "warning"
assert provider_config["provider"] == "isolated-provider"
def test_provider_acceptance_allows_local_codex_development_evidence_only() -> None:
class Client:
def get(self, path: str):
if path == "/api/v1/runtime/schema-check":
return {
"status": "warning",
"checks": [
{
"name": "repository_backend",
"status": "warning",
"details": {"repository_backend": "memory"},
},
{"name": "anti_abuse_runtime", "status": "warning", "details": {}},
{
"name": "provider_runtime",
"status": "ok",
"details": {
"environment": "local",
"provider_call_mode": "codex_cli",
},
},
],
}, 1
if path == "/api/v1/model-configs":
return [
{
"id": "legacy-development-provider",
"provider": "legacy-development-config",
"model": "gpt-pro-development",
"enabled": True,
"is_default": True,
}
], 1
raise AssertionError(path)
schema, provider_config = provider_acceptance.validate_runtime(
Client(),
development_codex_cli=True,
)
assert schema["status"] == "warning"
assert provider_config["provider"] == "codex-cli-development"
assert provider_config["model"] == "gpt-5.6-sol"
assert provider_config["development_runtime_override"] is True
def test_provider_acceptance_never_promotes_development_codex_evidence() -> None:
with pytest.raises(RuntimeError, match="cannot be treated as shared-staging"):
provider_acceptance.validate_runtime(
object(),
require_shared_staging=True,
development_codex_cli=True,
)
summary = provider_acceptance.mark_development_evidence(
provider_acceptance.build_evidence_summary(
status="needs_legal_review",
provider=None,
model=None,
cases=[],
)
)
assert summary["provider"] == "codex-cli-development"
assert summary["model"] == "gpt-5.6-sol"
assert summary["evidence_scope"] == "local_development_codex_cli"
assert summary["production_eligible"] is False
assert summary["provider_acceptance_registration_allowed"] is False
assert summary["controlled_pilot_gate_effect"] == "none"
def test_provider_acceptance_reports_operational_quality_and_latency_separately() -> None:
result = provider_acceptance.evaluate_case(
row={
"sample_id": "QA-DEVELOPMENT-SLOW-001",
"expected_citation_required": "false",
"production_ready": "false",
"status": "LEGAL_REVIEW",
"legal_reviewer": "待法务复核",
},
suite="legal_qa",
response={"answer": "有内容的技术回答", "citations": []},
audit={
"id": "run-development-slow",
"status": "succeeded",
"provider_status": "succeeded",
"provider": "codex-cli-development",
"model": "gpt-5.6-sol",
"quality_status": "failed",
"quality_score": 60,
"quality_labels": [{"code": "latency_critical"}],
},
latency_ms=128_000,
expected_boundary="",
expected_severity=None,
)
summary = provider_acceptance.build_evidence_summary(
status="needs_operational_review",
provider=result.provider,
model=result.model,
cases=[result],
)
assert result.technical_status == "passed"
assert result.quality_status == "failed"
assert result.quality_score == 60
assert summary["counts"]["technical_passed"] == 1
assert summary["counts"]["quality_failed"] == 1
assert summary["latency_ms"] == {
"min": 128_000,
"median": 128_000,
"p95": 128_000,
"max": 128_000,
}
def test_provider_acceptance_encodes_chinese_request_as_utf8(monkeypatch) -> None:
captured: dict[str, object] = {}
class Response:
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self, _limit: int) -> bytes:
return b'{"status":"ok"}'
def fake_urlopen(request, *, timeout):
captured["body"] = request.data
captured["timeout"] = timeout
return Response()
monkeypatch.setattr(provider_acceptance, "urlopen", fake_urlopen)
client = provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key="service-key-that-is-long-enough-for-testing", # pragma: allowlist secret
tenant_id="default",
timeout_seconds=30,
)
payload, _ = client.post("/api/v1/legal/qa", {"question": "电子邮件能否作为合同证据？"})
assert payload == {"status": "ok"}
assert json.loads(captured["body"].decode("utf-8"))["question"] == "电子邮件能否作为合同证据？"
assert captured["timeout"] == 30
def test_provider_acceptance_runs_all_phase13_cases_against_isolated_http_service() -> None:
qa_rows = provider_acceptance.load_rows(provider_acceptance.QA_INVENTORY)
review_rows = provider_acceptance.load_rows(provider_acceptance.REVIEW_INVENTORY)
qa_by_question = {row["question"]: row for row in qa_rows}
review_by_id = {row["sample_id"]: row for row in review_rows}
audits: dict[str, dict[str, object]] = {}
class Handler(BaseHTTPRequestHandler):
def send_json(self, payload: object) -> None:
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/api/v1/runtime/schema-check":
self.send_json(
{
"status": "ok",
"checks": [{"name": "provider_runtime", "status": "ok"}],
}
)
return
if path == "/api/v1/model-configs":
self.send_json(
[
{
"id": "provider-isolated",
"provider": "isolated-provider",
"model": "isolated-model",
"enabled": True,
"is_default": True,
"base_url": "https://provider.invalid/v1",
"api_key_set": True,
}
]
)
return
if path == "/api/v1/audit/ai-runs":
self.send_json(list(reversed(list(audits.values()))))
return
run_id = path.removeprefix("/api/v1/audit/ai-runs/")
if run_id in audits:
self.send_json(audits[run_id])
return
self.send_error(404)
def do_POST(self) -> None: # noqa: N802
content_length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(content_length).decode("utf-8"))
run_id = f"isolated-run-{len(audits) + 1}"
if self.path == "/api/v1/legal/qa":
row = qa_by_question[payload["question"]]
citations = [
{"document_id": f"doc-{index}", "source_ids": [source_id]}
for index, source_id in enumerate(provider_acceptance.split_ids(row["expected_source_ids"]))
]
operation = "legal_qa"
response = {"answer": "基于检索依据给出边界明确的答复。", "citations": citations}
elif self.path == "/api/v1/legal/contract-review":
row = review_by_id[payload["contract_title"]]
citations = [
{"document_id": f"doc-{index}", "source_ids": [source_id]}
for index, source_id in enumerate(provider_acceptance.split_ids(row["expected_source_citations"]))
]
cited_source_ids = provider_acceptance.split_ids(row["expected_source_citations"])
operation = "contract_review"
response = {
"summary": "已按约定审查重点完成审查。",
"citations": citations,
"risk_items": [
{
"severity": row["expected_severity"],
"citation_source_ids": cited_source_ids,
}
],
"audit_run_id": run_id,
}
else:
self.send_error(404)
return
audits[run_id] = {
"id": run_id,
"operation": operation,
"status": "succeeded",
"provider_status": "succeeded",
"provider": "isolated-provider",
"model": "isolated-model",
"quality_labels": [],
"created_at": datetime.now(timezone.utc).isoformat(),
}
self.send_json(response)
def log_message(self, _format: str, *_args: object) -> None:
return
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
client = provider_acceptance.ApiClient(
f"http://127.0.0.1:{server.server_port}",
service_key="isolated-non-secret-service-key-for-tests", # pragma: allowlist secret
tenant_id="default",
timeout_seconds=10,
)
schema, provider_config = provider_acceptance.validate_runtime(client)
seen_run_ids: set[str] = set()
results = [
provider_acceptance.run_qa_case(client, row, seen_run_ids)[0]
for row in qa_rows
]
results.extend(
provider_acceptance.run_review_case(client, row, seen_run_ids)[0]
for row in review_rows
)
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
assert len(qa_rows) == 10
assert len(review_rows) == 8
assert len(results) == 18
assert schema["status"] == "ok"
assert provider_config["provider"] == "isolated-provider"
assert all(result.technical_status == "passed" for result in results)
assert all(result.legal_status == "needs_legal_review" for result in results)
def test_production_tool_bandit_suppressions_are_exactly_allowlisted() -> None:
expected = {
"yuqei-infra/backup/run-production-backup.py": ["B404", "B603"],
"yuqei-infra/backup/verify-production-backup.py": ["B404", "B603", "B603"],
"yuqei-infra/observability/check-production-health.py": ["B310", "B310"],
"yuqei-infra/scripts/run-ai-provider-acceptance.py": ["B310"],
}
observed: dict[str, list[str]] = {}
for relative_path in expected:
lines = (REPO_ROOT / relative_path).read_text(encoding="utf-8").splitlines()
observed[relative_path] = [
line.split("# nosec", 1)[1].strip()
for line in lines
if "# nosec" in line
]
assert observed == expected
Reference in New Issue View Git Blame Copy Permalink