Coverage for python/pyairflowtester/report.py: 20%

1"""

2Report generation module.

4Generates reports in multiple formats: JSON, HTML, Markdown, SARIF.

5"""

7import json

8import logging

9from datetime import datetime

10from pathlib import Path

11from typing import Any, Dict, List

13logger = logging.getLogger(__name__)

16class ReportGenerator:

17 """Generates analysis reports in multiple formats."""

19 def generate(self, format: str, violations: List[Dict[str, Any]], output_path: Path) -> Path:

20 """

21 Generate report in specified format.

23 Args:

24 format: Report format (json, html, markdown, sarif)

25 violations: List of violations

26 output_path: Output file path

28 Returns:

29 Path to generated report

30 """

31 output_path = Path(output_path)

32 output_path.parent.mkdir(parents=True, exist_ok=True)

34 logger.info(f"Generating {format} report to {output_path}")

36 if format == "json":

37 return self.generate_json(violations, output_path)

38 elif format == "html":

39 return self.generate_html(violations, output_path)

40 elif format == "markdown":

41 return self.generate_markdown(violations, output_path)

42 elif format == "sarif":

43 return self.generate_sarif(violations, output_path)

44 else:

45 raise ValueError(f"Unsupported format: {format}")

47 def generate_json(self, violations: List[Dict[str, Any]], output_path: Path) -> Path:

48 """Generate JSON report."""

49 report = {

50 "generated_at": datetime.utcnow().isoformat(),

51 "total_violations": len(violations),

52 "violations": violations,

53 }

55 with open(output_path, "w") as f:

56 json.dump(report, f, indent=2)

58 logger.info(f"JSON report generated: {output_path}")

59 return output_path

61 def generate_html(self, violations: List[Dict[str, Any]], output_path: Path) -> Path:

62 """Generate HTML report."""

63 html = f"""<!DOCTYPE html>

64<html>

65<head>

66 <meta charset="utf-8">

67 <title>PyAirflowTester Report</title>

68 <style>

69 body {{ font-family: Arial, sans-serif; margin: 20px; background-color: #f5f5f5; }}

70 .container {{

71 max-width: 1000px; margin: 0 auto; background-color: white;

72 padding: 20px; border-radius: 8px;

73 }}

74 h1 {{ color: #333; }}

75 .summary {{ background-color: #e8f4f8; padding: 15px; border-radius: 5px; margin: 20px 0; }}

76 .violation {{

77 border-left: 4px solid #dc3545; padding: 15px; margin: 10px 0;

78 background-color: #f8f9fa;

79 }}

80 .critical {{ border-left-color: #dc3545; }}

81 .high {{ border-left-color: #fd7e14; }}

82 .medium {{ border-left-color: #ffc107; }}

83 .low {{ border-left-color: #17a2b8; }}

84 table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}

85 th, td {{ padding: 12px; text-align: left; border-bottom: 1px solid #ddd; }}

86 th {{ background-color: #f8f9fa; font-weight: bold; }}

87 .timestamp {{ color: #666; font-size: 0.9em; }}

88 </style>

89</head>

90<body>

91 <div class="container">

92 <h1>PyAirflowTester Analysis Report</h1>

93 <p class="timestamp">Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>

95 <div class="summary">

96 <h2>Summary</h2>

97 <p><strong>Total Violations:</strong> {len(violations)}</p>

98 </div>

100 <h2>Violations</h2>

101 <table>

102 <thead>

103 <tr>

104 <th>Rule</th>

105 <th>Severity</th>

106 <th>Resource</th>

107 <th>Message</th>

108 </tr>

109 </thead>

110 <tbody>

111"""

112

113 for v in violations:

114 severity = v.get("severity", "info")

115 html += f"""

116 <tr class="{severity}">

117 <td>{v.get("rule_id", "")}</td>

118 <td><strong>{severity.upper()}</strong></td>

119 <td>{v.get("affected_resource", "")}</td>

120 <td>{v.get("message", "")}</td>

121 </tr>

122"""

123

124 html += """

125 </tbody>

126 </table>

127 </div>

128</body>

129</html>

130"""

131

132 with open(output_path, "w") as f:

133 f.write(html)

134

135 logger.info(f"HTML report generated: {output_path}")

136 return output_path

137

138 def generate_markdown(self, violations: List[Dict[str, Any]], output_path: Path) -> Path:

139 """Generate Markdown report."""

140 md = f"""# PyAirflowTester Analysis Report

141

142Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

143

144## Summary

145

146- **Total Violations:** {len(violations)}

147

148## Violations

149

151|------|----------|----------|---------|

152"""

153

154 for v in violations:

155 rule_id = v.get("rule_id", "")

156 severity = v.get("severity", "info")

157 resource = v.get("affected_resource", "")

158 message = v.get("message", "")

159 md += f"| {rule_id} | {severity} | {resource} | {message} |\n"

160

161 with open(output_path, "w") as f:

162 f.write(md)

163

164 logger.info(f"Markdown report generated: {output_path}")

165 return output_path

166

167 def generate_sarif(self, violations: List[Dict[str, Any]], output_path: Path) -> Path:

168 """Generate SARIF report for GitHub integration."""

169 sarif = {

170 "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",

171 "version": "2.1.0",

172 "runs": [

173 {

174 "tool": {

175 "driver": {

176 "name": "PyAirflowTester",

177 "version": "0.1.0",

178 "informationUri": "https://github.com/mullassery/pyairflowtester",

179 }

180 },

181 "results": [

182 {

183 "ruleId": v.get("rule_id", ""),

184 "message": {"text": v.get("message", "")},

185 "level": self._severity_to_level(v.get("severity", "note")),

186 "locations": [

187 {

188 "physicalLocation": {

189 "artifactLocation": {"uri": v.get("affected_resource", "")}

190 }

191 }

192 ],

193 }

194 for v in violations

195 ],

196 }

197 ],

198 }

199

200 with open(output_path, "w") as f:

201 json.dump(sarif, f, indent=2)

202

203 logger.info(f"SARIF report generated: {output_path}")

204 return output_path

205

206 @staticmethod

207 def _severity_to_level(severity: str) -> str:

208 """Convert severity to SARIF level."""

209 mapping = {

210 "critical": "error",

211 "high": "error",

212 "medium": "warning",

213 "low": "note",

214 "info": "note",

215 }

216 return mapping.get(severity, "note")