Coverage for python/pyairflowtester/scanner.py: 66%
94 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 20:43 +0530
1"""
2Scanner module for artifact analysis.
4Performs static analysis on Airflow DAGs and dbt projects to detect
5violations before deployment.
6"""
8import logging
9from pathlib import Path
10from typing import Any, Dict, List
12logger = logging.getLogger(__name__)
15class Scanner:
16 """Static analysis scanner for artifacts."""
18 def __init__(self):
19 """Initialize scanner."""
20 self.dag_rules = self._get_dag_rules()
21 self.dbt_rules = self._get_dbt_rules()
22 self.config_rules = self._get_config_rules()
24 def scan_dags(self, dags_path: Path) -> List[Dict[str, Any]]:
25 """
26 Scan Airflow DAGs directory.
28 Args:
29 dags_path: Path to DAGs directory
31 Returns:
32 List of violations found
33 """
34 violations = []
35 logger.info(f"Scanning DAGs in {dags_path}")
37 for dag_file in dags_path.glob("*.py"):
38 if dag_file.name.startswith("_"): 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true
39 continue
41 try:
42 with open(dag_file) as f:
43 source_code = f.read()
44 except Exception as e:
45 logger.error(f"Error reading {dag_file}: {e}")
46 continue
48 # Apply DAG rules. Each rule is isolated: a rule raising an
49 # exception must not suppress findings from the other rules
50 # for this file.
51 for rule in self.dag_rules:
52 try:
53 rule_violations = rule.evaluate(source_code, dag_file.name)
54 violations.extend(rule_violations)
55 except Exception as e:
56 logger.warning(
57 f"Rule {getattr(rule, 'id', rule.__class__.__name__)} "
58 f"failed on {dag_file.name}: {e}"
59 )
61 logger.info(f"Found {len(violations)} DAG violations")
62 return violations
64 def scan_config(self, config_path: Path) -> List[Dict[str, Any]]:
65 """
66 Scan an airflow.cfg configuration file.
68 Args:
69 config_path: Path to an airflow.cfg (INI-format) file
71 Returns:
72 List of violations found
73 """
74 violations = []
75 logger.info(f"Scanning Airflow configuration at {config_path}")
77 try:
78 config = self._parse_airflow_cfg(config_path)
79 except Exception as e:
80 logger.error(f"Error reading {config_path}: {e}")
81 return violations
83 # Apply config rules. Each rule is isolated so one failing rule
84 # does not suppress findings from the others.
85 for rule in self.config_rules:
86 try:
87 rule_violations = rule.evaluate(config)
88 violations.extend(rule_violations)
89 except Exception as e:
90 logger.warning(
91 f"Rule {getattr(rule, 'id', rule.__class__.__name__)} "
92 f"failed on {config_path}: {e}"
93 )
95 logger.info(f"Found {len(violations)} configuration violations")
96 return violations
98 @staticmethod
99 def _parse_airflow_cfg(config_path: Path) -> Dict[str, Dict[str, Any]]:
100 """Parse an INI-format airflow.cfg into a nested section -> key -> value dict.
102 configparser yields every value as a raw string. Config rules such
103 as TLSConfigurationRule/RBACConfigurationRule do truthiness checks
104 (`if not enable_ssl`) that need real booleans -- the string "False"
105 is truthy in Python, so without coercion those rules would silently
106 never fire against a real airflow.cfg. Values are coerced using the
107 same boolean vocabulary as configparser.ConfigParser.getboolean
108 (1/yes/true/on -> True, 0/no/false/off -> False); everything else
109 is left as a string so rules that expect ints/strings (and call
110 int(...) themselves) keep working unchanged.
111 """
112 import configparser
114 parser = configparser.ConfigParser()
115 parser.read(config_path)
117 return {
118 section: {
119 key: Scanner._coerce_config_value(value) for key, value in parser.items(section)
120 }
121 for section in parser.sections()
122 }
124 @staticmethod
125 def _coerce_config_value(value: str) -> Any:
126 """Coerce an INI string value to bool where it unambiguously looks boolean."""
127 lowered = value.strip().lower()
128 if lowered in ("1", "yes", "true", "on"):
129 return True
130 if lowered in ("0", "no", "false", "off"): 130 ↛ 132line 130 didn't jump to line 132 because the condition on line 130 was always true
131 return False
132 return value
134 def scan_dbt(self, dbt_path: Path) -> List[Dict[str, Any]]:
135 """
136 Scan dbt project.
138 Args:
139 dbt_path: Path to dbt project
141 Returns:
142 List of violations found
143 """
144 violations = []
145 logger.info(f"Scanning dbt project in {dbt_path}")
147 # Look for manifest.json
148 manifest_path = dbt_path / "target" / "manifest.json"
149 if not manifest_path.exists():
150 logger.warning(f"manifest.json not found in {dbt_path / 'target'}")
151 return violations
153 try:
154 import json
156 with open(manifest_path) as f:
157 manifest = json.load(f)
158 except Exception as e:
159 logger.error(f"Error reading manifest for {dbt_path}: {e}")
160 return violations
162 # Apply dbt rules. Each rule is isolated so one failing rule does
163 # not suppress findings from the others.
164 for rule in self.dbt_rules:
165 try:
166 rule_violations = rule.evaluate(manifest)
167 violations.extend(rule_violations)
168 except Exception as e:
169 logger.warning(
170 f"Rule {getattr(rule, 'id', rule.__class__.__name__)} "
171 f"failed on dbt manifest: {e}"
172 )
174 logger.info(f"Found {len(violations)} dbt violations")
175 return violations
177 def _get_dag_rules(self) -> List:
178 """Get DAG analysis rules (AFW001-AFW015)."""
179 from pyairflowtester.rules.dag import (
180 CircularDependencyRule,
181 ExpensiveImportsRule,
182 MissingSLARule,
183 ParseTimeRule,
184 )
185 from pyairflowtester.rules.dag_advanced import (
186 AlertingConfigurationRule,
187 BranchComplexityRule,
188 CatchupConfigRule,
189 DocumentationRule,
190 HardcodedConnectionRule,
191 OperatorDeprecationRule,
192 RetryConfigurationRule,
193 SecretsInCodeRule,
194 SensorTimeoutRule,
195 SourceCodePoolConfigurationRule,
196 TaskCountRule,
197 )
199 return [
200 # Basic rules (AFW001-AFW004)
201 CircularDependencyRule(),
202 MissingSLARule(),
203 ExpensiveImportsRule(),
204 ParseTimeRule(),
205 # Advanced rules (AFW005-AFW015)
206 TaskCountRule(),
207 CatchupConfigRule(),
208 SourceCodePoolConfigurationRule(),
209 HardcodedConnectionRule(),
210 SecretsInCodeRule(),
211 RetryConfigurationRule(),
212 SensorTimeoutRule(),
213 BranchComplexityRule(),
214 DocumentationRule(),
215 AlertingConfigurationRule(),
216 OperatorDeprecationRule(),
217 ]
219 def _get_dbt_rules(self) -> List:
220 """Get dbt analysis rules (DBT001-DBT003)."""
221 from pyairflowtester.rules.dbt import (
222 MissingTestsRule,
223 RedundantTestsRule,
224 UntestedModelRule,
225 )
227 return [
228 MissingTestsRule(),
229 RedundantTestsRule(),
230 UntestedModelRule(),
231 ]
233 def _get_config_rules(self) -> List:
234 """Get Airflow configuration audit rules (CFG001-CFG015)."""
235 from pyairflowtester.rules.config import (
236 AirflowCfgPoolConfigurationRule,
237 ConcurrencyConfigurationRule,
238 DAGFolderConfigurationRule,
239 DatabaseBackupRule,
240 EncryptionConfigurationRule,
241 ExecutorConfigurationRule,
242 LogRetentionRule,
243 LogStorageRule,
244 MaxActiveRunsRule,
245 QueueConfigurationRule,
246 RBACConfigurationRule,
247 SchedulerConfigurationRule,
248 TLSConfigurationRule,
249 WorkerConfigurationRule,
250 XComConfigurationRule,
251 )
253 return [
254 ExecutorConfigurationRule(),
255 AirflowCfgPoolConfigurationRule(),
256 ConcurrencyConfigurationRule(),
257 QueueConfigurationRule(),
258 MaxActiveRunsRule(),
259 XComConfigurationRule(),
260 LogRetentionRule(),
261 EncryptionConfigurationRule(),
262 TLSConfigurationRule(),
263 RBACConfigurationRule(),
264 SchedulerConfigurationRule(),
265 WorkerConfigurationRule(),
266 LogStorageRule(),
267 DatabaseBackupRule(),
268 DAGFolderConfigurationRule(),
269 ]
271 def _get_scorer(self):
272 """Get scorer instance."""
273 from pyairflowtester.scoring import Scorer
275 return Scorer()