- 新增 model_deploy 与 model_operations 双库查询,支持模型列表、模型详情、单月监控结果和运维工作台真实接口。 - 关闭运维模块生产 Mock 数据,补充缺表/空数据降级、工作台空状态和首条纵向链路测试。 - 按业务团队、模型团队、管理员权限矩阵接入菜单、页面、操作按钮和后端接口权限校验,支持 business_team 角色。 - 更新工作台布局、深色欢迎卡片、全宽页面适配、顶部回退,以及权限分组展示。 - 新增架构实现基线、周目标完成情况和角色权限矩阵初始化 SQL 文档。
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import ProgrammingError
|
|
|
|
from backend.api.operations._deps import OperationsContext, require_operations_permission
|
|
from backend.services.operations.deploy_model_queries import (
|
|
_load_deploy_rows,
|
|
_load_monitor_snapshots,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_deploy_table_degrades_to_empty_models() -> None:
|
|
class MissingTableSession:
|
|
async def execute(self, *args, **kwargs):
|
|
raise ProgrammingError(
|
|
"SELECT ...",
|
|
{},
|
|
Exception(1146, "table does not exist"),
|
|
)
|
|
|
|
rows = await _load_deploy_rows(
|
|
MissingTableSession(),
|
|
workspace_code="001",
|
|
workspace_name="001",
|
|
)
|
|
|
|
assert rows == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_monitoring_table_degrades_to_empty_snapshots() -> None:
|
|
class MissingTableSession:
|
|
async def execute(self, *args, **kwargs):
|
|
raise ProgrammingError(
|
|
"SELECT ...",
|
|
{},
|
|
Exception(1146, "table does not exist"),
|
|
)
|
|
|
|
snapshots = await _load_monitor_snapshots(MissingTableSession(), "W" * 26)
|
|
|
|
assert snapshots == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_operations_permission_dependency_checks_effective_permissions() -> None:
|
|
allowed = OperationsContext(
|
|
request_id="R" * 26,
|
|
user_id="U" * 26,
|
|
workspace_id="W" * 26,
|
|
workspace_code="001",
|
|
workspace_name="001",
|
|
role="business_team",
|
|
permissions=frozenset({"operations:model-overview:view"}),
|
|
)
|
|
denied = OperationsContext(
|
|
request_id="R" * 26,
|
|
user_id="U" * 26,
|
|
workspace_id="W" * 26,
|
|
workspace_code="001",
|
|
workspace_name="001",
|
|
role="business_team",
|
|
permissions=frozenset(),
|
|
)
|
|
|
|
dependency = require_operations_permission("operations:model-overview:view")
|
|
assert await dependency(allowed) is allowed
|
|
with pytest.raises(HTTPException) as error:
|
|
await dependency(denied)
|
|
assert error.value.status_code == 403
|