Coverage for python/pyairflowtester/dependency_intelligence/analytics.py: 88%
205 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 20:43 +0530
1"""Analytics engines for dependency intelligence (Phase 2: Weeks 5-8)."""
3import logging
4from dataclasses import dataclass
5from datetime import datetime
6from typing import Dict, List, Optional, Set, Tuple
8from .graph import DependencyGraphEngine
9from .models import (
10 DependencyGraph,
11 NodeSeverity,
12 NodeType,
13)
15logger = logging.getLogger(__name__)
18@dataclass
19class OwnershipAnalysis:
20 """Ownership and team impact analysis."""
22 owner: str
23 owned_nodes: List[str]
24 downstream_impact: int
25 critical_dependencies: List[str]
26 team_risk_score: float
27 affected_teams: Set[str]
28 cross_team_edges: int
31@dataclass
32class SchemaEvolution:
33 """Schema change tracking."""
35 node_id: str
36 changed_at: datetime
37 change_type: str # added_column, removed_column, type_change, renamed
38 old_schema: Dict
39 new_schema: Dict
40 affected_downstream: List[str]
43@dataclass
44class SLAValidation:
45 """SLA compliance tracking."""
47 node_id: str
48 has_sla: bool
49 sla_target: Optional[str]
50 actual_performance: Optional[str]
51 compliance_status: str # compliant, at_risk, violated
52 severity: NodeSeverity
55@dataclass
56class TestCoverageAnalysis:
57 """Test coverage metrics."""
59 node_id: str
60 total_tests: int
61 test_types: Dict[str, int] # unit, integration, end_to_end
62 coverage_percentage: float
63 coverage_status: str # good, adequate, poor
64 missing_tests: List[str]
67class OwnershipAnalyzer:
68 """Analyze ownership and team impact."""
70 def __init__(self, graph: DependencyGraph):
71 self.graph = graph
72 self.engine = DependencyGraphEngine(graph)
74 def analyze_owner(self, owner: str) -> OwnershipAnalysis:
75 """
76 Analyze impact and dependencies for an owner/team.
78 Returns ownership analysis with risk metrics.
79 """
80 # Get all nodes owned by this owner
81 owned_nodes = [n.id for n in self.graph.nodes.values() if n.owner == owner]
83 # Calculate downstream impact
84 all_downstream = set()
85 for node_id in owned_nodes:
86 all_downstream.update(self.engine.get_downstream_nodes(node_id))
88 # Get critical upstream dependencies
89 critical_deps = []
90 for node_id in owned_nodes:
91 upstream = self.engine.get_upstream_nodes(node_id)
92 for up_id in upstream:
93 node = self.graph.nodes.get(up_id)
94 if node and node.severity == NodeSeverity.CRITICAL:
95 critical_deps.append(up_id)
97 # Get all affected teams
98 affected_teams = set()
99 for node_id in all_downstream:
100 node = self.graph.nodes.get(node_id)
101 if node and node.owner != owner:
102 affected_teams.add(node.owner)
104 # Count cross-team edges
105 cross_team_edges = 0
106 for edge in self.graph.edges:
107 source = self.graph.nodes.get(edge.source)
108 target = self.graph.nodes.get(edge.target)
109 if source and target and source.owner != target.owner:
110 if source.owner == owner or target.owner == owner:
111 cross_team_edges += 1
113 # Calculate team risk score
114 team_risk = self._calculate_team_risk(
115 len(owned_nodes), len(all_downstream), len(critical_deps), len(affected_teams)
116 )
118 return OwnershipAnalysis(
119 owner=owner,
120 owned_nodes=owned_nodes,
121 downstream_impact=len(all_downstream),
122 critical_dependencies=list(set(critical_deps)),
123 team_risk_score=team_risk,
124 affected_teams=affected_teams,
125 cross_team_edges=cross_team_edges,
126 )
128 def _calculate_team_risk(self, owned: int, downstream: int, critical: int, teams: int) -> float:
129 """Calculate team risk score (0-10)."""
130 score = 0.0
132 # Owned node count factor
133 if owned > 100: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 score += 2.0
135 elif owned > 50: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 score += 1.0
138 # Downstream impact factor
139 if downstream > 200: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 score += 3.0
141 elif downstream > 100: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 score += 2.0
143 elif downstream > 50: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 score += 1.0
146 # Critical dependency factor
147 if critical > 5: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 score += 2.0
149 elif critical > 0: 149 ↛ 153line 149 didn't jump to line 153 because the condition on line 149 was always true
150 score += 1.0
152 # Team coordination complexity
153 score += min(3.0, teams / 5.0)
155 return min(10.0, score)
157 def analyze_all_owners(self) -> Dict[str, OwnershipAnalysis]:
158 """Analyze all owners/teams."""
159 owners = set(n.owner for n in self.graph.nodes.values() if n.owner)
160 return {owner: self.analyze_owner(owner) for owner in owners}
162 def find_critical_ownership_gaps(self) -> List[Tuple[str, str]]:
163 """Find nodes with no owner."""
164 unowned = [(n.id, n.name) for n in self.graph.nodes.values() if not n.owner]
165 return unowned
168class SchemaEvolutionTracker:
169 """Track schema changes and impact."""
171 def __init__(self, graph: DependencyGraph):
172 self.graph = graph
173 self.engine = DependencyGraphEngine(graph)
174 self.evolution_history: List[SchemaEvolution] = []
176 def add_schema_change(
177 self, node_id: str, change_type: str, old_schema: Dict, new_schema: Dict
178 ) -> SchemaEvolution:
179 """Record a schema change."""
180 downstream = self.engine.get_downstream_nodes(node_id)
182 evolution = SchemaEvolution(
183 node_id=node_id,
184 changed_at=datetime.utcnow(),
185 change_type=change_type,
186 old_schema=old_schema,
187 new_schema=new_schema,
188 affected_downstream=downstream,
189 )
191 self.evolution_history.append(evolution)
192 return evolution
194 def detect_breaking_changes(self) -> List[SchemaEvolution]:
195 """Identify potentially breaking schema changes."""
196 breaking = []
198 for evolution in self.evolution_history:
199 # Check for removed fields
200 old_fields = set(evolution.old_schema.get("fields", []))
201 new_fields = set(evolution.new_schema.get("fields", []))
202 removed = old_fields - new_fields
204 if removed and evolution.affected_downstream: 204 ↛ 208line 204 didn't jump to line 208 because the condition on line 204 was always true
205 breaking.append(evolution)
207 # Check for type changes
208 for field in old_fields & new_fields:
209 old_type = evolution.old_schema.get("fields", {}).get(field, {}).get("type")
210 new_type = evolution.new_schema.get("fields", {}).get(field, {}).get("type")
212 if old_type and new_type and old_type != new_type: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 if evolution.affected_downstream:
214 breaking.append(evolution)
216 return breaking
218 def get_evolution_timeline(self, node_id: str) -> List[SchemaEvolution]:
219 """Get schema evolution timeline for a node."""
220 return [e for e in self.evolution_history if e.node_id == node_id]
223class SLAValidator:
224 """Validate SLA compliance."""
226 def __init__(self, graph: DependencyGraph):
227 self.graph = graph
228 self.sla_definitions: Dict[str, str] = {}
229 self.performance_metrics: Dict[str, str] = {}
231 def set_sla(self, node_id: str, sla_target: str) -> None:
232 """Set SLA target for a node."""
233 self.sla_definitions[node_id] = sla_target
235 def record_performance(self, node_id: str, actual: str) -> None:
236 """Record actual performance."""
237 self.performance_metrics[node_id] = actual
239 def validate_node(self, node_id: str) -> SLAValidation:
240 """Validate SLA compliance for a node."""
241 has_sla = node_id in self.sla_definitions
242 sla_target = self.sla_definitions.get(node_id)
243 actual = self.performance_metrics.get(node_id)
245 node = self.graph.nodes.get(node_id)
246 severity = node.severity if node else NodeSeverity.MEDIUM
248 # Determine compliance status
249 if not has_sla: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 compliance_status = "no_sla"
251 elif not actual: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 compliance_status = "pending"
253 elif self._is_compliant(sla_target, actual):
254 compliance_status = "compliant"
255 else:
256 compliance_status = "violated"
258 return SLAValidation(
259 node_id=node_id,
260 has_sla=has_sla,
261 sla_target=sla_target,
262 actual_performance=actual,
263 compliance_status=compliance_status,
264 severity=severity,
265 )
267 def _is_compliant(self, target: str, actual: str) -> bool:
268 """Check if actual meets target."""
269 # Simple comparison (can be enhanced)
270 try:
271 target_val = float(target.replace("ms", "").replace("s", ""))
272 actual_val = float(actual.replace("ms", "").replace("s", ""))
273 return actual_val <= target_val
274 except (ValueError, AttributeError):
275 return False
277 def validate_all(self) -> Dict[str, SLAValidation]:
278 """Validate all nodes with SLAs."""
279 return {node_id: self.validate_node(node_id) for node_id in self.sla_definitions}
281 def get_sla_violations(self) -> List[SLAValidation]:
282 """Get nodes violating SLAs."""
283 validations = self.validate_all()
284 return [v for v in validations.values() if v.compliance_status == "violated"]
286 def get_missing_slas(self) -> List[str]:
287 """Get critical nodes without SLAs."""
288 critical_nodes = [
289 n.id for n in self.graph.nodes.values() if n.severity == NodeSeverity.CRITICAL
290 ]
291 return [n for n in critical_nodes if n not in self.sla_definitions]
294class TestCoverageAnalyzer:
295 """Analyze test coverage for nodes."""
297 # Not a pytest test case; this only starts with "Test" because it
298 # analyzes test coverage. Prevents pytest from warning when it tries
299 # (and fails) to collect this as a test class.
300 __test__ = False
302 def __init__(self, graph: DependencyGraph):
303 self.graph = graph
304 self.test_assignments: Dict[str, List[str]] = {}
306 def assign_tests(self, node_id: str, tests: List[str], test_type: str = "unit") -> None:
307 """Assign tests to a node."""
308 if node_id not in self.test_assignments: 308 ↛ 312line 308 didn't jump to line 312 because the condition on line 308 was always true
309 self.test_assignments[node_id] = []
311 # Add tests with type metadata
312 for test in tests:
313 self.test_assignments[node_id].append(f"{test}:{test_type}")
315 def analyze_coverage(self, node_id: str) -> TestCoverageAnalysis:
316 """Analyze test coverage for a node."""
317 tests = self.test_assignments.get(node_id, [])
319 # Count test types
320 test_types = {}
321 for test in tests:
322 parts = test.split(":")
323 ttype = parts[1] if len(parts) > 1 else "unit"
324 test_types[ttype] = test_types.get(ttype, 0) + 1
326 # Calculate coverage percentage
327 coverage = min(100.0, (len(tests) / 5.0) * 100) if tests else 0.0
329 # Determine status
330 if coverage >= 80:
331 status = "good"
332 elif coverage >= 50:
333 status = "adequate"
334 else:
335 status = "poor"
337 # Identify missing tests
338 node = self.graph.nodes.get(node_id)
339 missing = []
340 if node:
341 if node.type == NodeType.DBT_MODEL and not tests:
342 missing.append("dbt_test")
343 if node.severity == NodeSeverity.CRITICAL and len(tests) < 3:
344 missing.append("additional_tests")
346 return TestCoverageAnalysis(
347 node_id=node_id,
348 total_tests=len(tests),
349 test_types=test_types,
350 coverage_percentage=coverage,
351 coverage_status=status,
352 missing_tests=missing,
353 )
355 def analyze_all(self) -> Dict[str, TestCoverageAnalysis]:
356 """Analyze coverage for all nodes with tests."""
357 return {node_id: self.analyze_coverage(node_id) for node_id in self.test_assignments}
359 def get_poorly_tested_nodes(self) -> List[str]:
360 """Get nodes with poor test coverage."""
361 analyses = self.analyze_all()
362 return [
363 node_id for node_id, analysis in analyses.items() if analysis.coverage_status == "poor"
364 ]
366 def get_critical_test_gaps(self) -> List[Tuple[str, List[str]]]:
367 """Get critical nodes missing tests."""
368 gaps = []
369 for node_id, analysis in self.analyze_all().items():
370 node = self.graph.nodes.get(node_id)
371 if node and node.severity == NodeSeverity.CRITICAL: 371 ↛ 369line 371 didn't jump to line 369 because the condition on line 371 was always true
372 if analysis.total_tests == 0: 372 ↛ 374line 372 didn't jump to line 374 because the condition on line 372 was always true
373 gaps.append((node_id, ["all_types"]))
374 elif analysis.missing_tests:
375 gaps.append((node_id, analysis.missing_tests))
377 return gaps