285 lines
9.5 KiB
Python
285 lines
9.5 KiB
Python
|
"""ODF 光纤配线单元管理系统 - 基础功能测试。"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
|
||
|
|
from app.main import app
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def client():
|
||
|
|
"""创建测试客户端。"""
|
||
|
|
return TestClient(app)
|
||
|
|
|
||
|
|
|
||
|
|
class TestRackAPI:
|
||
|
|
"""机架 API 测试。"""
|
||
|
|
|
||
|
|
def test_list_racks(self, client):
|
||
|
|
"""测试获取机架列表。"""
|
||
|
|
resp = client.get("/api/racks")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert isinstance(data, list)
|
||
|
|
assert len(data) >= 1
|
||
|
|
# 验证初始假数据中的机架
|
||
|
|
rack_names = [r["rack_name"] for r in data]
|
||
|
|
assert "ODF-001" in rack_names
|
||
|
|
|
||
|
|
def test_get_rack_detail(self, client):
|
||
|
|
"""测试获取机架详情。"""
|
||
|
|
resp = client.get("/api/racks/rack-001")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["rack_id"] == "rack-001"
|
||
|
|
assert data["rack_name"] == "ODF-001"
|
||
|
|
assert len(data["units"]) >= 1
|
||
|
|
|
||
|
|
def test_get_rack_not_found(self, client):
|
||
|
|
"""测试获取不存在的机架。"""
|
||
|
|
resp = client.get("/api/racks/not-exist")
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
def test_create_rack(self, client):
|
||
|
|
"""测试创建新机架。"""
|
||
|
|
resp = client.post("/api/racks", json={
|
||
|
|
"rack_name": "ODF-TEST-001",
|
||
|
|
"location": "测试位置",
|
||
|
|
})
|
||
|
|
assert resp.status_code == 201
|
||
|
|
data = resp.json()
|
||
|
|
assert data["rack_name"] == "ODF-TEST-001"
|
||
|
|
assert data["location"] == "测试位置"
|
||
|
|
assert "rack_id" in data
|
||
|
|
|
||
|
|
def test_delete_rack(self, client):
|
||
|
|
"""测试删除机架。"""
|
||
|
|
# 先创建一个新机架用于删除测试
|
||
|
|
create_resp = client.post("/api/racks", json={
|
||
|
|
"rack_name": "ODF-DELETE",
|
||
|
|
"location": "待删除",
|
||
|
|
})
|
||
|
|
rack_id = create_resp.json()["rack_id"]
|
||
|
|
|
||
|
|
resp = client.delete(f"/api/racks/{rack_id}")
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
# 确认已删除
|
||
|
|
get_resp = client.get(f"/api/racks/{rack_id}")
|
||
|
|
assert get_resp.status_code == 404
|
||
|
|
|
||
|
|
def test_delete_rack_not_found(self, client):
|
||
|
|
"""测试删除不存在的机架。"""
|
||
|
|
resp = client.delete("/api/racks/not-exist")
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
class TestUnitAPI:
|
||
|
|
"""配线单元 API 测试。"""
|
||
|
|
|
||
|
|
def test_create_unit(self, client):
|
||
|
|
"""测试创建配线单元。"""
|
||
|
|
resp = client.post("/api/racks/rack-001/units", json={
|
||
|
|
"unit_number": 10,
|
||
|
|
"unit_name": "测试单元",
|
||
|
|
"position": "左侧",
|
||
|
|
"port_count": 8,
|
||
|
|
"port_type": "LC",
|
||
|
|
})
|
||
|
|
assert resp.status_code == 201
|
||
|
|
data = resp.json()
|
||
|
|
assert data["unit_number"] == 10
|
||
|
|
assert data["unit_name"] == "测试单元"
|
||
|
|
assert len(data["ports"]) == 8
|
||
|
|
|
||
|
|
def test_create_unit_rack_not_found(self, client):
|
||
|
|
"""测试在不存在机架上创建单元。"""
|
||
|
|
resp = client.post("/api/racks/not-exist/units", json={
|
||
|
|
"unit_number": 1,
|
||
|
|
"port_count": 4,
|
||
|
|
"port_type": "SC",
|
||
|
|
})
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
def test_delete_unit(self, client):
|
||
|
|
"""测试删除配线单元。"""
|
||
|
|
# 先创建一个单元
|
||
|
|
create_resp = client.post("/api/racks/rack-001/units", json={
|
||
|
|
"unit_number": 20,
|
||
|
|
"unit_name": "待删除单元",
|
||
|
|
"port_count": 4,
|
||
|
|
})
|
||
|
|
unit_id = create_resp.json()["unit_id"]
|
||
|
|
|
||
|
|
resp = client.delete(f"/api/units/{unit_id}")
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
|
||
|
|
class TestPortAPI:
|
||
|
|
"""端口 API 测试。"""
|
||
|
|
|
||
|
|
def test_get_free_ports(self, client):
|
||
|
|
"""测试查询空闲端口。"""
|
||
|
|
resp = client.get("/api/ports/free")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert isinstance(data, list)
|
||
|
|
# 至少有空闲端口
|
||
|
|
assert len(data) >= 1
|
||
|
|
for p in data:
|
||
|
|
assert "port_id" in p
|
||
|
|
assert "port_number" in p
|
||
|
|
|
||
|
|
def test_get_free_ports_by_rack(self, client):
|
||
|
|
"""测试按机架过滤空闲端口。"""
|
||
|
|
resp = client.get("/api/ports/free?rack_id=rack-001")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
for p in data:
|
||
|
|
assert p["rack_id"] == "rack-001"
|
||
|
|
|
||
|
|
def test_get_port_detail(self, client):
|
||
|
|
"""测试获取端口详情。"""
|
||
|
|
# 先获取一个空闲端口
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if free_ports:
|
||
|
|
port_id = free_ports[0]["port_id"]
|
||
|
|
resp = client.get(f"/api/ports/{port_id}")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["port_id"] == port_id
|
||
|
|
assert "rack_name" in data
|
||
|
|
assert "unit_number" in data
|
||
|
|
|
||
|
|
def test_get_port_detail_not_found(self, client):
|
||
|
|
"""测试获取不存在端口的详情。"""
|
||
|
|
resp = client.get("/api/ports/not-exist")
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
class TestConnectionAPI:
|
||
|
|
"""跳接连接 API 测试。"""
|
||
|
|
|
||
|
|
def test_list_connections(self, client):
|
||
|
|
"""测试获取连接列表。"""
|
||
|
|
resp = client.get("/api/connections")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert isinstance(data, list)
|
||
|
|
|
||
|
|
def test_create_connection(self, client):
|
||
|
|
"""测试创建跳接连接。"""
|
||
|
|
# 获取两个空闲端口
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if len(free_ports) >= 2:
|
||
|
|
pa = free_ports[0]["port_id"]
|
||
|
|
pb = free_ports[1]["port_id"]
|
||
|
|
resp = client.post("/api/connections", json={
|
||
|
|
"port_a_id": pa,
|
||
|
|
"port_b_id": pb,
|
||
|
|
"fiber_length": 5.0,
|
||
|
|
"remark": "测试连接",
|
||
|
|
})
|
||
|
|
assert resp.status_code == 201
|
||
|
|
data = resp.json()
|
||
|
|
assert data["port_a_id"] == pa
|
||
|
|
assert data["port_b_id"] == pb
|
||
|
|
|
||
|
|
def test_create_connection_same_port(self, client):
|
||
|
|
"""测试同一端口自连接（应失败）。"""
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if free_ports:
|
||
|
|
port_id = free_ports[0]["port_id"]
|
||
|
|
resp = client.post("/api/connections", json={
|
||
|
|
"port_a_id": port_id,
|
||
|
|
"port_b_id": port_id,
|
||
|
|
})
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
def test_create_connection_busy_port(self, client):
|
||
|
|
"""测试连接到已使用端口（应失败）。"""
|
||
|
|
# 使用已存在的示例连接中的端口
|
||
|
|
resp = client.get("/api/connections")
|
||
|
|
conns = resp.json()
|
||
|
|
if conns:
|
||
|
|
busy_port = conns[0]["port_a_id"]
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if free_ports:
|
||
|
|
free_port = free_ports[0]["port_id"]
|
||
|
|
resp2 = client.post("/api/connections", json={
|
||
|
|
"port_a_id": busy_port,
|
||
|
|
"port_b_id": free_port,
|
||
|
|
})
|
||
|
|
assert resp2.status_code == 400
|
||
|
|
|
||
|
|
def test_delete_connection(self, client):
|
||
|
|
"""测试删除跳接连接。"""
|
||
|
|
# 先创建一个新连接
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if len(free_ports) >= 2:
|
||
|
|
pa = free_ports[0]["port_id"]
|
||
|
|
pb = free_ports[1]["port_id"]
|
||
|
|
create_resp = client.post("/api/connections", json={
|
||
|
|
"port_a_id": pa,
|
||
|
|
"port_b_id": pb,
|
||
|
|
})
|
||
|
|
conn_id = create_resp.json()["connection_id"]
|
||
|
|
|
||
|
|
del_resp = client.delete(f"/api/connections/{conn_id}")
|
||
|
|
assert del_resp.status_code == 204
|
||
|
|
|
||
|
|
# 确认已删除
|
||
|
|
get_resp = client.get("/api/connections")
|
||
|
|
ids = [c["connection_id"] for c in get_resp.json()]
|
||
|
|
assert conn_id not in ids
|
||
|
|
|
||
|
|
|
||
|
|
class TestPortPathAPI:
|
||
|
|
"""端口路径查询 API 测试。"""
|
||
|
|
|
||
|
|
def test_get_port_path(self, client):
|
||
|
|
"""测试查询端口路径。"""
|
||
|
|
# 使用示例连接中的端口
|
||
|
|
resp = client.get("/api/connections")
|
||
|
|
conns = resp.json()
|
||
|
|
if conns:
|
||
|
|
port_id = conns[0]["port_a_id"]
|
||
|
|
path_resp = client.get(f"/api/ports/{port_id}/path")
|
||
|
|
assert path_resp.status_code == 200
|
||
|
|
data = path_resp.json()
|
||
|
|
assert "port" in data
|
||
|
|
assert "connection" in data
|
||
|
|
assert data["connection"] is not None
|
||
|
|
|
||
|
|
def test_get_port_path_free_port(self, client):
|
||
|
|
"""测试查询空闲端口的路径。"""
|
||
|
|
free_resp = client.get("/api/ports/free")
|
||
|
|
free_ports = free_resp.json()
|
||
|
|
if free_ports:
|
||
|
|
port_id = free_ports[0]["port_id"]
|
||
|
|
resp = client.get(f"/api/ports/{port_id}/path")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["connection"] is None
|
||
|
|
|
||
|
|
def test_get_port_path_not_found(self, client):
|
||
|
|
"""测试查询不存在端口的路径。"""
|
||
|
|
resp = client.get("/api/ports/not-exist/path")
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
class TestHealthAPI:
|
||
|
|
"""健康检查 API 测试。"""
|
||
|
|
|
||
|
|
def test_health_check(self, client):
|
||
|
|
"""测试健康检查接口。"""
|
||
|
|
resp = client.get("/api/health")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["status"] == "ok"
|