Roog (顾新培)
31af5e2286
新增内容： - 创建基础项目结构。 - 添加 `.gitignore` 和 `.dockerignore` 文件。 - 编写 `pyproject.toml` 和依赖文件。 - 添加算法模块及示例算法。 - 实现核心功能模块（日志、错误处理、指标）。 - 添加开发和运行所需的相关脚本文件及文档。
111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""API 集成测试"""
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
|
|
|
|
class TestInvokeEndpoint:
|
|
"""测试 /invoke 端点"""
|
|
|
|
def test_invoke_prime_number(self, client):
|
|
"""测试质数判断"""
|
|
response = client.post("/invoke", json={"number": 17})
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
|
|
assert "request_id" in data
|
|
assert data["status"] == "success"
|
|
assert data["result"]["number"] == 17
|
|
assert data["result"]["is_prime"] is True
|
|
assert data["result"]["factors"] == []
|
|
|
|
def test_invoke_composite_number(self, client):
|
|
"""测试合数判断"""
|
|
response = client.post("/invoke", json={"number": 12})
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
|
|
assert data["status"] == "success"
|
|
assert data["result"]["number"] == 12
|
|
assert data["result"]["is_prime"] is False
|
|
assert len(data["result"]["factors"]) > 0
|
|
|
|
def test_invoke_edge_cases(self, client):
|
|
"""测试边界情况"""
|
|
# 测试 0
|
|
response = client.post("/invoke", json={"number": 0})
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["result"]["is_prime"] is False
|
|
|
|
# 测试 1
|
|
response = client.post("/invoke", json={"number": 1})
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["result"]["is_prime"] is False
|
|
|
|
# 测试 2
|
|
response = client.post("/invoke", json={"number": 2})
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["result"]["is_prime"] is True
|
|
|
|
def test_invoke_invalid_input(self, client):
|
|
"""测试无效输入"""
|
|
# 缺少必需字段
|
|
response = client.post("/invoke", json={})
|
|
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
|
|
# 错误的数据类型
|
|
response = client.post("/invoke", json={"number": "not a number"})
|
|
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
|
|
# 浮点数
|
|
response = client.post("/invoke", json={"number": 3.14})
|
|
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
|
|
|
|
class TestHealthEndpoints:
|
|
"""测试健康检查端点"""
|
|
|
|
def test_healthz(self, client):
|
|
"""测试存活检查"""
|
|
response = client.get("/healthz")
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
|
|
assert data["status"] == "healthy"
|
|
assert "timestamp" in data
|
|
|
|
def test_readyz(self, client):
|
|
"""测试就绪检查"""
|
|
response = client.get("/readyz")
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
|
|
assert data["status"] == "ready"
|
|
assert "timestamp" in data
|
|
assert "checks" in data
|
|
|
|
|
|
class TestMetricsEndpoint:
|
|
"""测试指标端点"""
|
|
|
|
def test_metrics(self, client):
|
|
"""测试 Prometheus 指标"""
|
|
response = client.get("/metrics")
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert "text/plain" in response.headers["content-type"]
|
|
|
|
|
|
class TestJobsEndpoint:
|
|
"""测试异步任务端点"""
|
|
|
|
def test_jobs_not_implemented(self, client):
|
|
"""测试异步任务接口（未实现）"""
|
|
response = client.post("/jobs", json={"number": 17})
|
|
|
|
assert response.status_code == status.HTTP_501_NOT_IMPLEMENTED
|