From b331aa613924b5c13d7b731dc6cd10852bc44ba6 Mon Sep 17 00:00:00 2001 From: Kimi Agent Date: Thu, 19 Mar 2026 20:03:28 -0400 Subject: [PATCH] refactor: break up capture_error() into testable helpers (#523) Co-authored-by: Kimi Agent Co-committed-by: Kimi Agent --- src/infrastructure/error_capture.py | 119 ++++++++++++++------- tests/infrastructure/test_error_capture.py | 93 ++++++++++++++++ 2 files changed, 176 insertions(+), 36 deletions(-) diff --git a/src/infrastructure/error_capture.py b/src/infrastructure/error_capture.py index ee75afb..dd1798a 100644 --- a/src/infrastructure/error_capture.py +++ b/src/infrastructure/error_capture.py @@ -100,36 +100,14 @@ def _get_git_context() -> dict: return {"branch": "unknown", "commit": "unknown"} -def capture_error( - exc: Exception, - source: str = "unknown", - context: dict | None = None, -) -> str | None: - """Capture an error and optionally create a bug report. - - Args: - exc: The exception to capture - source: Module/component where the error occurred - context: Optional dict of extra context (request path, etc.) +def _extract_traceback_info(exc: Exception) -> tuple[str, str, int]: + """Extract formatted traceback, affected file, and line number. Returns: - Task ID of the created bug report, or None if deduplicated/disabled + Tuple of (traceback_string, affected_file, affected_line). """ - from config import settings - - if not settings.error_feedback_enabled: - return None - - error_hash = _stack_hash(exc) - - if _is_duplicate(error_hash): - logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash) - return None - - # Format the stack trace tb_str = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) - # Extract file/line from traceback tb_obj = exc.__traceback__ affected_file = "unknown" affected_line = 0 @@ -139,9 +117,18 @@ def capture_error( affected_file = tb_obj.tb_frame.f_code.co_filename affected_line = tb_obj.tb_lineno - git_ctx = _get_git_context() + return tb_str, affected_file, affected_line - # 1. Log to event_log + +def _log_error_event( + exc: Exception, + source: str, + error_hash: str, + affected_file: str, + affected_line: int, + git_ctx: dict, +) -> None: + """Log the captured error to the event log.""" try: from swarm.event_log import EventType, log_event @@ -161,8 +148,18 @@ def capture_error( except Exception as log_exc: logger.debug("Failed to log error event: %s", log_exc) - # 2. Create bug report task - task_id = None + +def _create_bug_report( + exc: Exception, + source: str, + context: dict | None, + error_hash: str, + tb_str: str, + affected_file: str, + affected_line: int, + git_ctx: dict, +) -> str | None: + """Create a bug report task and return the task ID (or None on failure).""" try: from swarm.task_queue.models import create_task @@ -195,7 +192,6 @@ def capture_error( ) task_id = task.id - # Log the creation event try: from swarm.event_log import EventType, log_event @@ -210,12 +206,16 @@ def capture_error( ) except Exception as exc: logger.warning("Bug report screenshot error: %s", exc) - pass + + return task_id except Exception as task_exc: logger.debug("Failed to create bug report task: %s", task_exc) + return None - # 3. Send notification + +def _notify_bug_report(exc: Exception, source: str) -> None: + """Send a push notification about the captured error.""" try: from infrastructure.notifications.push import notifier @@ -224,11 +224,12 @@ def capture_error( message=f"{type(exc).__name__} in {source}: {str(exc)[:80]}", category="system", ) - except Exception as exc: - logger.warning("Bug report notification error: %s", exc) - pass + except Exception as notify_exc: + logger.warning("Bug report notification error: %s", notify_exc) - # 4. Record in session logger (via registered callback) + +def _record_to_session(exc: Exception, source: str) -> None: + """Record the error via the registered session callback.""" if _error_recorder is not None: try: _error_recorder( @@ -238,4 +239,50 @@ def capture_error( except Exception as log_exc: logger.warning("Bug report session logging error: %s", log_exc) + +def capture_error( + exc: Exception, + source: str = "unknown", + context: dict | None = None, +) -> str | None: + """Capture an error and optionally create a bug report. + + Args: + exc: The exception to capture + source: Module/component where the error occurred + context: Optional dict of extra context (request path, etc.) + + Returns: + Task ID of the created bug report, or None if deduplicated/disabled + """ + from config import settings + + if not settings.error_feedback_enabled: + return None + + error_hash = _stack_hash(exc) + + if _is_duplicate(error_hash): + logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash) + return None + + tb_str, affected_file, affected_line = _extract_traceback_info(exc) + git_ctx = _get_git_context() + + _log_error_event(exc, source, error_hash, affected_file, affected_line, git_ctx) + + task_id = _create_bug_report( + exc, + source, + context, + error_hash, + tb_str, + affected_file, + affected_line, + git_ctx, + ) + + _notify_bug_report(exc, source) + _record_to_session(exc, source) + return task_id diff --git a/tests/infrastructure/test_error_capture.py b/tests/infrastructure/test_error_capture.py index ddf369c..ab47d2f 100644 --- a/tests/infrastructure/test_error_capture.py +++ b/tests/infrastructure/test_error_capture.py @@ -5,9 +5,14 @@ from datetime import UTC, datetime, timedelta from unittest.mock import patch from infrastructure.error_capture import ( + _create_bug_report, _dedup_cache, + _extract_traceback_info, _get_git_context, _is_duplicate, + _log_error_event, + _notify_bug_report, + _record_to_session, _stack_hash, capture_error, ) @@ -193,3 +198,91 @@ class TestCaptureError: def teardown_method(self): _dedup_cache.clear() + + +class TestExtractTracebackInfo: + """Test _extract_traceback_info helper.""" + + def test_returns_three_tuple(self): + try: + raise ValueError("extract test") + except ValueError as e: + tb_str, affected_file, affected_line = _extract_traceback_info(e) + assert "ValueError" in tb_str + assert "extract test" in tb_str + assert affected_file.endswith(".py") + assert affected_line > 0 + + def test_file_points_to_raise_site(self): + try: + _make_exception() + except ValueError as e: + _, affected_file, _ = _extract_traceback_info(e) + assert "test_error_capture" in affected_file + + +class TestLogErrorEvent: + """Test _log_error_event helper.""" + + def test_does_not_crash_on_missing_deps(self): + try: + raise RuntimeError("log test") + except RuntimeError as e: + _log_error_event(e, "test", "abc123", "file.py", 42, {"branch": "main"}) + + +class TestCreateBugReport: + """Test _create_bug_report helper.""" + + def test_does_not_crash_on_missing_deps(self): + try: + raise RuntimeError("report test") + except RuntimeError as e: + result = _create_bug_report( + e, "test", None, "abc123", "traceback...", "file.py", 42, {} + ) + # May return None if swarm deps unavailable — that's fine + assert result is None or isinstance(result, str) + + def test_with_context(self): + try: + raise RuntimeError("ctx test") + except RuntimeError as e: + result = _create_bug_report(e, "test", {"path": "/api"}, "abc", "tb", "f.py", 1, {}) + assert result is None or isinstance(result, str) + + +class TestNotifyBugReport: + """Test _notify_bug_report helper.""" + + def test_does_not_crash(self): + try: + raise RuntimeError("notify test") + except RuntimeError as e: + _notify_bug_report(e, "test") + + +class TestRecordToSession: + """Test _record_to_session helper.""" + + def test_does_not_crash_without_recorder(self): + try: + raise RuntimeError("session test") + except RuntimeError as e: + _record_to_session(e, "test") + + def test_calls_registered_recorder(self): + from infrastructure.error_capture import register_error_recorder + + calls = [] + register_error_recorder(lambda **kwargs: calls.append(kwargs)) + try: + try: + raise RuntimeError("callback test") + except RuntimeError as e: + _record_to_session(e, "test_source") + assert len(calls) == 1 + assert "RuntimeError" in calls[0]["error"] + assert calls[0]["context"] == "test_source" + finally: + register_error_recorder(None)