16e791e044
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본 폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을 git 이력으로 고정하는 것이 목적이다. - contracts/routes/services: measurement, outcome_trajectory, rupture_repair, deliberate_practice, calibration_transfer, supervision_research, multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트 - infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행) - apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가 - docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG - scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트 engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
566 lines
23 KiB
Python
566 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Any
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT_PATH = Path(__file__).with_name("probe-nas-preview-g8-rollback.py")
|
|
SPEC = importlib.util.spec_from_file_location("g8_rollback_probe", SCRIPT_PATH)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = MODULE
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
EXECUTOR_PATH = Path(__file__).with_name("serve-nas-preview-rollback-executor.py")
|
|
EXECUTOR_SPEC = importlib.util.spec_from_file_location(
|
|
"nas_preview_rollback_executor_contract", EXECUTOR_PATH
|
|
)
|
|
assert EXECUTOR_SPEC is not None and EXECUTOR_SPEC.loader is not None
|
|
EXECUTOR_MODULE = importlib.util.module_from_spec(EXECUTOR_SPEC)
|
|
sys.modules[EXECUTOR_SPEC.name] = EXECUTOR_MODULE
|
|
EXECUTOR_SPEC.loader.exec_module(EXECUTOR_MODULE)
|
|
|
|
INTERNAL_TOKEN = "internal-probe-token-" + "x" * 32
|
|
ADMIN_EMAIL = "g8-control-plane-admin@hs.ac.kr"
|
|
COOKIE_VALUE = "secret-cookie-value-must-never-be-printed"
|
|
|
|
|
|
def sample_plan(suffix: str = "proof-20260807") -> dict[str, Any]:
|
|
return MODULE.build_plan(
|
|
suffix=suffix,
|
|
artifact_id=f"nas-preview-rollback-{suffix}",
|
|
artifact_sha256="a" * 64,
|
|
artifact_provenance_uri=(
|
|
f"repo://vignette/releases/nas-preview-rollback-{suffix}"
|
|
),
|
|
evidence_refs=[
|
|
f"audit://vignette-nas-preview/g8-rollback/{suffix}/authorization",
|
|
f"repo://vignette/releases/{suffix}/manifest",
|
|
],
|
|
compose_sha256="b" * 64,
|
|
env_sha256="c" * 64,
|
|
resolved_config_sha256="d" * 64,
|
|
api_image="sha256:" + "e" * 64,
|
|
web_image="sha256:" + "f" * 64,
|
|
)
|
|
|
|
|
|
class FakeState:
|
|
def __init__(
|
|
self,
|
|
plan: dict[str, Any],
|
|
*,
|
|
release_failure_status: int | None = None,
|
|
lifecycle_status: str = "executed",
|
|
mismatch_artifact: bool = False,
|
|
) -> None:
|
|
self.plan = plan
|
|
self.release_failure_status = release_failure_status
|
|
self.lifecycle_status = lifecycle_status
|
|
self.mismatch_artifact = mismatch_artifact
|
|
self.requests: list[dict[str, Any]] = []
|
|
self.release_calls = 0
|
|
self.approval_calls = 0
|
|
|
|
def read_model(self) -> dict[str, Any]:
|
|
release = self.plan["release_gate"]["payload"]
|
|
approval = self.plan["approval"]["payload"]
|
|
expected = self.plan["expected_executor_request"]
|
|
artifact_sha256 = expected["artifact_sha256"]
|
|
if self.mismatch_artifact:
|
|
artifact_sha256 = "0" * 64
|
|
executed = self.lifecycle_status == "executed"
|
|
executor_refs = (
|
|
["audit://vignette-nas-preview/rollback-executions/nas-g8-proof"]
|
|
if executed
|
|
else []
|
|
)
|
|
receipt_id = "nas-g8-0123456789abcdef01234567" if executed else None
|
|
lifecycle_refs = [*approval["evidence_refs"], *executor_refs]
|
|
return {
|
|
"content_qualifications": [],
|
|
"model_change_gates": [],
|
|
"release_gates": [
|
|
{
|
|
"gate_id": release["gate_id"],
|
|
"release_id": release["manifest"]["release_id"],
|
|
"qualified": True,
|
|
"state": "pending_human_approval",
|
|
"created_at": "2026-08-07T00:00:00Z",
|
|
}
|
|
],
|
|
"gate_artifacts": [
|
|
{
|
|
"artifact_record_id": expected["artifact_record_id"],
|
|
"owner_kind": "release_gate",
|
|
"owner_id": release["gate_id"],
|
|
"artifact_kind": "rollback",
|
|
"artifact_id": expected["artifact_id"],
|
|
"content_sha256": artifact_sha256,
|
|
"provenance_uri": expected["artifact_provenance_uri"],
|
|
"created_at": "2026-08-07T00:00:00Z",
|
|
}
|
|
],
|
|
"approvals": [
|
|
{
|
|
"approval_event_id": approval["approval_event_id"],
|
|
"target_kind": "release_gate",
|
|
"target_id": release["gate_id"],
|
|
"decision": "authorize_rollback",
|
|
"reason_code": approval["reason_code"],
|
|
"evidence_refs": approval["evidence_refs"],
|
|
"created_at": "2026-08-07T00:00:00Z",
|
|
}
|
|
],
|
|
"catalog_entries": [],
|
|
"lifecycle_events": [
|
|
{
|
|
"lifecycle_event_id": approval["effect_record_id"],
|
|
"target_kind": "release_gate",
|
|
"target_id": release["gate_id"],
|
|
"event_type": "rollback",
|
|
"event_status": self.lifecycle_status,
|
|
"approval_event_id": approval["approval_event_id"],
|
|
"artifact_record_id": expected["artifact_record_id"],
|
|
"evidence_refs": lifecycle_refs,
|
|
"executor_receipt_id": receipt_id,
|
|
"executor_evidence_refs": executor_refs,
|
|
"created_at": "2026-08-07T00:00:00Z",
|
|
}
|
|
],
|
|
"incidents": [],
|
|
"regression_dag_nodes": [],
|
|
"data_classification": MODULE.DATA_CLASSIFICATION,
|
|
"silent_auto_promotion_allowed": False,
|
|
"raw_transcript_included": False,
|
|
"pii_included": False,
|
|
"clinical_claim_allowed": False,
|
|
}
|
|
|
|
|
|
def make_handler(state: FakeState) -> type[BaseHTTPRequestHandler]:
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
return
|
|
|
|
def _body(self) -> dict[str, Any]:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
raw = self.rfile.read(length)
|
|
return json.loads(raw) if raw else {}
|
|
|
|
def _reply(
|
|
self,
|
|
status: int,
|
|
body: dict[str, Any],
|
|
*,
|
|
cookie: bool = False,
|
|
) -> None:
|
|
raw = json.dumps(body).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
if cookie:
|
|
self.send_header("Set-Cookie", f"sid={COOKIE_VALUE}; HttpOnly; Path=/")
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def do_POST(self) -> None:
|
|
body = self._body()
|
|
state.requests.append(
|
|
{
|
|
"method": "POST",
|
|
"path": self.path,
|
|
"body": body,
|
|
"internal_token": self.headers.get(MODULE.INTERNAL_TOKEN_HEADER),
|
|
"cookie": self.headers.get("Cookie"),
|
|
"origin": self.headers.get("Origin"),
|
|
}
|
|
)
|
|
if self.path == MODULE.RELEASE_GATE_PATH:
|
|
if state.release_failure_status is not None:
|
|
self._reply(
|
|
state.release_failure_status,
|
|
{"detail": f"sensitive-{INTERNAL_TOKEN}-{ADMIN_EMAIL}"},
|
|
)
|
|
return
|
|
state.release_calls += 1
|
|
payload = state.plan["release_gate"]["payload"]
|
|
self._reply(
|
|
201,
|
|
{
|
|
"submission_id": payload["submission_id"],
|
|
"gate_id": payload["gate_id"],
|
|
"qualified": True,
|
|
"state": "pending_human_approval",
|
|
"human_approval_required": True,
|
|
"promotion_executed": False,
|
|
"idempotent_replay": state.release_calls > 1,
|
|
},
|
|
)
|
|
return
|
|
if self.path == MODULE.DEV_LOGIN_PATH:
|
|
self._reply(
|
|
200,
|
|
{"user_id": "not-printed", "role": "admin"},
|
|
cookie=True,
|
|
)
|
|
return
|
|
if self.path == MODULE.APPROVAL_PATH:
|
|
if self.headers.get("Cookie") != f"sid={COOKIE_VALUE}":
|
|
self._reply(403, {"detail": "cookie required"})
|
|
return
|
|
state.approval_calls += 1
|
|
payload = state.plan["approval"]["payload"]
|
|
self._reply(
|
|
201,
|
|
{
|
|
"submission_id": payload["submission_id"],
|
|
"approval_event_id": payload["approval_event_id"],
|
|
"target_kind": payload["target_kind"],
|
|
"target_id": payload["target_id"],
|
|
"decision": payload["decision"],
|
|
"effect_record_id": payload["effect_record_id"],
|
|
"idempotent_replay": state.approval_calls > 1,
|
|
},
|
|
)
|
|
return
|
|
self._reply(404, {"detail": "not found"})
|
|
|
|
def do_GET(self) -> None:
|
|
state.requests.append(
|
|
{
|
|
"method": "GET",
|
|
"path": self.path,
|
|
"body": None,
|
|
"internal_token": self.headers.get(MODULE.INTERNAL_TOKEN_HEADER),
|
|
"cookie": self.headers.get("Cookie"),
|
|
}
|
|
)
|
|
if self.path == MODULE.READ_MODEL_PATH:
|
|
self._reply(200, state.read_model())
|
|
return
|
|
self._reply(404, {"detail": "not found"})
|
|
|
|
return Handler
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def fake_api(state: FakeState):
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(state))
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
class G8RollbackProbeTest(unittest.TestCase):
|
|
def test_prepare_is_deterministic_and_exactly_binds_executor_manifest(self) -> None:
|
|
first = sample_plan()
|
|
second = sample_plan()
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(MODULE.PLAN_FIELDS, set(first))
|
|
self.assertEqual(
|
|
first["approval"]["payload"]["effect_record_id"],
|
|
first["expected_executor_request"]["idempotency_key"],
|
|
)
|
|
self.assertEqual(
|
|
first["approval"]["payload"]["approval_event_id"],
|
|
first["expected_executor_request"]["approval_event_id"],
|
|
)
|
|
self.assertEqual(
|
|
first["release_gate"]["payload"]["gate_id"],
|
|
first["expected_executor_request"]["target_id"],
|
|
)
|
|
self.assertEqual(
|
|
first["release_gate"]["payload"]["manifest"]["release_id"],
|
|
first["expected_executor_request"]["subject_id"],
|
|
)
|
|
manifest = first["executor_manifest"]["payload"]
|
|
self.assertEqual(MODULE.EXECUTOR_MANIFEST_SCHEMA, manifest["schema_version"])
|
|
self.assertEqual(1, len(manifest["artifacts"]))
|
|
self.assertEqual(
|
|
first["expected_executor_request"]["artifact_record_id"],
|
|
manifest["artifacts"][0]["artifact_record_id"],
|
|
)
|
|
self.assertFalse(first["safety"]["contains_secrets"])
|
|
self.assertFalse(first["safety"]["contains_pii"])
|
|
|
|
def test_prepared_manifest_loads_in_current_d31b0d_executor_contract(self) -> None:
|
|
plan = sample_plan("executor-contract")
|
|
with TemporaryDirectory(prefix="g8-probe-executor-contract-") as temp:
|
|
manifest_path = Path(temp) / "manifest.json"
|
|
manifest_path.write_bytes(
|
|
MODULE._manifest_bytes(plan["executor_manifest"]["payload"])
|
|
)
|
|
loaded = EXECUTOR_MODULE.load_manifest(
|
|
manifest_path,
|
|
plan["executor_manifest"]["file_sha256"],
|
|
)
|
|
artifact = next(iter(loaded.artifacts.values()))
|
|
expected = plan["expected_executor_request"]
|
|
self.assertEqual(expected["idempotency_key"], artifact.idempotency_key)
|
|
self.assertEqual(expected["approval_event_id"], artifact.approval_event_id)
|
|
self.assertEqual(expected["artifact_record_id"], artifact.artifact_record_id)
|
|
self.assertEqual(expected["artifact_sha256"], artifact.artifact_sha256)
|
|
self.assertEqual(expected["target_id"], artifact.target_id)
|
|
self.assertEqual(expected["subject_id"], artifact.subject_id)
|
|
self.assertEqual(expected["rollback_target_id"], artifact.rollback_target_id)
|
|
|
|
def test_prepare_cli_emits_repeatable_plan_and_exact_manifest_bytes(self) -> None:
|
|
arguments = [
|
|
"prepare",
|
|
"--suffix",
|
|
"prepare-cli",
|
|
"--artifact-id",
|
|
"nas-preview-rollback-prepare-cli",
|
|
"--artifact-sha256",
|
|
"a" * 64,
|
|
"--artifact-provenance-uri",
|
|
"repo://vignette/releases/nas-preview-rollback-prepare-cli",
|
|
"--evidence-ref",
|
|
"audit://vignette-nas-preview/g8-rollback/prepare-cli/authorization",
|
|
"--compose-sha256",
|
|
"b" * 64,
|
|
"--env-sha256",
|
|
"c" * 64,
|
|
"--resolved-config-sha256",
|
|
"d" * 64,
|
|
"--api-image",
|
|
"sha256:" + "e" * 64,
|
|
"--web-image",
|
|
"sha256:" + "f" * 64,
|
|
]
|
|
with TemporaryDirectory(prefix="g8-probe-prepare-cli-") as temp:
|
|
plan_path = Path(temp) / "plan.json"
|
|
manifest_path = Path(temp) / "manifest.json"
|
|
full_arguments = [
|
|
*arguments,
|
|
"--output",
|
|
str(plan_path),
|
|
"--executor-manifest-output",
|
|
str(manifest_path),
|
|
]
|
|
first_stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(first_stdout):
|
|
first_exit = MODULE.main(full_arguments)
|
|
first_plan_bytes = plan_path.read_bytes()
|
|
first_manifest_bytes = manifest_path.read_bytes()
|
|
second_stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(second_stdout):
|
|
second_exit = MODULE.main(full_arguments)
|
|
self.assertEqual(0, first_exit)
|
|
self.assertEqual(0, second_exit)
|
|
self.assertEqual(first_stdout.getvalue(), second_stdout.getvalue())
|
|
plan = json.loads(first_plan_bytes)
|
|
self.assertEqual(plan, json.loads(first_stdout.getvalue()))
|
|
self.assertEqual(
|
|
plan["executor_manifest"]["file_sha256"],
|
|
hashlib.sha256(first_manifest_bytes).hexdigest(),
|
|
)
|
|
|
|
def test_plan_extra_field_and_semantic_tampering_fail_closed(self) -> None:
|
|
extra = sample_plan()
|
|
extra["unexpected"] = True
|
|
with self.assertRaisesRegex(MODULE.ProbeError, "invalid_plan_shape"):
|
|
MODULE.validate_plan(extra)
|
|
|
|
changed = copy.deepcopy(sample_plan())
|
|
changed["expected_executor_request"]["artifact_sha256"] = "0" * 64
|
|
core = {key: value for key, value in changed.items() if key != "plan_sha256"}
|
|
changed["plan_sha256"] = MODULE._sha256_json(core)
|
|
with self.assertRaisesRegex(
|
|
MODULE.ProbeError, "executor_request_binding_mismatch"
|
|
):
|
|
MODULE.validate_plan(changed)
|
|
|
|
def test_prepare_rejects_pii_or_query_material_in_evidence_refs(self) -> None:
|
|
common = {
|
|
"suffix": "unsafe-evidence",
|
|
"artifact_id": "nas-preview-rollback-unsafe-evidence",
|
|
"artifact_sha256": "a" * 64,
|
|
"artifact_provenance_uri": (
|
|
"repo://vignette/releases/nas-preview-rollback-unsafe-evidence"
|
|
),
|
|
"compose_sha256": "b" * 64,
|
|
"env_sha256": "c" * 64,
|
|
"resolved_config_sha256": "d" * 64,
|
|
"api_image": "sha256:" + "e" * 64,
|
|
"web_image": "sha256:" + "f" * 64,
|
|
}
|
|
for evidence_ref in (
|
|
"audit://vignette/users/person@example.com/authorization",
|
|
"https://evidence.invalid/proof?token=secret",
|
|
):
|
|
with self.subTest(evidence_ref=evidence_ref):
|
|
with self.assertRaisesRegex(MODULE.ProbeError, "invalid_evidence_ref"):
|
|
MODULE.build_plan(evidence_refs=[evidence_ref], **common)
|
|
|
|
def test_execute_uses_current_shapes_and_verifies_idempotent_receipt(self) -> None:
|
|
plan = sample_plan()
|
|
state = FakeState(plan)
|
|
with fake_api(state) as control_url:
|
|
result = MODULE.execute_plan(
|
|
plan=plan,
|
|
control_plane_base_url=control_url,
|
|
preview_base_url="http://100.116.83.60:8088/api",
|
|
internal_token=INTERNAL_TOKEN,
|
|
admin_email=ADMIN_EMAIL,
|
|
timeout=5,
|
|
)
|
|
self.assertEqual(MODULE.RESULT_FIELDS, set(result))
|
|
self.assertEqual("executed", result["lifecycle_status"])
|
|
self.assertTrue(result["binding_verified"])
|
|
self.assertTrue(result["release_gate_idempotent_replay"])
|
|
self.assertTrue(result["approval_idempotent_replay"])
|
|
self.assertFalse(result["contains_secrets"])
|
|
self.assertFalse(result["contains_pii"])
|
|
|
|
release_requests = [
|
|
item for item in state.requests if item["path"] == MODULE.RELEASE_GATE_PATH
|
|
]
|
|
approval_requests = [
|
|
item for item in state.requests if item["path"] == MODULE.APPROVAL_PATH
|
|
]
|
|
self.assertEqual(2, len(release_requests))
|
|
self.assertEqual(2, len(approval_requests))
|
|
self.assertTrue(
|
|
all(
|
|
item["body"] == plan["release_gate"]["payload"]
|
|
for item in release_requests
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(item["internal_token"] == INTERNAL_TOKEN for item in release_requests)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
item["body"] == plan["approval"]["payload"]
|
|
for item in approval_requests
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(item["cookie"] == f"sid={COOKIE_VALUE}" for item in approval_requests)
|
|
)
|
|
|
|
def test_equal_control_and_preview_origin_is_rejected_before_http(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
MODULE.ProbeError, "main_preview_self_rollback_forbidden"
|
|
):
|
|
MODULE.execute_plan(
|
|
plan=sample_plan(),
|
|
control_plane_base_url="http://127.0.0.1:8010/api",
|
|
preview_base_url="http://127.0.0.1:8010",
|
|
internal_token=INTERNAL_TOKEN,
|
|
admin_email=ADMIN_EMAIL,
|
|
timeout=5,
|
|
)
|
|
|
|
def test_read_model_artifact_mismatch_and_failed_receipt_fail_closed(self) -> None:
|
|
for state, expected in (
|
|
(FakeState(sample_plan("mismatch"), mismatch_artifact=True), "artifact"),
|
|
(FakeState(sample_plan("failed"), lifecycle_status="failed"), "receipt"),
|
|
):
|
|
with self.subTest(expected=expected):
|
|
with fake_api(state) as control_url:
|
|
with self.assertRaisesRegex(MODULE.ProbeError, expected):
|
|
MODULE.execute_plan(
|
|
plan=state.plan,
|
|
control_plane_base_url=control_url,
|
|
preview_base_url="http://127.0.0.1:9",
|
|
internal_token=INTERNAL_TOKEN,
|
|
admin_email=ADMIN_EMAIL,
|
|
timeout=5,
|
|
)
|
|
|
|
def test_401_403_409_never_expose_response_body_or_secrets(self) -> None:
|
|
for status in (401, 403, 409):
|
|
state = FakeState(
|
|
sample_plan(f"http-{status}"), release_failure_status=status
|
|
)
|
|
with self.subTest(status=status):
|
|
with fake_api(state) as control_url:
|
|
with self.assertRaisesRegex(
|
|
MODULE.ProbeError, f"http_{status}_post"
|
|
) as raised:
|
|
MODULE.execute_plan(
|
|
plan=state.plan,
|
|
control_plane_base_url=control_url,
|
|
preview_base_url="http://127.0.0.1:9",
|
|
internal_token=INTERNAL_TOKEN,
|
|
admin_email=ADMIN_EMAIL,
|
|
timeout=5,
|
|
)
|
|
message = str(raised.exception)
|
|
self.assertNotIn(INTERNAL_TOKEN, message)
|
|
self.assertNotIn(ADMIN_EMAIL, message)
|
|
self.assertNotIn(COOKIE_VALUE, message)
|
|
self.assertNotIn("sensitive", message)
|
|
|
|
def test_cli_output_contains_no_token_cookie_or_admin_identity(self) -> None:
|
|
plan = sample_plan("cli-output")
|
|
state = FakeState(plan)
|
|
with TemporaryDirectory(prefix="g8-probe-test-") as temp:
|
|
plan_path = Path(temp) / "plan.json"
|
|
MODULE._write_json(plan_path, plan)
|
|
with fake_api(state) as control_url:
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{
|
|
"PROBE_TEST_INTERNAL_TOKEN": INTERNAL_TOKEN,
|
|
"PROBE_TEST_ADMIN_EMAIL": ADMIN_EMAIL,
|
|
},
|
|
clear=False,
|
|
):
|
|
with (
|
|
contextlib.redirect_stdout(stdout),
|
|
contextlib.redirect_stderr(stderr),
|
|
):
|
|
exit_code = MODULE.main(
|
|
[
|
|
"execute",
|
|
"--plan",
|
|
str(plan_path),
|
|
"--control-plane-base-url",
|
|
control_url,
|
|
"--preview-base-url",
|
|
"http://127.0.0.1:9",
|
|
"--internal-token-env",
|
|
"PROBE_TEST_INTERNAL_TOKEN",
|
|
"--admin-email-env",
|
|
"PROBE_TEST_ADMIN_EMAIL",
|
|
]
|
|
)
|
|
self.assertEqual(0, exit_code)
|
|
combined = stdout.getvalue() + stderr.getvalue()
|
|
self.assertNotIn(INTERNAL_TOKEN, combined)
|
|
self.assertNotIn(ADMIN_EMAIL, combined)
|
|
self.assertNotIn(COOKIE_VALUE, combined)
|
|
self.assertEqual(MODULE.RESULT_FIELDS, set(json.loads(stdout.getvalue())))
|
|
|
|
def test_help_states_separate_control_plane_prerequisite(self) -> None:
|
|
help_text = MODULE._build_parser().format_help()
|
|
self.assertIn("main preview API must not self-authorize rollback", help_text)
|
|
self.assertIn("separately deployed control-plane API", help_text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|