1658 lines
66 KiB
Python
1658 lines
66 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
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
|
|
|
|
|
|
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")
|
|
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",
|
|
"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/minio@sha256:{'7' * 64}",
|
|
"MINIO_MC_IMAGE": f"registry.internal/minio-mc@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)
|
|
return values
|
|
|
|
|
|
def test_minio_backup_invokes_pinned_mc_directly_without_a_shell(tmp_path: Path) -> None:
|
|
partial_dir = tmp_path / ".backup-20260811T120000Z.partial"
|
|
credentials = tmp_path / "minio.env"
|
|
values = {
|
|
"COMPOSE_PROJECT_NAME": "yuqei-v2-production",
|
|
"AI_PLATFORM_OBJECT_STORAGE_BUCKET": "ai-platform-uploads",
|
|
"MINIO_MC_IMAGE": f"minio/mc@sha256:{'a' * 64}",
|
|
}
|
|
|
|
command = backup_run.minio_mirror_command(
|
|
values=values,
|
|
partial_dir=partial_dir,
|
|
credential_env_file=credentials,
|
|
)
|
|
|
|
assert command[:2] == ["docker", "run"]
|
|
assert "--entrypoint" not in command
|
|
assert "/bin/sh" not in command
|
|
assert command[-4:] == [
|
|
"mirror",
|
|
"--overwrite",
|
|
"source/ai-platform-uploads",
|
|
"/backup/minio",
|
|
]
|
|
assert command[command.index("--env-file") + 1] == str(credentials)
|
|
assert all("secret" not in token for token in command)
|
|
|
|
|
|
@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_table_count", "information_schema.tables"),
|
|
],
|
|
)
|
|
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.v1_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 test_backup_manifest_and_contract_archive_are_verified(tmp_path: Path) -> None:
|
|
backup_dir = tmp_path / "backup-20260716T120000Z"
|
|
backup_dir.mkdir()
|
|
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))
|
|
dump_path = backup_dir / "postgres-contract.dump"
|
|
dump_path.write_bytes(b"postgres-dump")
|
|
artifacts = []
|
|
for path in (archive_path, dump_path):
|
|
artifacts.append(
|
|
{
|
|
"path": path.name,
|
|
"size_bytes": path.stat().st_size,
|
|
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
|
|
}
|
|
)
|
|
(backup_dir / "manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "yuqei-production-backup.v1",
|
|
"backup_id": backup_dir.name,
|
|
"created_at": "2026-07-16T12:00:00+00:00",
|
|
"artifacts": artifacts,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
manifest, verified = backup_verify.validate_backup(backup_dir)
|
|
|
|
assert manifest["backup_id"] == backup_dir.name
|
|
assert len(verified) == 2
|
|
|
|
|
|
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
|