张翔
225 lines
9.0 KiB
Python
225 lines
9.0 KiB
Python
import pytest
|
|
import yaml
|
|
import os
|
|
from pathlib import Path
|
|
from apitest.config.config_manager import ConfigManager
|
|
from apitest.models.exceptions import ConfigException
|
|
|
|
|
|
class TestConfigManager:
|
|
"""测试ConfigManager配置管理器"""
|
|
|
|
@pytest.fixture
|
|
def temp_config_file(self, tmp_path):
|
|
"""创建临时配置文件"""
|
|
config_data = {
|
|
"target": {
|
|
"base_url": "http://localhost:8080",
|
|
"timeout": 5000,
|
|
"max_retries": 3
|
|
},
|
|
"auth": {
|
|
"username": "test_user",
|
|
"password": "test_pass",
|
|
"login_endpoint": "/sys/auth/login"
|
|
},
|
|
"test": {
|
|
"data_dir": "data",
|
|
"test_cases_dir": "test_cases",
|
|
"parallel": True,
|
|
"parallel_threads": 4,
|
|
"retry_count": 2,
|
|
"stop_on_failure": False,
|
|
"max_response_time": 5000
|
|
},
|
|
"report": {
|
|
"output_dir": "reports",
|
|
"format": "html"
|
|
},
|
|
"logging": {
|
|
"level": "INFO",
|
|
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
"file": "logs/api_test.log"
|
|
}
|
|
}
|
|
|
|
config_file = tmp_path / "config.yaml"
|
|
with open(config_file, "w", encoding="utf-8") as f:
|
|
yaml.dump(config_data, f)
|
|
|
|
return config_file
|
|
|
|
def test_config_manager_initialization(self, temp_config_file):
|
|
"""测试配置管理器初始化"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.config_path == temp_config_file
|
|
assert config_manager._config is not None
|
|
|
|
def test_config_manager_nonexistent_file(self):
|
|
"""测试配置文件不存在"""
|
|
with pytest.raises(ConfigException, match="配置文件不存在"):
|
|
ConfigManager("/nonexistent/config.yaml")
|
|
|
|
def test_get_config_value(self, temp_config_file):
|
|
"""测试获取配置值"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
|
|
assert config_manager.get("target.base_url") == "http://localhost:8080"
|
|
assert config_manager.get("target.timeout") == 5000
|
|
assert config_manager.get("target.max_retries") == 3
|
|
|
|
def test_get_nested_config_value(self, temp_config_file):
|
|
"""测试获取嵌套配置值"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
|
|
assert config_manager.get("auth.username") == "test_user"
|
|
assert config_manager.get("auth.password") == "test_pass"
|
|
assert config_manager.get("auth.login_endpoint") == "/sys/auth/login"
|
|
|
|
def test_get_config_value_with_default(self, temp_config_file):
|
|
"""测试获取配置值（带默认值）"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
|
|
assert config_manager.get("nonexistent.key", "default_value") == "default_value"
|
|
assert config_manager.get("target.nonexistent", 0) == 0
|
|
|
|
def test_get_target_config(self, temp_config_file):
|
|
"""测试获取目标系统配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
target_config = config_manager.get_target_config()
|
|
|
|
assert target_config["base_url"] == "http://localhost:8080"
|
|
assert target_config["timeout"] == 5000
|
|
assert target_config["max_retries"] == 3
|
|
|
|
def test_get_auth_config(self, temp_config_file):
|
|
"""测试获取认证配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
auth_config = config_manager.get_auth_config()
|
|
|
|
assert auth_config["username"] == "test_user"
|
|
assert auth_config["password"] == "test_pass"
|
|
assert auth_config["login_endpoint"] == "/sys/auth/login"
|
|
|
|
def test_get_test_config(self, temp_config_file):
|
|
"""测试获取测试配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
test_config = config_manager.get_test_config()
|
|
|
|
assert test_config["data_dir"] == "data"
|
|
assert test_config["test_cases_dir"] == "test_cases"
|
|
assert test_config["parallel"] == True
|
|
assert test_config["parallel_threads"] == 4
|
|
|
|
def test_get_report_config(self, temp_config_file):
|
|
"""测试获取报告配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
report_config = config_manager.get_report_config()
|
|
|
|
assert report_config["output_dir"] == "reports"
|
|
assert report_config["format"] == "html"
|
|
|
|
def test_get_logging_config(self, temp_config_file):
|
|
"""测试获取日志配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
logging_config = config_manager.get_logging_config()
|
|
|
|
assert logging_config["level"] == "INFO"
|
|
assert logging_config["file"] == "logs/api_test.log"
|
|
|
|
def test_get_base_url(self, temp_config_file):
|
|
"""测试获取基础URL"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_base_url() == "http://localhost:8080"
|
|
|
|
def test_get_timeout(self, temp_config_file):
|
|
"""测试获取超时时间"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_timeout() == 5000
|
|
|
|
def test_get_max_retries(self, temp_config_file):
|
|
"""测试获取最大重试次数"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_max_retries() == 3
|
|
|
|
def test_get_login_endpoint(self, temp_config_file):
|
|
"""测试获取登录端点"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_login_endpoint() == "/sys/auth/login"
|
|
|
|
def test_is_parallel_enabled(self, temp_config_file):
|
|
"""测试是否启用并行执行"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.is_parallel_enabled() == True
|
|
|
|
def test_get_parallel_threads(self, temp_config_file):
|
|
"""测试获取并行线程数"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_parallel_threads() == 4
|
|
|
|
def test_get_retry_count(self, temp_config_file):
|
|
"""测试获取重试次数"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_retry_count() == 2
|
|
|
|
def test_should_stop_on_failure(self, temp_config_file):
|
|
"""测试是否在失败时停止"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.should_stop_on_failure() == False
|
|
|
|
def test_get_max_response_time(self, temp_config_file):
|
|
"""测试获取最大响应时间"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_max_response_time() == 5000
|
|
|
|
def test_get_report_format(self, temp_config_file):
|
|
"""测试获取报告格式"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_report_format() == "html"
|
|
|
|
def test_get_log_level(self, temp_config_file):
|
|
"""测试获取日志级别"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
assert config_manager.get_log_level() == "INFO"
|
|
|
|
def test_get_log_format(self, temp_config_file):
|
|
"""测试获取日志格式"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
expected_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
assert config_manager.get_log_format() == expected_format
|
|
|
|
def test_get_log_file(self, temp_config_file):
|
|
"""测试获取日志文件路径"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
log_file = config_manager.get_log_file()
|
|
assert log_file is not None
|
|
assert log_file.name == "api_test.log"
|
|
|
|
def test_get_auth_credentials_with_env(self, temp_config_file, monkeypatch):
|
|
"""测试获取认证凭据（环境变量）"""
|
|
monkeypatch.setenv("TEST_USERNAME", "env_user")
|
|
monkeypatch.setenv("TEST_PASSWORD", "env_pass")
|
|
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
credentials = config_manager.get_auth_credentials()
|
|
|
|
assert credentials["username"] == "env_user"
|
|
assert credentials["password"] == "env_pass"
|
|
|
|
def test_reload_config(self, temp_config_file):
|
|
"""测试重新加载配置"""
|
|
config_manager = ConfigManager(str(temp_config_file))
|
|
|
|
original_url = config_manager.get_base_url()
|
|
assert original_url == "http://localhost:8080"
|
|
|
|
with open(temp_config_file, "r+", encoding="utf-8") as f:
|
|
config_data = yaml.safe_load(f)
|
|
config_data["target"]["base_url"] = "http://new-url:8080"
|
|
f.seek(0)
|
|
yaml.dump(config_data, f)
|
|
f.truncate()
|
|
|
|
config_manager.reload()
|
|
assert config_manager.get_base_url() == "http://new-url:8080"
|