密封测试环境

概念定义

连着互联网的测试不是测试——它是对别人在线率的许愿。密封性把许愿换成所有权：运行掌控它触碰的一切，结果在任何地方都可复现。

工作机制

依赖按运行或按套件启动：固定镜像的容器、本地套接字上的进程内 fake、云服务的模拟器。出站防火墙让任何意外外呼大声失败；确定性种子与注入时钟消除环境时间。套件输出成为代码的纯函数。

使用场景

flaky 追踪外部可用性、CI 队列被共享环境卡死、以及必须证明出站被封闭的不可信载荷安全测试。前期投入会在第一次外部故障时回本。

常见误区

密封不等于“不许用数据库”——按测试启动的容器化 Postgres 就是密封的；把集成测试指向共享预发环境，等于把每次部署都继承成测试失败。stub 一个 API 而代码仍调用真实 CDN，慢速漂移照样渗进套件——只有出站阻断能证明没有静默外呼。时钟与随机性在网络封死后依然是环境因素。模拟器保真度会偏离真实服务——用面向提供方的契约测试互补。

代码示例

import tempfile
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator, Optional


class SandboxEscapeError(ValueError):
    """Raised when a requested path would leave the sandbox root."""


@dataclass(frozen=True)
class Sandbox:
    root: Path

    def resolve(self, *parts: str) -> Path:
        target = self.root.joinpath(*parts).resolve()
        if target != self.root and self.root not in target.parents:
            raise SandboxEscapeError("path escapes the sandbox: " + str(target))
        return target

    def write(self, relative_path: str, content: str) -> Path:
        target = self.resolve(relative_path)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content, encoding="utf-8")
        return target

    def read(self, relative_path: str) -> str:
        return self.resolve(relative_path).read_text(encoding="utf-8")

    def exists(self, relative_path: str) -> bool:
        return self.resolve(relative_path).exists()


@contextmanager
def sandbox(prefix: Optional[str] = None) -> Iterator[Sandbox]:
    """Yield a private temporary sandbox and always remove it afterwards."""
    with tempfile.TemporaryDirectory(prefix=prefix or "devfactory-") as raw:
        yield Sandbox(root=Path(raw).resolve())

export type Outcome<T> = { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: Error };

export function ok<T>(value: T): Outcome<T> {
  return { ok: true, value };
}

export function fail(message: string): Outcome<never> {
  return { ok: false, error: new Error(message) };
}

export interface AttemptLog {
  readonly index: number;
  readonly at: number;
  readonly result: "ok" | "error";
}

export interface RetryHarness<T> {
  readonly attempts: AttemptLog[];
  readonly loader: () => Promise<T>;
  readonly exhausted: () => boolean;
}

export function createHarness<T>(
  script: readonly Outcome<T>[],
  clock: { now: () => number; setTimeout: (fn: () => void, ms: number) => unknown },
): RetryHarness<T> {
  const attempts: AttemptLog[] = [];
  let index = 0;

  const loader = async (): Promise<T> =>
    new Promise<T>((resolve, reject) => {
      const outcome = script[index];
      attempts.push({
        index,
        at: clock.now(),
        result: outcome.ok ? "ok" : "error",
      });
      index += 1;
      if (outcome.ok) resolve(outcome.value);
      else reject(outcome.error);
    });

  return {
    attempts,
    loader,
    exhausted: () => index >= script.length,
  };
}