Coverage for python/pyairflowtester/rules/dbt.py: 22%

1""" 

2dbt analysis rules. 

3""" 

4 

5from typing import Any, Dict, List 

6 

7 

8class BaseRule: 

9 """Base rule class.""" 

10 

11 def __init__(self): 

12 self.id = "" 

13 self.name = "" 

14 self.severity = "" 

15 self.category = "" 

16 self.execution_mode = "" 

17 

18 def evaluate(self, manifest: Dict[str, Any]) -> List[Dict[str, Any]]: 

19 """Evaluate rule against manifest.""" 

20 raise NotImplementedError 

21 

22 

23class MissingTestsRule(BaseRule): 

24 """Detect models without tests.""" 

25 

26 def __init__(self): 

27 super().__init__() 

28 self.id = "DBT001" 

29 self.name = "Missing Tests" 

30 self.severity = "high" 

31 self.category = "data_quality" 

32 self.execution_mode = "static" 

33 

34 def evaluate(self, manifest: Dict[str, Any]) -> List[Dict[str, Any]]: 

35 """Detect models without tests.""" 

36 violations = [] 

37 

38 # Extract models 

39 models = {} 

40 if "nodes" in manifest: 

41 for key, node in manifest["nodes"].items(): 

42 if key.startswith("model."): 

43 model_name = node.get("name", "") 

44 models[model_name] = node 

45 

46 # Check which models have tests 

47 tested_models = set() 

48 if "nodes" in manifest: 

49 for key, node in manifest["nodes"].items(): 

50 if key.startswith("test."): 

51 # Find which model this test belongs to 

52 if "attached_node" in node: 

53 tested_models.add(node["attached_node"]) 

54 elif "depends_on" in node and "nodes" in node["depends_on"]: 

55 for dep in node["depends_on"]["nodes"]: 

56 if dep.startswith("model."): 

57 tested_models.add(dep) 

58 

59 # Find untested models (excluding ephemeral and temp) 

60 for model_name, model in models.items(): 

61 materialization = model.get("config", {}).get("materialized", "table") 

62 is_public = model.get("config", {}).get("meta", {}).get("public", True) 

63 

64 if is_public and materialization not in ["ephemeral", "temporary"]: 

65 model_key = f"model.{model.get('package_name', '')}.{model_name}" 

66 if model_key not in tested_models: 

67 violations.append( 

68 { 

69 "rule_id": self.id, 

70 "severity": self.severity, 

71 "affected_resource": model_name, 

72 "message": f"Model '{model_name}' has no tests", 

73 "remediation": "Add tests for this model in schema.yml", 

74 } 

75 ) 

76 

77 return violations 

78 

79 

80class RedundantTestsRule(BaseRule): 

81 """Detect redundant test definitions.""" 

82 

83 def __init__(self): 

84 super().__init__() 

85 self.id = "DBT002" 

86 self.name = "Redundant Tests" 

87 self.severity = "low" 

88 self.category = "maintainability" 

89 self.execution_mode = "static" 

90 

91 def evaluate(self, manifest: Dict[str, Any]) -> List[Dict[str, Any]]: 

92 """Detect redundant tests.""" 

93 violations = [] 

94 

95 # Track test signatures (model + test_type + column) 

96 test_signatures = {} 

97 

98 if "nodes" in manifest: 

99 for key, node in manifest["nodes"].items(): 

100 if key.startswith("test."): 

101 test_name = node.get("name", "") 

102 test_type = node.get("test_metadata", {}).get("name", "") 

103 

104 # Build signature 

105 if "attached_node" in node: 

106 signature = f"{node['attached_node']}:{test_type}" 

107 else: 

108 signature = f"{test_name}:{test_type}" 

109 

110 if signature not in test_signatures: 

111 test_signatures[signature] = [] 

112 

113 test_signatures[signature].append(test_name) 

114 

115 # Find duplicates 

116 for signature, test_list in test_signatures.items(): 

117 if len(test_list) > 1: 

118 violations.append( 

119 { 

120 "rule_id": self.id, 

121 "severity": self.severity, 

122 "affected_resource": signature, 

123 "message": f"Redundant tests found: {', '.join(test_list[:2])}...", 

124 "remediation": "Consolidate duplicate tests", 

125 } 

126 ) 

127 

128 return violations 

129 

130 

131class UntestedModelRule(BaseRule): 

132 """Detect untested public models.""" 

133 

134 def __init__(self): 

135 super().__init__() 

136 self.id = "DBT003" 

137 self.name = "Untested Public Model" 

138 self.severity = "medium" 

139 self.category = "data_quality" 

140 self.execution_mode = "static" 

141 

142 def evaluate(self, manifest: Dict[str, Any]) -> List[Dict[str, Any]]: 

143 """Detect untested public models.""" 

144 violations = [] 

145 

146 if "nodes" not in manifest: 

147 return violations 

148 

149 # dbt's manifest schema has no "test_dependencies" field on model 

150 # nodes. Real test coverage has to be derived by cross-referencing 

151 # the manifest's "test." nodes and following attached_node / 

152 # depends_on.nodes back to the model(s) they cover (same approach 

153 # as MissingTestsRule). 

154 test_counts: Dict[str, int] = {} 

155 for key, node in manifest["nodes"].items(): 

156 if not key.startswith("test."): 

157 continue 

158 

159 attached_node = node.get("attached_node") 

160 if attached_node: 

161 test_counts[attached_node] = test_counts.get(attached_node, 0) + 1 

162 elif "depends_on" in node and "nodes" in node["depends_on"]: 

163 for dep in node["depends_on"]["nodes"]: 

164 if dep.startswith("model."): 

165 test_counts[dep] = test_counts.get(dep, 0) + 1 

166 

167 # Find high-importance models with no tests 

168 for key, node in manifest["nodes"].items(): 

169 if not key.startswith("model."): 

170 continue 

171 

172 model_name = node.get("name", "") 

173 description = node.get("description", "") 

174 test_count = test_counts.get(key, 0) 

175 

176 # Check if model is marked as critical 

177 is_critical = ( 

178 "critical" in description.lower() 

179 or test_count == 0 

180 and len(node.get("columns", {})) > 5 

181 ) 

182 

183 if is_critical and test_count == 0: 

184 violations.append( 

185 { 

186 "rule_id": self.id, 

187 "severity": self.severity, 

188 "affected_resource": model_name, 

189 "message": f"Critical model '{model_name}' has no tests", 

190 "remediation": "Add comprehensive tests for this critical model", 

191 } 

192 ) 

193 

194 return violations