"""幂等键:客户端网络抖动重试时,不能建出两个工作区。""" from __future__ import annotations from httpx import AsyncClient from .conftest import login async def test_same_key_returns_first_result(client: AsyncClient) -> None: user = await login(client) headers = {**user["headers"], "Idempotency-Key": "k-1"} payload = {"name": "幂等工作区"} first = await client.post("/api/v1/workspaces", json=payload, headers=headers) second = await client.post("/api/v1/workspaces", json=payload, headers=headers) assert first.status_code == 200 assert first.json()["data"]["id"] == second.json()["data"]["id"] assert second.headers.get("X-Idempotent-Replay") == "1" listed = await client.get("/api/v1/workspaces", headers=user["headers"]) assert len(listed.json()["data"]["items"]) == 1 async def test_different_keys_create_different_workspaces(client: AsyncClient) -> None: user = await login(client) a = await client.post( "/api/v1/workspaces", json={"name": "一号"}, headers={**user["headers"], "Idempotency-Key": "k-a"}, ) b = await client.post( "/api/v1/workspaces", json={"name": "二号"}, headers={**user["headers"], "Idempotency-Key": "k-b"}, ) assert a.json()["data"]["id"] != b.json()["data"]["id"] async def test_errors_are_not_cached(client: AsyncClient) -> None: user = await login(client) headers = {**user["headers"], "Idempotency-Key": "k-err"} bad = await client.post("/api/v1/workspaces", json={"name": ""}, headers=headers) assert bad.status_code == 422 good = await client.post("/api/v1/workspaces", json={"name": "修好了"}, headers=headers) assert good.status_code == 200 assert "X-Idempotent-Replay" not in good.headers async def test_keys_are_scoped_per_user(client: AsyncClient) -> None: alice = await login(client) bob = await login(client) a = await client.post( "/api/v1/workspaces", json={"name": "Alice 的"}, headers={**alice["headers"], "Idempotency-Key": "shared"}, ) b = await client.post( "/api/v1/workspaces", json={"name": "Bob 的"}, headers={**bob["headers"], "Idempotency-Key": "shared"}, ) assert a.json()["data"]["id"] != b.json()["data"]["id"]