尖峰测试（Spike Tests）

测试什么

尖峰测试让流量从基线几乎瞬间跳到高位，短暂保持后回落。问题不是“最大吞吐量是多少”，而是“当需求变化快于系统预期时会发生什么”。自动伸缩需要时间反应，缓存是冷的，连接池按基线规模配置，队列吸收差额且可能永远排不空。尖峰测试暴露的正是这个序列。

设计测试

从真实基线开始，在数秒到几分钟内陡升，保持足够长以触发自动伸缩并让队列堆积，然后回到基线并继续观测。必须包含恢复窗口，因为系统往往在尖峰之后而非期间失败：重试积压、重连惊群，或仍在追赶的数据库。与浸泡测试相比，它短促而猛烈，而不是漫长而平稳。

观察什么

在高峰和恢复期间都跟踪错误率和延迟。观察队列深度、消费者滞后、伸缩事件、冷启动时长以及依赖返回的限流响应。检查重试是否放大了负载。系统可能扛过尖峰，却在所有超时客户端同时重试时崩溃。

与开发尖峰的区别及局限

同一个词也指敏捷规划中的限时调研。语境可以区分：负载尖峰衡量容量，开发尖峰降低技术不确定性。本文讨论的是负载测试含义。尖峰测试可能伤害共享环境和下游伙伴，因此要隔离并提前通知负责人。它不能替代容量规划和减载设计。例如，服务可能通过一次尖峰测试，却在流量组合不同的真实活动中失败。应把结果用于调整限流和伸缩策略，而不是当作一次性证书。

代码示例

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())

package main

import (
	"fmt"
	"strings"
	"testing"
)

func normalizeEmail(raw string) (string, error) {
	email := strings.TrimSpace(strings.ToLower(raw))
	at := strings.IndexByte(email, '@')
	if at <= 0 || at == len(email)-1 || strings.Count(email, "@") != 1 {
		return "", fmt.Errorf("invalid email %q", raw)
	}
	return email, nil
}

func TestNormalizeEmail(t *testing.T) {
	t.Parallel()
	cases := []struct {
		name    string
		input   string
		want    string
		wantErr bool
	}{
		{"lowercases and trims", "  [email protected] ", "[email protected]", false},
		{"rejects missing at", "ada.example.com", "", true},
		{"rejects empty local part", "@example.com", "", true},
		{"rejects double at", "ada@@example.com", "", true},
	}

	for _, testCase := range cases {
		testCase := testCase
		t.Run(testCase.name, func(t *testing.T) {
			t.Parallel()
			got, err := normalizeEmail(testCase.input)
			if testCase.wantErr {
				if err == nil {
					t.Fatalf("expected error, got %q", got)
				}
				return
			}
			if err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
			if got != testCase.want {
				t.Fatalf("got %q, want %q", got, testCase.want)
			}
		})
	}
}

func ExampleNormalizeEmail() {
	email, _ := normalizeEmail("  [email protected] ")
	fmt.Println(email)
	// Output: [email protected]
}