失败路径测试

概念定义

系统为存活而设计，却只为成功而测试。存活代码——超时、重试、回退、清理——从未执行就被发到线上，直到生产第一次替你运行它。

工作机制

枚举依赖失败分类学：连接拒绝、读超时、写超时、截断响应、语义错误、慢速劣化。逐项在边界注入——代理、mock、容器——并断言可观察契约：什么错误暴露、留下什么状态、重试是否尊重预算与抖动、断路器是否打开。

使用场景

任何要在事故里依赖的韧性机制都要先测：每个超时值与重试策略都是可测契约。集成套件跑注入矩阵，混沌实验验证这些行为在集群层的组合。

常见误区

在错误层级注入——mock 一个真实依赖根本抛不出的异常——验证的是虚构。超时测试需要可控时钟，否则按真实超时时长跑并撑爆套件。断言“调用抛了”不够：必须断言资源已释放、状态已回滚；两阶段写入被打断的部分失败态不显式断言，损坏就会以绿灯漏过。灾难性故障之外别漏慢速劣化——多数真实 bug 住在那里。

代码示例

package main

import (
	"errors"
	"testing"
)

var errNotFound = errors.New("order not found")

func ShippingCost(weightKg int, express bool, tier string) (int, error) {
	if weightKg <= 0 {
		return 0, errors.New("weight must be positive")
	}
	if weightKg > 30 {
		return 0, errors.New("weight exceeds courier limit")
	}

	base := 8
	if weightKg > 10 {
		base += (weightKg - 10) * 2
	}
	if express {
		base *= 2
	}
	if tier == "gold" {
		base -= 4
	}
	return base, nil
}

func TestShippingCost(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name    string
		weight  int
		express bool
		tier    string
		want    int
		wantErr error
	}{
		{name: "light standard", weight: 2, express: false, tier: "", want: 8},
		{name: "light express", weight: 2, express: true, tier: "", want: 16},
		{name: "heavy adds per kg", weight: 15, express: false, tier: "", want: 18},
		{name: "heavy express gold", weight: 15, express: true, tier: "gold", want: 32},
		{name: "gold floor at zero", weight: 1, express: false, tier: "gold", want: 4},
		{name: "zero weight rejected", weight: 0, express: false, tier: "", wantErr: errors.New("weight must be positive")},
		{name: "over courier limit", weight: 31, express: false, tier: "", wantErr: errors.New("weight exceeds courier limit")},
	}

	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			got, err := ShippingCost(tt.weight, tt.express, tt.tier)
			if tt.wantErr != nil {
				if err == nil || err.Error() != tt.wantErr.Error() {
					t.Fatalf("ShippingCost(%d, %v, %q) error = %v, want %v",
						tt.weight, tt.express, tt.tier, err, tt.wantErr)
				}
				return
			}
			if err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
			if got != tt.want {
				t.Errorf("ShippingCost(%d, %v, %q) = %d, want %d",
					tt.weight, tt.express, tt.tier, got, tt.want)
			}
		})
	}
}

import httpx
import pytest


def fetch_invoice(base_url: str, invoice_id: str) -> dict:
    with httpx.Client(base_url=base_url) as client:
        resp = client.get(f'/invoices/{invoice_id}')
        resp.raise_for_status()
        return resp.json()


def test_fetch_invoice_parses_payload():
    captured: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        captured['url'] = str(request.url)
        captured['auth'] = request.headers.get('Authorization')
        return httpx.Response(200, json={'id': 'inv-1', 'total': 4200})

    client = httpx.Client(
        transport=httpx.MockTransport(handler),
        base_url='https://api.example.test',
        headers={'Authorization': 'Bearer test-token'},
    )

    with client:
        resp = client.get('/invoices/inv-1')
        data = resp.json()

    assert data['total'] == 4200
    assert captured['url'].endswith('/invoices/inv-1')
    assert captured['auth'] == 'Bearer test-token'


@pytest.mark.parametrize('status', [500, 502, 503])
def test_fetch_invoice_raises_on_server_error(status):
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status)

    client = httpx.Client(transport=httpx.MockTransport(handler))
    with pytest.raises(httpx.HTTPStatusError):
        client.get('https://api.example.test/invoices/inv-2')