Send webhook notification when progress increases.
(passing: int, total: int, project_dir: Path)
| 143 | |
| 144 | |
| 145 | def send_progress_webhook(passing: int, total: int, project_dir: Path) -> None: |
| 146 | """Send webhook notification when progress increases.""" |
| 147 | if not WEBHOOK_URL: |
| 148 | return # Webhook not configured |
| 149 | |
| 150 | from autoforge_paths import get_progress_cache_path |
| 151 | cache_file = get_progress_cache_path(project_dir) |
| 152 | previous = 0 |
| 153 | previous_passing_ids = set() |
| 154 | |
| 155 | # Read previous progress and passing feature IDs |
| 156 | if cache_file.exists(): |
| 157 | try: |
| 158 | cache_data = json.loads(cache_file.read_text()) |
| 159 | previous = cache_data.get("count", 0) |
| 160 | previous_passing_ids = set(cache_data.get("passing_ids", [])) |
| 161 | except Exception: |
| 162 | previous = 0 |
| 163 | |
| 164 | # Only notify if progress increased |
| 165 | if passing > previous: |
| 166 | # Find which features are now passing via API |
| 167 | completed_tests = [] |
| 168 | current_passing_ids = [] |
| 169 | |
| 170 | # Detect transition from old cache format (had count but no passing_ids) |
| 171 | # In this case, we can't reliably identify which specific tests are new |
| 172 | is_old_cache_format = len(previous_passing_ids) == 0 and previous > 0 |
| 173 | |
| 174 | # Get all passing features via direct database access |
| 175 | all_passing = get_all_passing_features(project_dir) |
| 176 | for feature in all_passing: |
| 177 | feature_id = feature.get("id") |
| 178 | current_passing_ids.append(feature_id) |
| 179 | # Only identify individual new tests if we have previous IDs to compare |
| 180 | if not is_old_cache_format and feature_id not in previous_passing_ids: |
| 181 | # This feature is newly passing |
| 182 | name = feature.get("name", f"Feature #{feature_id}") |
| 183 | category = feature.get("category", "") |
| 184 | if category: |
| 185 | completed_tests.append(f"{category} {name}") |
| 186 | else: |
| 187 | completed_tests.append(name) |
| 188 | |
| 189 | payload = { |
| 190 | "event": "test_progress", |
| 191 | "passing": passing, |
| 192 | "total": total, |
| 193 | "percentage": round((passing / total) * 100, 1) if total > 0 else 0, |
| 194 | "previous_passing": previous, |
| 195 | "tests_completed_this_session": passing - previous, |
| 196 | "completed_tests": completed_tests, |
| 197 | "project": project_dir.name, |
| 198 | "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), |
| 199 | } |
| 200 | |
| 201 | try: |
| 202 | req = urllib.request.Request( |
no test coverage detected