""" 依赖解析器单元测试 """ import pytest from typing import List from unittest.mock import Mock, MagicMock from app.models.risk_detection import DetectionRule, RiskLevel from app.services.risk_detection.engine.dependency_resolver import ( DependencyResolver, DependencyGraph, DependencyNode, ) class TestDependencyNode: """依赖节点测试""" def test_init(self): """测试节点初始化""" node = DependencyNode( rule_id="rule_1", algorithm_code="ALGO_1", dependencies=["rule_0"], dependents=["rule_2"] ) assert node.rule_id == "rule_1" assert node.algorithm_code == "ALGO_1" assert node.dependencies == ["rule_0"] assert node.dependents == ["rule_2"] assert node.level == 0 class TestDependencyGraph: """依赖图测试""" def setup_method(self): """每个测试方法前执行""" self.graph = DependencyGraph() def test_add_node(self): """测试添加节点""" self.graph.add_node("rule_1", "ALGO_1") assert "rule_1" in self.graph.nodes assert self.graph.nodes["rule_1"].rule_id == "rule_1" assert self.graph.nodes["rule_1"].algorithm_code == "ALGO_1" assert self.graph.graph["rule_1"] == [] assert self.graph.reverse_graph["rule_1"] == [] assert self.graph.in_degree["rule_1"] == 0 def test_add_edge(self): """测试添加依赖边""" # 添加节点 self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") # 添加边：rule_2 依赖 rule_1 self.graph.add_edge("rule_2", "rule_1") # 验证依赖关系 assert "rule_1" in self.graph.nodes["rule_2"].dependencies assert "rule_2" in self.graph.nodes["rule_1"].dependents # 验证入度 assert self.graph.in_degree["rule_1"] == 0 assert self.graph.in_degree["rule_2"] == 1 def test_add_edge_invalid_rule(self): """测试添加边时规则不存在""" self.graph.add_node("rule_1", "ALGO_1") with pytest.raises(ValueError, match="规则不存在"): self.graph.add_edge("rule_2", "rule_1") def test_has_cycle_no_cycle(self): """测试无循环依赖""" self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") self.graph.add_node("rule_3", "ALGO_3") # rule_2 依赖 rule_1, rule_3 依赖 rule_2 self.graph.add_edge("rule_2", "rule_1") self.graph.add_edge("rule_3", "rule_2") has_cycle, cycle_path = self.graph.has_cycle() assert not has_cycle assert cycle_path is None def test_has_cycle_with_cycle(self): """测试有循环依赖""" self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") self.graph.add_node("rule_3", "ALGO_3") # 创建循环：rule_1 -> rule_2 -> rule_3 -> rule_1 self.graph.add_edge("rule_2", "rule_1") self.graph.add_edge("rule_3", "rule_2") self.graph.add_edge("rule_1", "rule_3") has_cycle, cycle_path = self.graph.has_cycle() assert has_cycle assert cycle_path is not None assert len(cycle_path) >= 3 # 至少包含3个节点 def test_calculate_levels(self): """测试计算层级""" self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") self.graph.add_node("rule_3", "ALGO_3") # rule_2 依赖 rule_1, rule_3 依赖 rule_2 self.graph.add_edge("rule_2", "rule_1") self.graph.add_edge("rule_3", "rule_2") self.graph.calculate_levels() # 验证层级 assert self.graph.nodes["rule_1"].level == 0 assert self.graph.nodes["rule_2"].level == 1 assert self.graph.nodes["rule_3"].level == 2 def test_get_levels(self): """测试获取层级分组""" self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") self.graph.add_node("rule_3", "ALGO_3") # rule_2 依赖 rule_1, rule_3 依赖 rule_2 self.graph.add_edge("rule_2", "rule_1") self.graph.add_edge("rule_3", "rule_2") self.graph.calculate_levels() levels = self.graph.get_levels() # 验证层级结构 assert len(levels) == 3 # 3个层级 assert "rule_1" in levels[0] assert "rule_2" in levels[1] assert "rule_3" in levels[2] def test_to_dict(self): """测试转换为字典""" self.graph.add_node("rule_1", "ALGO_1") self.graph.add_node("rule_2", "ALGO_2") self.graph.add_edge("rule_2", "rule_1") self.graph.calculate_levels() result = self.graph.to_dict() assert "nodes" in result assert "levels" in result assert len(result["nodes"]) == 2 assert len(result["levels"]) >= 1 class TestDependencyResolver: """依赖解析器测试""" def setup_method(self): """每个测试方法前执行""" self.resolver = DependencyResolver() def _create_mock_rule(self, rule_id: str, algorithm_code: str, dependencies: List[str] = None) -> DetectionRule: """创建模拟规则""" rule = Mock(spec=DetectionRule) rule.rule_id = rule_id rule.algorithm_code = algorithm_code rule.parameters = {"dependencies": dependencies} if dependencies else None return rule def test_analyze_simple_dependencies(self): """测试简单依赖分析""" rules = [ self._create_mock_rule("rule_1", "ALGO_1"), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), ] graph = self.resolver.analyze(rules) assert graph is not None assert "rule_1" in graph.nodes assert "rule_2" in graph.nodes assert "rule_1" in graph.nodes["rule_2"].dependencies def test_analyze_with_cycle(self): """测试循环依赖检测""" rules = [ self._create_mock_rule("rule_1", "ALGO_1", ["rule_2"]), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), ] with pytest.raises(ValueError, match="检测到循环依赖"): self.resolver.analyze(rules) def test_analyze_with_default_dependencies(self): """测试使用默认依赖关系""" # 设置默认依赖 self.resolver.DEFAULT_ALGORITHM_DEPENDENCIES = { "ALGO_2": ["ALGO_1"], } rules = [ self._create_mock_rule("rule_1", "ALGO_1"), self._create_mock_rule("rule_2", "ALGO_2"), # 无显式依赖，使用默认 ] graph = self.resolver.analyze(rules, use_default_dependencies=True) assert "rule_1" in graph.nodes["rule_2"].dependencies def test_validate_valid_dependencies(self): """测试有效依赖验证""" rules = [ self._create_mock_rule("rule_1", "ALGO_1"), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), ] is_valid, error = self.resolver.validate(rules) assert is_valid assert error is None def test_validate_invalid_dependencies(self): """测试无效依赖验证""" rules = [ self._create_mock_rule("rule_1", "ALGO_1", ["rule_2"]), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), ] is_valid, error = self.resolver.validate(rules) assert not is_valid assert error is not None assert "循环依赖" in error def test_extract_dependencies_from_params(self): """测试从参数中提取依赖""" rule = self._create_mock_rule("rule_1", "ALGO_1", ["rule_2", "rule_3"]) rule_dict = {"ALGO_2": "rule_2", "ALGO_3": "rule_3"} deps = self.resolver._extract_dependencies(rule, rule_dict, False) assert "rule_2" in deps assert "rule_3" in deps def test_extract_dependencies_from_algorithm_code(self): """测试从算法代码提取依赖""" rule = self._create_mock_rule("rule_1", "ALGO_1") # 无参数依赖 rule_dict = {"ALGO_2": "rule_2"} # 设置默认依赖 self.resolver.DEFAULT_ALGORITHM_DEPENDENCIES = { "ALGO_1": ["ALGO_2"], } deps = self.resolver._extract_dependencies(rule, rule_dict, True) assert "rule_2" in deps def test_get_execution_order(self): """测试获取执行顺序""" rules = [ self._create_mock_rule("rule_1", "ALGO_1"), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), self._create_mock_rule("rule_3", "ALGO_3", ["rule_2"]), ] self.resolver.analyze(rules) execution_order = self.resolver.get_execution_order() assert len(execution_order) == 3 # 验证顺序：rule_1 -> rule_2 -> rule_3 assert execution_order.index("rule_1") < execution_order.index("rule_2") assert execution_order.index("rule_2") < execution_order.index("rule_3") def test_get_execution_levels(self): """测试获取执行层级""" rules = [ self._create_mock_rule("rule_1", "ALGO_1"), self._create_mock_rule("rule_2", "ALGO_2", ["rule_1"]), self._create_mock_rule("rule_3", "ALGO_3", ["rule_1"]), ] self.resolver.analyze(rules) levels = self.resolver.get_execution_levels() assert len(levels) == 2 # 两个层级 assert "rule_1" in levels[0] assert "rule_2" in levels[1] or "rule_3" in levels[1] def test_get_execution_order_without_analyze(self): """测试未分析依赖时获取执行顺序""" with pytest.raises(ValueError, match="尚未进行依赖分析"): self.resolver.get_execution_order() def test_get_execution_levels_without_analyze(self): """测试未分析依赖时获取执行层级""" with pytest.raises(ValueError, match="尚未进行依赖分析"): self.resolver.get_execution_levels()