zhaoxi
d84212f780
正式发布 Pretor 平台的首个 alpha 版本。本项目旨在构建一个基于分布式架构的多智能体协同工作流水线。 核心功能实现： 1. 建立基于 BaseIndividual 的动态插件加载机制。 2. 实现三类核心 worker_individual 子个体。 3. 集成 Ray 框架支持分布式集群调度。 4. 基于 PostgreSQL 的全量持久化存储方案。 5. 提供完整的 FastAPI 后端与 React 前端交互界面。
96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
# Mock dependencies before importing the module under test
|
|
class MockHTTPException(Exception):
|
|
def __init__(self, status_code, detail=None, headers=None):
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
self.headers = headers
|
|
|
|
|
|
class MockValidationError(Exception):
|
|
pass
|
|
|
|
|
|
mock_fastapi = MagicMock()
|
|
mock_fastapi.HTTPException = MockHTTPException
|
|
mock_fastapi.status.HTTP_401_UNAUTHORIZED = 401
|
|
|
|
mock_pydantic = MagicMock()
|
|
mock_pydantic.ValidationError = MockValidationError
|
|
|
|
sys.modules["fastapi"] = mock_fastapi
|
|
sys.modules["pydantic"] = mock_pydantic
|
|
sys.modules["sqlmodel"] = MagicMock()
|
|
sys.modules["passlib"] = MagicMock()
|
|
sys.modules["passlib.context"] = MagicMock()
|
|
sys.modules["pretor.core.database.table.user"] = MagicMock()
|
|
|
|
import pytest
|
|
import jwt
|
|
from pretor.utils.access import Accessor
|
|
|
|
|
|
def test_decode_token_success():
|
|
"""Test successful token decoding."""
|
|
token = "valid.token.here"
|
|
payload = {"user_id": "123", "username": "testuser", "exp": 1234567890}
|
|
|
|
with patch("jwt.decode", return_value=payload) as mock_decode:
|
|
with patch("pretor.utils.access.TokenData") as mock_token_data_cls:
|
|
mock_token_data_instance = MagicMock()
|
|
mock_token_data_cls.return_value = mock_token_data_instance
|
|
|
|
result = Accessor._decode_token(token)
|
|
|
|
mock_decode.assert_called_once()
|
|
mock_token_data_cls.assert_called_once_with(**payload)
|
|
assert result == mock_token_data_instance
|
|
|
|
|
|
def test_decode_token_expired():
|
|
"""Test token decoding with an expired token."""
|
|
token = "expired.token.here"
|
|
|
|
from fastapi import HTTPException
|
|
with patch("jwt.decode", side_effect=jwt.ExpiredSignatureError):
|
|
with patch("pretor.utils.access.HTTPException", HTTPException):
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
Accessor._decode_token(token)
|
|
|
|
assert excinfo.value.status_code == 401
|
|
assert excinfo.value.detail == "Token 已过期"
|
|
|
|
|
|
def test_decode_token_invalid():
|
|
"""Test token decoding with an invalid token."""
|
|
token = "invalid.token.here"
|
|
|
|
from fastapi import HTTPException
|
|
with patch("jwt.decode", side_effect=jwt.InvalidTokenError):
|
|
with patch("pretor.utils.access.HTTPException", HTTPException):
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
Accessor._decode_token(token)
|
|
|
|
assert excinfo.value.status_code == 401
|
|
assert excinfo.value.detail == "无效的认证凭证"
|
|
|
|
|
|
def test_decode_token_validation_error():
|
|
"""Test token decoding with a payload that fails validation."""
|
|
token = "valid.jwt.invalid.payload"
|
|
payload = {"wrong": "payload"}
|
|
|
|
from fastapi import HTTPException
|
|
|
|
with patch("jwt.decode", return_value=payload):
|
|
with patch("pretor.utils.access.TokenData", side_effect=MockValidationError):
|
|
with patch("pretor.utils.access.ValidationError", MockValidationError):
|
|
with patch("pretor.utils.access.HTTPException", HTTPException):
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
Accessor._decode_token(token)
|
|
|
|
assert excinfo.value.status_code == 401
|
|
assert excinfo.value.detail == "无效的认证凭证" |