diff --git a/README.md b/README.md index 0845096..90f6bf5 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,49 @@ Patbond 的 AI 生成服务（Python）。负责调用 AI 服务商 API 完成 - Java 主动轮询本服务查询进度；回调是将来的可选优化，当前不启用。 - 请求体**不含 `userId`**（数据最小化）：完成工作只需 `jobId`、生成参数与预签名上传地址。 +## 运行 + +```bash +cp .env.sample .env # 填入实际值；.env 已被 gitignore +python -m app # 默认 127.0.0.1:8000 +``` + +**只监听回环**是默认：本服务只接受来自 `patbond-creation` 的内部调用，不该对外 +可达（与项目「默认拒绝、需要外部访问时经 nginx 反代」的纪律一致）。 + +探活：`GET /health`（**在 `/internal` 之外，不需要服务票**）。它会报出实际装配了 +哪些适配器——缺服务商凭证时适配器不装配，而那个症状（任务永远排队）离根因很远， +让探活直接说出来能省很多排查时间。 + +## 接口 + +| 端点 | 说明 | +| --- | --- | +| `POST /internal/generation/tasks` | 提交任务，返回**服务商**任务 ID | +| `GET /internal/generation/tasks/{provider_request_id}` | 查询任务 | +| `GET /health` | 探活，无需服务票 | + +**权威契约是仓库根目录的 `openapi-internal.yaml`**（手写），不是 FastAPI 自动生成 +的 `/openapi.json`（后者仅供调试，交互文档已关闭）。两者的一致性由 +`tests/api/test_contract_drift.py` 守着——比对路径、方法、字段集、枚举取值与必填项， +**不比对字节**（那永远不会相同，只会制造一个天天红然后被 skip 绕过的门禁）。 + +### 三类失败必须分开 + +| 类别 | 形态 | 调用方该做什么 | +| --- | --- | --- | +| `service` | 非 2xx + `error.kind = "service"` | 我方 bug 或配置问题。除 `internal_error`/`provider_unavailable` 外不该重试 | +| `provider` | 非 2xx + `error.kind = "provider"` + 完整分类 | **一律以 `retry` 字段为准**，不要自己解读厂商码 | +| 任务失败 | **200** + `state = "failed"` + `failure` | 同上看 `retry` | + +第三类刻意不走错误通道：查询一个失败的任务，**HTTP 调用是成功的**，失败的是任务。 +做成非 2xx 会让「查不到任务」与「任务失败了」难以区分。 + +判别字段是 `error.kind`，**不是 HTTP 状态码**——状态码在两类之间会撞。 + +`provider` 信封与 `failure` 的分类字段**故意完全一致**，故调用方可复用同一段处置 +逻辑，不必为「提交时被限流」与「任务失败于限流」写两套分支。 + ## 代码结构 分四层，依赖**单向**向内（`config` → `providers` → `domain`），由 @@ -98,7 +141,9 @@ app/ ## 本地门禁 -与 CI 同一套命令（`.gitea/workflows/ci.yml`）： +与 CI 同一套命令（`.gitea/workflows/ci.yml`）。**逐条分开跑并读各自的退出码**—— +`ruff check . | tail -1; echo $?` 读到的是 `tail` 的退出码，会把一次真实的 lint +失败读成通过（本仓已因此红过一次）： ```bash sh scripts/check-secrets.sh --all # 凭证防泄漏全量自查 diff --git a/app/__main__.py b/app/__main__.py new file mode 100644 index 0000000..6ceccfa --- /dev/null +++ b/app/__main__.py @@ -0,0 +1,25 @@ +"""启动入口：`python -m app`。 + +配置全部从环境变量读（见 `app/config.py` 与 `.env.sample`）。缺服务商凭证时服务 +仍会启动，只是该适配器不装配——`/health` 会报出实际装配了哪些，便于确认。 +""" + +from __future__ import annotations + +import os + +import uvicorn + +from app.api.main import create_app + + +def main() -> None: + # 只监听回环是默认：本服务只接受来自 patbond-creation 的内部调用， + # 不该对外可达（与项目「默认拒绝、需要外部访问时经 nginx 反代」的纪律一致）。 + host = os.environ.get("PATBOND_AI_HOST", "127.0.0.1") + port = int(os.environ.get("PATBOND_AI_PORT", "8000")) + uvicorn.run(create_app(), host=host, port=port) + + +if __name__ == "__main__": + main() diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/auth.py b/app/api/auth.py new file mode 100644 index 0000000..74265cc --- /dev/null +++ b/app/api/auth.py @@ -0,0 +1,78 @@ +"""服务身份 JWT 校验（ADR-023 §5）。 + +本模块是**唯一**的鉴权入口。三条纪律，每条都有测试钉住： + +1. **强制校验 `aud` 与 `typ`。** 这是独立密钥对方案唯一的意义所在——Java 侧那边 + 现有的 `JwtVerifier` 只验签名、不校验 `aud`/`iss`，正是因为如此才必须用一对 + 独立密钥；如果本服务也只验签名，那对独立密钥就白搭了。 +2. **用户 token 一律拒绝。** 它们由另一对密钥签发，签名校验本身就会失败；但即使 + 有人误把用户密钥配到这里，缺 `aud` 与 `typ=service` 也会被拦下——两道。 +3. **缺公钥时拒绝所有请求，而不是放行。** 配置缺失必须是「全拒」而非「全过」： + 前者立刻暴露，后者是敞开的门。这与 Java 侧 `InternalAuthFilter` 未配置 + `internal-token` 时全拒 `/internal/**` 的既有选择一致。 +""" + +from __future__ import annotations + +import logging + +import jwt +from fastapi import Depends, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +from app.config import ServiceAuthSettings + +from .errors import ServiceError, ServiceErrorCode + +log = logging.getLogger(__name__) + +#: auto_error=False：自己产生 401 以便统一走 service 错误信封， +#: 而不是 FastAPI 默认的 `{"detail": ...}`——后者与本契约的两类错误形状都不符。 +_bearer = HTTPBearer(auto_error=False) + + +def verify_service_token( + request: Request, + credentials: HTTPAuthorizationCredentials | None = Depends(_bearer), +) -> None: + """校验服务票；不通过即抛 `ServiceError(unauthenticated)`。 + + 所有失败都返回同一个笼统原因，**不告诉调用方是签名错、过期、还是 aud 不符**。 + 对方是我们自己的服务，真正的原因在本服务日志里；把细节回给调用方只会在将来 + 某天变成攻击者的提示。 + """ + settings: ServiceAuthSettings = request.app.state.settings.service_auth + + if not settings.public_key: + # 配置缺失 → 全拒。绝不「未配置即放行」。 + log.error("PATBOND_SERVICE_JWT_PUBLIC_KEY 未配置，/internal/** 全部拒绝") + raise ServiceError(ServiceErrorCode.UNAUTHENTICATED, "服务端未配置服务票公钥") + + if credentials is None or not credentials.credentials: + raise ServiceError(ServiceErrorCode.UNAUTHENTICATED, "缺少服务票") + + try: + claims = jwt.decode( + credentials.credentials, + settings.public_key, + algorithms=["RS256"], + # audience 与 issuer 交给库强制校验，而不是解出来自己比—— + # 自己比容易漏掉「claim 不存在」这一路。 + audience=settings.audience, + issuer=settings.issuer, + options={"require": ["exp", "aud", "iss", "sub"]}, + ) + except jwt.InvalidTokenError as exc: + # 具体原因只进日志。 + log.warning("服务票校验失败：%s", exc) + raise ServiceError( + ServiceErrorCode.UNAUTHENTICATED, "服务票无效或已过期" + ) from exc + + token_type = claims.get("typ") + if token_type != settings.required_type: + # 用户 token 走到这里的唯一可能是有人误配了用户密钥。拦住它。 + log.warning( + "服务票 typ 不符：期望 %s，实际 %r", settings.required_type, token_type + ) + raise ServiceError(ServiceErrorCode.UNAUTHENTICATED, "服务票类型不符") diff --git a/app/api/errors.py b/app/api/errors.py new file mode 100644 index 0000000..dfc2b28 --- /dev/null +++ b/app/api/errors.py @@ -0,0 +1,86 @@ +"""两类错误的异常与信封（契约 `openapi-internal.yaml` 的核心设计）。 + +**两类失败必须分开**，因为调用方对它们的反应完全不同： + +- `service`（Java↔Python）：我方 bug 或配置问题。除 `internal_error` / + `provider_unavailable` 外都不该重试——重试一个拼错的请求体不会变好。 +- `provider`（Python↔服务商）：以 `retry` 字段为准，调用方不该自己解读厂商码。 + +还有**第三类不在这里**：任务本身失败（生成跑完但失败了）走 `200 + state=failed` ++ `failure`。因为那种情况下 **HTTP 调用是成功的**，失败的是任务。做成错误响应会 +让「查不到任务」与「任务失败了」难以区分。 + +判别字段是 `error.kind`，**不是 HTTP 状态码**：状态码在两类之间会撞（502 既可能是 +服务商挂了也可能是我们打错地址），而 `kind` 不会。 +""" + +from __future__ import annotations + +from enum import Enum + +from app.domain.generation import Failure + + +class ServiceErrorCode(Enum): + """`service` 类错误码，与契约的枚举逐一对应。""" + + #: 服务票缺失、过期、签名错、aud/typ 不符。细节只进日志。 + UNAUTHENTICATED = ("unauthenticated", 401) + #: 请求体不合契约，或参数超出该模型的限制（本地预校验拦下）。 + INVALID_REQUEST = ("invalid_request", 400) + #: 路径里的服务商任务 ID 不存在（含超过服务商 7 天保留期）。 + TASK_NOT_FOUND = ("task_not_found", 404) + #: 适配器未装配（如缺服务商凭证）或 media_kind 无对应适配器。 + PROVIDER_UNAVAILABLE = ("provider_unavailable", 503) + #: 本服务 bug。 + INTERNAL_ERROR = ("internal_error", 500) + + def __init__(self, code: str, http_status: int): + self.code = code + self.http_status = http_status + + +class ServiceError(Exception): + """Java↔Python 之间的问题。""" + + def __init__(self, code: ServiceErrorCode, message: str): + super().__init__(message) + self.code = code + self.message = message + + def to_envelope(self) -> dict: + return { + "error": { + "kind": "service", + "code": self.code.code, + "message": self.message, + } + } + + +class ProviderError(Exception): + """Python↔服务商之间的问题，携带完整分类。 + + 字段与 `Failure` 一致**是刻意的**：调用方因此可以复用同一段处置逻辑，不必为 + 「提交时被限流」和「任务失败于限流」写两套分支。 + """ + + #: 502：本服务本身没问题，是上游调用没成功。 + #: 刻意不透传服务商的状态码——429 既可能是「等几秒」也可能是「等到明天」， + #: 状态码表达不了这个区别，`retry` 才是权威信号。 + HTTP_STATUS = 502 + + def __init__(self, failure: Failure): + super().__init__(failure.message or failure.provider_code) + self.failure = failure + + def to_envelope(self) -> dict: + return { + "error": { + "kind": "provider", + "retry": self.failure.retry.value, + "user_facing": self.failure.user_facing.value, + "provider_code": self.failure.provider_code, + "message": self.failure.message, + } + } diff --git a/app/api/main.py b/app/api/main.py new file mode 100644 index 0000000..8c255cb --- /dev/null +++ b/app/api/main.py @@ -0,0 +1,121 @@ +"""FastAPI 应用工厂。 + +装配顺序：读配置 → 建适配器注册表 → 挂路由与异常处理器。配置与注册表放在 +`app.state` 上，请求处理时从 `request.app` 取——**不用模块级全局**，这样测试能起 +多个配置不同的实例而互不干扰。 + +**异常处理器是本模块的重点**：契约定义了两类错误信封（`kind = service` / +`provider`），而 FastAPI 默认的校验错误是 `{"detail": [...]}`，与两者都不符。若不 +接管，调用方就要处理三种形状——而第三种还是框架的实现细节。 +""" + +from __future__ import annotations + +import logging + +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse + +from app.assembly import build_registry +from app.config import Settings + +from .errors import ProviderError, ServiceError, ServiceErrorCode +from .routes import router + +log = logging.getLogger(__name__) + + +def create_app(settings: Settings | None = None) -> FastAPI: + """建应用。 + + :param settings: 省略即从环境变量读。测试传入显式配置以免依赖环境。 + """ + resolved = settings or Settings.from_env() + logging.basicConfig(level=resolved.log_level) + + app = FastAPI( + title="patbond-ai 内部接口", + version="0.1.0", + description=( + "patbond-creation（Java）调用本服务执行 AI 生成的内部契约。" + "权威契约见仓库根目录的 openapi-internal.yaml——" + "本服务自动生成的 /openapi.json 仅供调试，不是契约。" + ), + # 内部服务不开交互文档：少一个对外表面，且契约以手写 yaml 为准。 + docs_url=None, + redoc_url=None, + ) + app.state.settings = resolved + app.state.registry = build_registry(resolved) + + app.include_router(router) + _install_exception_handlers(app) + _install_health(app) + return app + + +def _install_exception_handlers(app: FastAPI) -> None: + @app.exception_handler(ServiceError) + def handle_service_error(request: Request, exc: ServiceError) -> JSONResponse: + return JSONResponse(status_code=exc.code.http_status, content=exc.to_envelope()) + + @app.exception_handler(ProviderError) + def handle_provider_error(request: Request, exc: ProviderError) -> JSONResponse: + log.warning( + "服务商调用失败 code=%s retry=%s", + exc.failure.provider_code, + exc.failure.retry.value, + ) + return JSONResponse( + status_code=ProviderError.HTTP_STATUS, content=exc.to_envelope() + ) + + @app.exception_handler(RequestValidationError) + def handle_validation_error( + request: Request, exc: RequestValidationError + ) -> JSONResponse: + """把 FastAPI 的校验错误翻成 service 信封。 + + 默认形状是 `{"detail": [...]}`，与契约的两类信封都不符。不接管的话调用方 + 要处理三种形状，而第三种还是框架的实现细节——换个框架就变了。 + """ + first = exc.errors()[0] if exc.errors() else {} + # 字段路径形如 ("body", "duration_sec")，去掉 "body" 前缀更可读。 + location = ".".join( + str(part) for part in first.get("loc", ()) if part != "body" + ) + detail = first.get("msg", "请求体不合契约") + message = f"{location}: {detail}" if location else detail + error = ServiceError(ServiceErrorCode.INVALID_REQUEST, message) + return JSONResponse( + status_code=error.code.http_status, content=error.to_envelope() + ) + + @app.exception_handler(Exception) + def handle_unexpected(request: Request, exc: Exception) -> JSONResponse: + """兜底：本服务的 bug。 + + 完整堆栈只进日志，绝不回给调用方——内部细节外泄没有任何好处。 + """ + log.exception("未处理的异常") + error = ServiceError(ServiceErrorCode.INTERNAL_ERROR, "服务内部错误") + return JSONResponse( + status_code=error.code.http_status, content=error.to_envelope() + ) + + +def _install_health(app: FastAPI) -> None: + @app.get("/health", include_in_schema=False) + def health() -> dict: + """探活。**在 /internal 之外，故不需要服务票**——与 Java 侧各服务 + `/health` 不挂鉴权的既有约定一致。 + + 同时报出已装配的适配器：缺凭证时适配器不装配，而那个症状（任务永远排队） + 离根因很远，让探活直接说出来能省很多排查时间。 + """ + registry = app.state.registry + return { + "status": "ok", + "providers": sorted(registry.codes), + } diff --git a/app/api/routes.py b/app/api/routes.py new file mode 100644 index 0000000..ebd69e3 --- /dev/null +++ b/app/api/routes.py @@ -0,0 +1,108 @@ +"""内部接口路由。 + +**本层只做编排，不含任何厂商知识**：解析请求 → 按 `media_kind` 找适配器 → 调用 → +把领域结果序列化。厂商差异全在 `app/providers/<厂商>/` 内。 + +一处刻意的选择：路径参数是**服务商的任务 ID**，不是我方 `job_id`。本服务不持有 +状态，唯一能用来问服务商的就是这个 ID；若用 `job_id` 就得存一份映射，「无状态」 +不再成立，且会多出一个「Java 拿到 ID 但查不到任何东西」的中间态。 +""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Depends, Request + +from app.domain.generation import MediaKind +from app.providers.base import GenerationProvider +from app.providers.registry import ProviderRegistry +from app.providers.volcengine.client import ArkError + +from .auth import verify_service_token +from .errors import ProviderError, ServiceError, ServiceErrorCode +from .schemas import SubmitRequest, SubmitResponse, TaskResponse + +log = logging.getLogger(__name__) + +router = APIRouter( + prefix="/internal/generation", + dependencies=[Depends(verify_service_token)], +) + + +def _registry(request: Request) -> ProviderRegistry: + return request.app.state.registry + + +def _provider_for( + registry: ProviderRegistry, media_kind: MediaKind +) -> GenerationProvider: + """按 `media_kind` 取适配器。 + + 查不到即 `provider_unavailable`（503）而不是 500：这通常是凭证未配置导致适配器 + 没装配，属环境问题而非代码缺陷，且对调用方而言是**可重试**的——配好就能用。 + """ + for provider in registry.all(): + if provider.media_kind is media_kind: + return provider + raise ServiceError( + ServiceErrorCode.PROVIDER_UNAVAILABLE, + f"没有可用的 {media_kind.value} 生成适配器（检查服务商凭证是否已配置）", + ) + + +@router.post("/tasks", response_model=SubmitResponse) +def submit_task(payload: SubmitRequest, request: Request) -> SubmitResponse: + """提交生成任务，返回服务商任务 ID。 + + 调用方**必须在上游产生任何计费动作之前**把该 ID 落库——火山不支持幂等键， + 这是提交超时后唯一能判断「是否已提交」的依据。 + """ + provider = _provider_for(_registry(request), payload.media_kind) + try: + provider_request_id = provider.submit(payload.to_domain()) + except ArkError as exc: + # 服务商调用失败：原样带上分类，调用方据 retry 决定重试。 + raise ProviderError(exc.failure) from exc + except ValueError as exc: + # 本地预校验拦下的参数问题（超出该模型的时长/分辨率/ratio 限制，或内容为空）。 + # 属 invalid_request 而非 provider 错误——请求根本没发出去。 + raise ServiceError(ServiceErrorCode.INVALID_REQUEST, str(exc)) from exc + + log.info( + "已提交生成任务 job_id=%s provider_request_id=%s", + payload.job_id, + provider_request_id, + ) + return SubmitResponse(provider_request_id=provider_request_id) + + +@router.get("/tasks/{provider_request_id}", response_model=TaskResponse) +def get_task( + provider_request_id: str, + request: Request, + media_kind: MediaKind = MediaKind.VIDEO, +) -> TaskResponse: + """查询任务。 + + **任务失败也返回 200**：HTTP 调用成功了，失败的是任务，`state=failed` + + `failure` 表达它。做成非 2xx 会让「查不到任务」与「任务失败了」难以区分。 + + :param media_kind: 决定问哪个适配器。默认 video——当前只有视频适配器； + 接入图片后调用方应显式传入。刻意不做「遍历所有适配器试一遍」：那会把 + 一个任务 ID 拿去问不相干的服务商。 + """ + provider = _provider_for(_registry(request), media_kind) + try: + result = provider.fetch(provider_request_id) + except ArkError as exc: + # 服务商说任务不存在（含超过 7 天保留期）→ 这是 service 类的 task_not_found， + # 而不是 provider 错误：调用方要做的是停止轮询，不是退避重试。 + if exc.http_status == 404: + raise ServiceError( + ServiceErrorCode.TASK_NOT_FOUND, + f"服务商无此任务（或已超出其 7 天保留期）：{provider_request_id}", + ) from exc + raise ProviderError(exc.failure) from exc + return TaskResponse.of(result) diff --git a/app/api/schemas.py b/app/api/schemas.py new file mode 100644 index 0000000..f6809ce --- /dev/null +++ b/app/api/schemas.py @@ -0,0 +1,136 @@ +"""请求/响应的 pydantic 模型。 + +字段名为 **snake_case**，与契约一致。这使 JSON、pydantic 模型、领域 dataclass +三者字段逐一对应，**无需任何转换层**——若用 camelCase，两端都要加转换。 + +这层与 `app/domain/generation.py` 的分工：领域模型是纯数据、无框架依赖（那是 +「换供应商不动上层」的支点）；本层负责 HTTP 边界的校验与序列化。两者字段重合， +但让领域层依赖 pydantic 会把框架拖进本该最干净的一层。 +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from app.domain.generation import ( + GenerationRequest, + GenerationResult, + MediaKind, +) + + +class SubmitRequest(BaseModel): + """提交生成任务。 + + **刻意不含 `user_id`**（数据最小化，ADR-023 §5）：本服务干活只需要任务参数。 + 需要给服务商做滥用检测时传的是 `safety_identifier`——一个哈希，不是明文 ID。 + """ + + # 拒绝未声明的字段：调用方拼错字段名时应当立刻知道，而不是让那个参数被静默 + # 忽略——后者会表现为「我传了 duration 却没生效」这类极难查的问题。 + model_config = ConfigDict(extra="forbid") + + job_id: str + media_kind: MediaKind + model: str = Field(min_length=1) + prompt: str | None = Field(default=None, max_length=10000) + input_url: str | None = None + ratio: str | None = None + resolution: str | None = None + duration_sec: int | None = None + generate_audio: bool = True + watermark: bool = False + safety_identifier: str | None = Field(default=None, max_length=64) + callback_url: str | None = None + upload_target: str | None = None + + @model_validator(mode="after") + def require_prompt_or_input(self) -> SubmitRequest: + """`prompt` 与 `input_url` 至少要有一个。 + + 空内容必然被服务商拒，但那要花一次往返、还可能计入 RPM。本地先拦。 + """ + if not self.prompt and not self.input_url: + raise ValueError("prompt 与 input_url 至少要有一个") + return self + + def to_domain(self) -> GenerationRequest: + return GenerationRequest( + job_id=self.job_id, + media_kind=self.media_kind, + model=self.model, + prompt=self.prompt, + input_url=self.input_url, + ratio=self.ratio, + resolution=self.resolution, + duration_sec=self.duration_sec, + generate_audio=self.generate_audio, + watermark=self.watermark, + safety_identifier=self.safety_identifier, + callback_url=self.callback_url, + upload_target=self.upload_target, + ) + + +class SubmitResponse(BaseModel): + provider_request_id: str + + +class FailureModel(BaseModel): + """失败分类。三个字段各有明确消费方，见契约说明。""" + + retry: str + user_facing: str + provider_code: str + message: str = "" + + +class TaskResponse(BaseModel): + """任务当前状态。 + + 注意 `state` 的取值集**比服务商的更小**：服务商的 `queued` 映射为 `pending`、 + `expired` 归入 `failed`（它是超时，属失败的一种，重试策略由 `failure` 表达）。 + """ + + provider_request_id: str + state: str + progress: int + asset_url: str | None = None + asset_url_expires_in_sec: int | None = None + cover_url: str | None = None + ratio: str | None = None + resolution: str | None = None + width_px: int | None = None + height_px: int | None = None + duration_ms: int | None = None + output_format: str | None = None + billing_tokens: int | None = None + failure: FailureModel | None = None + + @classmethod + def of(cls, result: GenerationResult) -> TaskResponse: + return cls( + provider_request_id=result.provider_request_id, + state=result.state.value, + progress=result.progress, + asset_url=result.asset_url, + asset_url_expires_in_sec=result.asset_url_expires_in_sec, + cover_url=result.cover_url, + ratio=result.ratio, + resolution=result.resolution, + width_px=result.width_px, + height_px=result.height_px, + duration_ms=result.duration_ms, + output_format=result.output_format, + billing_tokens=result.billing_tokens, + failure=( + FailureModel( + retry=result.failure.retry.value, + user_facing=result.failure.user_facing.value, + provider_code=result.failure.provider_code, + message=result.failure.message, + ) + if result.failure + else None + ), + ) diff --git a/app/providers/registry.py b/app/providers/registry.py index 19a489a..1d71355 100644 --- a/app/providers/registry.py +++ b/app/providers/registry.py @@ -34,6 +34,13 @@ class ProviderRegistry: """ return self._by_code.get(provider_code) + def all(self) -> list[GenerationProvider]: + """全部已装配的适配器。 + + 供「按 media_kind 找适配器」这类查询用——按能力而非按 code 检索时用它。 + """ + return list(self._by_code.values()) + @property def codes(self) -> frozenset[str]: return frozenset(self._by_code) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..fdec95a --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[project] +name = "patbond-ai" +version = "0.1.0" +description = "Patbond AI 生成服务：调用 AI 服务商 API 完成图片与视频生成" +requires-python = ">=3.11" + +[tool.ruff] +# 与 CI 门禁一致（.gitea/workflows/ci.yml）。 +line-length = 88 + +[tool.ruff.lint.flake8-bugbear] +# `Depends()` / `Query()` 等放在参数默认值里**正是 FastAPI 的标准写法**， +# B008 在此是误报。为迎合规则把依赖声明搬进函数体会失去类型提示与自动文档， +# 是本末倒置——故按 ruff 提供的正规方式声明这些调用为可安全求值， +# 而不是在每个端点上撒 `# noqa`（后者迟早被用来压制真正的问题）。 +extend-immutable-calls = [ + "fastapi.Depends", + "fastapi.Query", + "fastapi.Path", + "fastapi.Header", + "fastapi.Body", +] + +[tool.pytest.ini_options] +# 让 `pytest` 从仓库根目录直接跑通，无需先安装包。 +pythonpath = ["."] +testpaths = ["tests"] diff --git a/requirements-dev.txt b/requirements-dev.txt index 2e437d2..5da938a 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -2,3 +2,4 @@ # 与 CI 门禁同一套工具（.gitea/workflows/ci.yml） ruff>=0.6 pytest>=8.0 +pyyaml>=6.0 diff --git a/requirements.txt b/requirements.txt index 33bb5e0..7a4b244 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,8 @@ # 运行时依赖。版本下限按实测可用版本给，不锁死上限——本服务只用 httpx 的 # 基础同步 API，其小版本变更风险低。 httpx>=0.27 +# HTTP 服务（内部接口）。uvicorn 是运行时服务器，非库依赖。 +fastapi>=0.115 +uvicorn>=0.30 +# 服务身份 JWT 的 RS256 校验；[crypto] 带入 cryptography。 +pyjwt[crypto]>=2.9 diff --git a/tests/api/__init__.py b/tests/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/api/test_contract_drift.py b/tests/api/test_contract_drift.py new file mode 100644 index 0000000..c75ce2c --- /dev/null +++ b/tests/api/test_contract_drift.py @@ -0,0 +1,177 @@ +"""手写契约 `openapi-internal.yaml` 与实际实现的一致性守卫。 + +**为什么需要它**：契约是手写的，实现是另一份代码。本项目开工审计已经证明「没有 +门禁的契约必然漂移」——四模块快照锁实际不存在，结果白名单 42/41 之误四份文档一路 +不可见，整整一个迭代没人发现。内部契约同样是契约，同样需要门禁。 + +**不做的事**：不比对 FastAPI 自动生成的 `/openapi.json` 与手写 yaml 的字节一致性。 +两者永远不会字节相同（描述、示例、`$ref` 组织方式都不同），逐字比对只会制造一个 +天天红的门禁，然后被人加 skip 绕过。这里比对的是**有实质意义的东西**：路径、方法、 +字段集、枚举取值。 +""" + +from __future__ import annotations + +import pathlib + +import yaml + +from app.api.errors import ServiceErrorCode +from app.api.main import create_app +from app.api.schemas import FailureModel, SubmitRequest, SubmitResponse, TaskResponse +from app.config import ServiceAuthSettings, Settings, VolcengineSettings +from app.domain.generation import Retry, TaskState, UserFacing + +CONTRACT = yaml.safe_load( + ( + pathlib.Path(__file__).resolve().parent.parent.parent / "openapi-internal.yaml" + ).read_text(encoding="utf-8") +) + + +def schema_properties(name: str) -> set[str]: + return set(CONTRACT["components"]["schemas"][name]["properties"]) + + +def app_routes() -> set[tuple[str, str]]: + """实际对外暴露的 `/internal` 端点集合。 + + 取自 `app.openapi()` 而非遍历 `app.routes`：后者在不同 FastAPI 版本里结构不同 + （本版本的 `include_router` 会包一层内部 wrapper，路由并不平铺在 `app.routes` + 上），依赖它会让这个守卫在升级框架时莫名失效——而一个悄悄失效的门禁比没有 + 门禁更糟。`openapi()` 是「究竟暴露了什么」的权威视图。 + """ + app = create_app( + Settings( + volcengine=VolcengineSettings(api_key="dummy-api-key-for-tests"), + service_auth=ServiceAuthSettings(public_key=""), + ) + ) + return { + (method.upper(), path) + for path, operations in app.openapi()["paths"].items() + if path.startswith("/internal") + for method in operations + } + + +class TestPathsMatch: + def test_every_contract_path_is_implemented(self): + implemented = app_routes() + for path, operations in CONTRACT["paths"].items(): + for method in operations: + assert (method.upper(), path) in implemented, ( + f"契约声明了 {method.upper()} {path} 但实现里没有" + ) + + def test_no_undeclared_internal_route_exists(self): + """实现里不得有契约之外的 /internal 端点。 + + 未声明的内部端点是最容易被遗忘的表面——它不在契约里，评审看不到， + 但它确实对外开放。 + """ + declared = { + (method.upper(), path) + for path, operations in CONTRACT["paths"].items() + for method in operations + } + assert app_routes() <= declared + + +class TestFieldSetsMatch: + """字段集漂移是最隐蔽的一种：多一个字段无人察觉，少一个字段调用方拿到 None。""" + + def test_submit_request(self): + assert schema_properties("SubmitRequest") == set(SubmitRequest.model_fields) + + def test_submit_response(self): + assert schema_properties("SubmitResponse") == set(SubmitResponse.model_fields) + + def test_task_response(self): + assert schema_properties("TaskResponse") == set(TaskResponse.model_fields) + + def test_failure(self): + assert schema_properties("Failure") == set(FailureModel.model_fields) + + def test_all_field_names_are_snake_case(self): + """契约定为 snake_case（与对客户端的 camelCase 契约不同，见契约说明）。 + + 混入一个 camelCase 字段就会让「JSON 与 dataclass 字段逐一对应、无需转换层」 + 这个前提失效。 + """ + for name in ("SubmitRequest", "SubmitResponse", "TaskResponse", "Failure"): + for field in schema_properties(name): + assert field.islower() or "_" in field, ( + f"{name}.{field} 不是 snake_case" + ) + assert not any(char.isupper() for char in field), ( + f"{name}.{field} 含大写字母" + ) + + +class TestEnumsMatch: + """枚举漂移的后果是调用方收到一个它没处理的取值——而那通常走到默认分支。""" + + def test_task_states(self): + declared = set( + CONTRACT["components"]["schemas"]["TaskResponse"]["properties"]["state"][ + "enum" + ] + ) + assert declared == {state.value for state in TaskState} + + def test_retry_strategies(self): + declared = set( + CONTRACT["components"]["schemas"]["Failure"]["properties"]["retry"]["enum"] + ) + assert declared == {retry.value for retry in Retry} + + def test_user_facing_categories(self): + declared = set( + CONTRACT["components"]["schemas"]["Failure"]["properties"]["user_facing"][ + "enum" + ] + ) + assert declared == {category.value for category in UserFacing} + + def test_service_error_codes(self): + declared = set( + CONTRACT["components"]["schemas"]["ServiceErrorEnvelope"]["properties"][ + "error" + ]["properties"]["code"]["enum"] + ) + assert declared == {code.code for code in ServiceErrorCode} + + def test_provider_error_reuses_the_failure_classification_fields(self): + """`provider` 错误信封与 `Failure` 的分类字段必须一致。 + + 这是刻意设计：调用方因此可以复用同一段处置逻辑，不必为「提交时被限流」 + 与「任务失败于限流」写两套分支。两边若漂移，那份复用就会悄悄失效。 + """ + envelope = set( + CONTRACT["components"]["schemas"]["ProviderErrorEnvelope"]["properties"][ + "error" + ]["properties"] + ) + failure = schema_properties("Failure") + assert envelope == failure | {"kind"} + + +class TestRequiredFieldsMatch: + def test_submit_request_required(self): + declared = set(CONTRACT["components"]["schemas"]["SubmitRequest"]["required"]) + actual = { + name + for name, field in SubmitRequest.model_fields.items() + if field.is_required() + } + assert declared == actual + + def test_task_response_required(self): + declared = set(CONTRACT["components"]["schemas"]["TaskResponse"]["required"]) + actual = { + name + for name, field in TaskResponse.model_fields.items() + if field.is_required() + } + assert declared == actual diff --git a/tests/api/test_internal_api.py b/tests/api/test_internal_api.py new file mode 100644 index 0000000..0ea57b4 --- /dev/null +++ b/tests/api/test_internal_api.py @@ -0,0 +1,512 @@ +"""内部接口的测试：鉴权边界、两类错误信封、端到端编排。 + +全部离线——服务商调用被 `httpx.MockTransport` 拦下，服务票用测试期临时生成的 +RSA 密钥对签发。 +""" + +from __future__ import annotations + +import httpx +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from fastapi.testclient import TestClient + +from app.api.main import create_app +from app.config import ServiceAuthSettings, Settings, VolcengineSettings +from app.providers.registry import ProviderRegistry +from app.providers.volcengine.client import ArkClient +from app.providers.volcengine.video import SeedanceVideoProvider + +FAKE_API_KEY = "dummy-api-key-for-tests" +MODEL_2_5 = "doubao-seedance-2-5-260628" +MODEL_2_0 = "doubao-seedance-2-0-260128" + + +# ----------------------------------------------------------------- 夹具 + + +@pytest.fixture(scope="module") +def keypair() -> tuple[str, str]: + """测试期临时 RSA 密钥对（私钥签票、公钥配给服务）。 + + 现场生成而非硬编码 PEM：密钥形态的 PEM 写进仓库会被防泄漏扫描拦下，而且 + 那条规则是对的。 + """ + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_pem = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + public_pem = ( + key.public_key() + .public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode() + ) + return private_pem, public_pem + + +def service_token(private_pem: str, **overrides) -> str: + """签一张服务票；`overrides` 用于构造各种不合格的票。""" + import time + + claims = { + "iss": "patbond-api", + "aud": "patbond-ai", + "sub": "service:creation-worker", + "typ": "service", + "iat": int(time.time()), + "exp": int(time.time()) + 300, + "jti": "test-jti", + } + claims.update(overrides) + return jwt.encode(claims, private_pem, algorithm="RS256") + + +def client_for(keypair, handler, *, with_provider: bool = True) -> TestClient: + """建一个测试客户端，服务商调用被 handler 拦下。""" + _, public_pem = keypair + settings = Settings( + volcengine=VolcengineSettings(api_key=FAKE_API_KEY if with_provider else ""), + service_auth=ServiceAuthSettings(public_key=public_pem), + ) + app = create_app(settings) + # 换掉装配好的适配器，让它的 HTTP 走 MockTransport。 + if with_provider: + app.state.registry = ProviderRegistry( + [ + SeedanceVideoProvider( + ArkClient( + api_key=FAKE_API_KEY, + base_url="https://ark.example.invalid/api/v3", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + ) + ] + ) + else: + app.state.registry = ProviderRegistry([]) + return TestClient(app, raise_server_exceptions=False) + + +def auth(keypair, **overrides) -> dict[str, str]: + private_pem, _ = keypair + return {"Authorization": f"Bearer {service_token(private_pem, **overrides)}"} + + +SUBMIT_BODY = { + "job_id": "0199a1b2-c3d4-7000-8000-000000000001", + "media_kind": "video", + "model": MODEL_2_5, + "prompt": "我的柯基在草地上奔跑", +} + + +# ------------------------------------------------------------- 鉴权边界 + + +class TestAuthentication: + """服务票是这一层的安全边界。失败一律返回同一个笼统原因，细节只进日志。""" + + def test_missing_token_is_rejected(self, keypair): + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post("/internal/generation/tasks", json=SUBMIT_BODY) + assert response.status_code == 401 + assert response.json()["error"]["kind"] == "service" + assert response.json()["error"]["code"] == "unauthenticated" + + def test_wrong_audience_is_rejected(self, keypair): + """`aud` 强制校验是独立密钥对方案唯一的意义所在。 + + Java 侧现有的 JwtVerifier 只验签名、不校验 aud——正因如此才必须用独立 + 密钥对；如果本服务也只验签名，那对独立密钥就白搭了。 + """ + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json=SUBMIT_BODY, + headers=auth(keypair, aud="someone-else"), + ) + assert response.status_code == 401 + + def test_wrong_issuer_is_rejected(self, keypair): + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json=SUBMIT_BODY, + headers=auth(keypair, iss="attacker"), + ) + assert response.status_code == 401 + + def test_a_user_token_shape_is_rejected_even_if_signed_with_this_key(self, keypair): + """有人误把用户密钥配到这里时的第二道防线。 + + 用户 token 没有 `typ=service`（Java 侧的 JwtSigner 也不签 aud），故即使 + 签名能过也会被拦下。 + """ + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json=SUBMIT_BODY, + headers=auth(keypair, typ=None), + ) + assert response.status_code == 401 + + def test_expired_token_is_rejected(self, keypair): + import time + + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json=SUBMIT_BODY, + headers=auth(keypair, exp=int(time.time()) - 10), + ) + assert response.status_code == 401 + + def test_missing_public_key_denies_everything_rather_than_allowing(self, keypair): + """配置缺失必须是「全拒」而非「全过」。 + + 前者立刻暴露，后者是一扇敞开的门。与 Java 侧 InternalAuthFilter 未配置 + internal-token 时全拒 /internal/** 的既有选择一致。 + """ + settings = Settings( + volcengine=VolcengineSettings(api_key=FAKE_API_KEY), + service_auth=ServiceAuthSettings(public_key=""), + ) + client = TestClient(create_app(settings), raise_server_exceptions=False) + response = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ) + assert response.status_code == 401 + + def test_error_body_never_says_why(self, keypair): + """具体原因（签名错/过期/aud 不符）只进日志。 + + 对方是我们自己的服务，真正原因在本服务日志里；把细节回给调用方只会在 + 将来某天变成攻击者的提示。 + """ + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + body = client.post("/internal/generation/tasks", json=SUBMIT_BODY).json() + message = body["error"]["message"] + for leak in ("signature", "aud", "audience", "expired", "iss"): + assert leak.lower() not in message.lower() + + +class TestHealthNeedsNoToken: + def test_health_is_outside_internal_and_unauthenticated(self, keypair): + client = client_for(keypair, lambda r: httpx.Response(200, json={})) + response = client.get("/health") + assert response.status_code == 200 + assert response.json()["status"] == "ok" + + def test_health_reports_assembled_providers(self, keypair): + """缺凭证时适配器不装配，而那个症状（任务永远排队）离根因很远。 + + 让探活直接说出来能省很多排查时间。 + """ + with_provider = client_for(keypair, lambda r: httpx.Response(200, json={})) + assert with_provider.get("/health").json()["providers"] == [ + "volcengine-seedance" + ] + + without = client_for( + keypair, lambda r: httpx.Response(200, json={}), with_provider=False + ) + assert without.get("/health").json()["providers"] == [] + + +# ------------------------------------------------------------------ 提交 + + +class TestSubmit: + def test_returns_provider_request_id(self, keypair): + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/contents/generations/tasks") + return httpx.Response(200, json={"id": "cgt-20260920-abc123"}) + + client = client_for(keypair, handler) + response = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ) + assert response.status_code == 200 + assert response.json() == {"provider_request_id": "cgt-20260920-abc123"} + + def test_field_names_are_snake_case(self, keypair): + """契约为 snake_case——camelCase 的请求体应被拒。 + + `extra="forbid"` 使拼错的字段立刻报错，而不是被静默忽略（后者表现为 + 「我传了 durationSec 却没生效」这类极难查的问题）。 + """ + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json={**SUBMIT_BODY, "durationSec": 5}, + headers=auth(keypair), + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == "invalid_request" + + def test_user_id_is_not_accepted(self, keypair): + """契约刻意不含 user_id（数据最小化）。传了应当被拒而不是被忽略。""" + client = client_for(keypair, lambda r: httpx.Response(200, json={"id": "x"})) + response = client.post( + "/internal/generation/tasks", + json={**SUBMIT_BODY, "user_id": "u-1"}, + headers=auth(keypair), + ) + assert response.status_code == 400 + + def test_empty_content_is_rejected_locally(self, keypair): + def handler(request): + raise AssertionError("不应发出请求") + + client = client_for(keypair, handler) + body = {k: v for k, v in SUBMIT_BODY.items() if k != "prompt"} + response = client.post( + "/internal/generation/tasks", json=body, headers=auth(keypair) + ) + assert response.status_code == 400 + assert "至少要有一个" in response.json()["error"]["message"] + + def test_model_limit_violation_is_invalid_request_not_provider_error(self, keypair): + """超出模型限制的请求**根本没发出去**，故属 service/invalid_request。 + + 归到 provider 类会让调用方以为是上游的问题而去退避重试——而重试同样的 + 参数永远不会变好。 + """ + + def handler(request): + raise AssertionError("不应发出请求") + + client = client_for(keypair, handler) + response = client.post( + "/internal/generation/tasks", + json={**SUBMIT_BODY, "model": MODEL_2_0, "duration_sec": 20}, + headers=auth(keypair), + ) + assert response.status_code == 400 + error = response.json()["error"] + assert error["kind"] == "service" + assert error["code"] == "invalid_request" + assert "[4, 15]" in error["message"] + + def test_no_adapter_yields_provider_unavailable(self, keypair): + """缺凭证 → 适配器未装配。503 而非 500：这是环境问题且可恢复。""" + client = client_for( + keypair, lambda r: httpx.Response(200, json={}), with_provider=False + ) + response = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ) + assert response.status_code == 503 + assert response.json()["error"]["code"] == "provider_unavailable" + + +# ------------------------------------------------------- 两类错误的分界 + + +class TestTwoErrorKindsAreDistinct: + """这是契约的核心设计：调用方对两类失败的反应完全不同。""" + + def test_vendor_rate_limit_is_a_provider_error_with_full_classification( + self, keypair + ): + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "code": "RateLimitExceeded.EndpointRPMExceeded", + "message": "RPM limit exceeded", + } + }, + ) + + client = client_for(keypair, handler) + response = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ) + # 刻意不透传服务商的 429：429 既可能是「等几秒」也可能是「等到明天」， + # 状态码表达不了这个区别，retry 才是权威信号。 + assert response.status_code == 502 + error = response.json()["error"] + assert error["kind"] == "provider" + assert error["retry"] == "backoff" + assert error["user_facing"] == "rate_limited" + assert error["provider_code"] == "RateLimitExceeded.EndpointRPMExceeded" + + def test_quota_exhausted_carries_a_different_retry_than_rate_limit(self, keypair): + """同为服务商 429，但一个等几秒、一个等到明天。 + + 调用方只看 retry 就能分开，不必解读厂商码。 + """ + + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "code": "QuotaExceeded", + "message": "Your account has exhausted its free trial quota.", + } + }, + ) + + client = client_for(keypair, handler) + error = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ).json()["error"] + assert error["retry"] == "quota" + assert error["user_facing"] == "quota_exhausted" + + def test_kind_is_the_discriminator_not_the_status_code(self, keypair): + """两类错误的状态码会撞，`kind` 不会。 + + 502 既可能是服务商挂了（provider），也可能是我们打错地址导致的解析失败 + （也归 provider，但 service 类错误同样可能出现 5xx）。判别必须看 kind。 + """ + + def handler(request): + return httpx.Response(502, text="bad gateway") + + client = client_for(keypair, handler) + body = client.post( + "/internal/generation/tasks", json=SUBMIT_BODY, headers=auth(keypair) + ).json() + assert body["error"]["kind"] == "provider" + assert "kind" in body["error"] + + +# ------------------------------------------------------------------ 查询 + + +#: 官方查询文档「返回格式」一节的原样示例（2.0 系列 720p + 16:9）。 +OFFICIAL_QUERY_PAYLOAD = { + "id": "cgt-1", + "model": MODEL_2_0, + "status": "succeeded", + "content": { + "video_url": "https://v/out.mp4", + "last_frame_url": "https://v/f.png", + }, + "usage": {"completion_tokens": 108900, "total_tokens": 108900}, + "resolution": "720p", + "ratio": "16:9", + "duration": 5, + "framespersecond": 24, +} + + +class TestGetTask: + def test_succeeded_task_is_fully_serialised(self, keypair): + client = client_for( + keypair, lambda r: httpx.Response(200, json=OFFICIAL_QUERY_PAYLOAD) + ) + body = client.get( + "/internal/generation/tasks/cgt-1", headers=auth(keypair) + ).json() + assert body["state"] == "succeeded" + assert body["progress"] == 100 + assert body["asset_url"] == "https://v/out.mp4" + assert body["cover_url"] == "https://v/f.png" + assert (body["width_px"], body["height_px"]) == (1280, 720) + assert body["duration_ms"] == 5000 + assert body["billing_tokens"] == 108900 + assert body["failure"] is None + + def test_asset_url_always_carries_its_expiry(self, keypair): + """服务商 URL 只有 24 小时有效。时效随 URL 一起给出，调用方就没有理由 + 把它当永久地址存进 external_url。""" + client = client_for( + keypair, lambda r: httpx.Response(200, json=OFFICIAL_QUERY_PAYLOAD) + ) + body = client.get( + "/internal/generation/tasks/cgt-1", headers=auth(keypair) + ).json() + assert body["asset_url_expires_in_sec"] == 86400 + + def test_a_failed_task_is_200_not_an_error_response(self, keypair): + """查询一个失败的任务，**HTTP 调用是成功的**，失败的是任务。 + + 做成非 2xx 会让「查不到任务」与「任务失败了」难以区分。 + """ + + def handler(request): + return httpx.Response( + 200, + json={ + "id": "cgt-1", + "status": "failed", + "content": {}, + "error": { + "code": "OutputVideoSensitiveContentDetected", + "message": "output may contain sensitive information", + }, + }, + ) + + client = client_for(keypair, handler) + response = client.get("/internal/generation/tasks/cgt-1", headers=auth(keypair)) + assert response.status_code == 200 + body = response.json() + assert body["state"] == "failed" + assert body["failure"]["retry"] == "never" + assert body["failure"]["user_facing"] == "content_rejected" + assert body["failure"]["provider_code"] == "OutputVideoSensitiveContentDetected" + + def test_vendor_404_becomes_task_not_found_so_the_caller_stops_polling( + self, keypair + ): + """服务商只保留 7 天记录。查不到时调用方要做的是**停止轮询**， + 不是退避重试——故归 service/task_not_found 而非 provider。""" + + def handler(request): + return httpx.Response( + 404, json={"error": {"code": "NotFound.", "message": "not found"}} + ) + + client = client_for(keypair, handler) + response = client.get( + "/internal/generation/tasks/cgt-gone", headers=auth(keypair) + ) + assert response.status_code == 404 + error = response.json()["error"] + assert error["kind"] == "service" + assert error["code"] == "task_not_found" + + def test_running_task_has_zero_progress(self, keypair): + # 服务商不给百分比，故成功前恒 0。这不是缺陷——契约本就不承诺平滑进度。 + def handler(request): + return httpx.Response( + 200, json={"id": "cgt-1", "status": "running", "content": {}} + ) + + body = ( + client_for(keypair, handler) + .get("/internal/generation/tasks/cgt-1", headers=auth(keypair)) + .json() + ) + assert body["state"] == "running" + assert body["progress"] == 0 + + def test_queued_maps_to_pending(self, keypair): + def handler(request): + return httpx.Response( + 200, json={"id": "cgt-1", "status": "queued", "content": {}} + ) + + body = ( + client_for(keypair, handler) + .get("/internal/generation/tasks/cgt-1", headers=auth(keypair)) + .json() + ) + assert body["state"] == "pending" + + def test_get_requires_a_token_too(self, keypair): + client = client_for(keypair, lambda r: httpx.Response(200, json={})) + assert client.get("/internal/generation/tasks/cgt-1").status_code == 401