ed
108e77e11d
Replaces the silent-swallow logging.debug bodies in _on_sigint and _install_sigint_exit_handler with proper Result[T] propagation: - _shutdown_io_pool_result() -> Result[None]: wraps io_pool.shutdown with OSError/RuntimeError/ValueError -> ErrorInfo(original=e) - _install_signal_handler_result(handler) -> Result[None]: wraps signal.signal() with ValueError/OSError -> ErrorInfo(original=e) - _install_sigint_exit_handler stores result.errors[0] on self._signal_handler_error: Optional[ErrorInfo] for sub-track 4 GUI The os._exit(0) inside the signal handler IS the drain (Pattern 3: intentional termination per error_handling.md:419). The stderr write before os._exit is part of the termination pattern (Heuristic D match). TIER-2 READ conductor/code_styleguides/error_handling.md before Phase 6. Audit: INTERNAL_SILENT_SWALLOW for src/app_controller.py: 30 -> 28.
209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
"""
|
|
Tests for the data-oriented error handling convention applied to src/app_controller.py.
|
|
|
|
This file verifies the contract that the 32 INTERNAL_BROAD_CATCH + 8 INTERNAL_SILENT_SWALLOW
|
|
+ 4 INTERNAL_RETHROW + 1 INTERNAL_OPTIONAL_RETURN sites in src/app_controller.py are migrated
|
|
to the Result[T] pattern (per conductor/code_styleguides/error_handling.md).
|
|
|
|
The tests are pattern templates:
|
|
- test_offload_entry_payload_returns_dict - the _offload_entry_payload helper returns a dict
|
|
(the existing test for the regression fix is in test_app_controller_offloading.py; this
|
|
one is a shape check).
|
|
- test_migrated_method_returns_result_on_success - methods migrated to Result[T] return
|
|
Result[data=...] with no errors on the success path.
|
|
- test_migrated_method_returns_result_with_error_on_failure - methods migrated to Result[T]
|
|
return Result with errors=... on the failure path.
|
|
- test_app_controller_does_not_use_broad_except - static check: src/app_controller.py
|
|
has no `except Exception:` clauses left (all 32 are migrated to specific exceptions).
|
|
- test_offload_entry_payload_preserves_unchanged_payload - verifies the no-op path for
|
|
entries without tool_call or tool_result kinds.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from src.app_controller import AppController, _install_sigint_exit_handler
|
|
from src.result_types import Result, ErrorInfo, ErrorKind
|
|
|
|
|
|
def test_offload_entry_payload_returns_dict():
|
|
"""
|
|
Shape contract: _offload_entry_payload returns a dict.
|
|
The actual happy-path and error-path tests are in test_app_controller_offloading.py.
|
|
"""
|
|
with patch("src.app_controller.performance_monitor.PerformanceMonitor"):
|
|
ctrl = AppController()
|
|
out = ctrl._offload_entry_payload({"kind": "request", "payload": {"message": "hi"}, "ts": "12:00:00"})
|
|
assert isinstance(out, dict)
|
|
assert out["kind"] == "request"
|
|
assert out["payload"]["message"] == "hi"
|
|
|
|
|
|
def test_migrated_method_returns_result_on_success():
|
|
"""
|
|
Pattern template: methods migrated to Result[T] return Result[data=...] with no errors
|
|
on the success path. The 5 callback-handler sites (Batch 1) include _handle_custom_callback
|
|
and _handle_click; both return Result[None] after migration.
|
|
"""
|
|
from src.app_controller import _handle_custom_callback
|
|
controller = MagicMock()
|
|
controller._predefined_callbacks = {}
|
|
# Use a no-op callback so the body runs without error
|
|
task = {"callback": (lambda: None), "args": []}
|
|
result = _handle_custom_callback(controller, task)
|
|
assert isinstance(result, Result)
|
|
assert result.ok is True
|
|
assert result.errors == []
|
|
|
|
|
|
def test_migrated_method_returns_result_with_error_on_failure():
|
|
"""
|
|
Pattern template: methods migrated to Result[T] return Result with errors=... when
|
|
the underlying call raises. The migrated callback sites catch the specific exception
|
|
(e.g. TypeError, ValueError) and convert it to ErrorInfo.
|
|
"""
|
|
from src.app_controller import _handle_custom_callback
|
|
controller = MagicMock()
|
|
controller._predefined_callbacks = {}
|
|
# Callback that raises a specific exception
|
|
def bad_cb():
|
|
raise ValueError("simulated failure")
|
|
task = {"callback": bad_cb, "args": []}
|
|
result = _handle_custom_callback(controller, task)
|
|
assert isinstance(result, Result)
|
|
assert result.ok is False
|
|
assert any("ValueError" in e.message or "simulated failure" in e.message for e in result.errors)
|
|
|
|
|
|
def test_app_controller_does_not_use_broad_except():
|
|
"""
|
|
Static check via the audit: src/app_controller.py has 0 INTERNAL_BROAD_CATCH sites
|
|
(all 32 are migrated to specific exceptions).
|
|
|
|
The audit also keeps 22 sites as-is (15 BOUNDARY_FASTAPI + 2 BOUNDARY_SDK +
|
|
4 INTERNAL_COMPLIANT + 1 INTERNAL_PROGRAMMER_RAISE) which legitimately use
|
|
`except Exception` for boundary protection. Those are not part of the
|
|
INTERNAL_BROAD_CATCH count.
|
|
|
|
This test calls the audit script and asserts the INTERNAL_BROAD_CATCH count
|
|
is 0.
|
|
"""
|
|
import json
|
|
import subprocess
|
|
r = subprocess.run(
|
|
['uv', 'run', 'python', 'scripts/audit_exception_handling.py', '--json'],
|
|
capture_output=True, text=True, cwd='.'
|
|
)
|
|
data = json.loads(r.stdout)
|
|
app = [f for f in data['files'] if 'app_controller' in f.get('filename', '')][0]
|
|
broad_sites = [f for f in app['findings'] if f.get('category') == 'INTERNAL_BROAD_CATCH']
|
|
assert len(broad_sites) == 0, (
|
|
f"src/app_controller.py still has {len(broad_sites)} INTERNAL_BROAD_CATCH sites at lines "
|
|
f"{[f.get('line') for f in broad_sites]}. All 32 must be migrated to specific exceptions."
|
|
)
|
|
|
|
|
|
def test_offload_entry_payload_preserves_unchanged_payload():
|
|
"""
|
|
Verifies the no-op path: entries without tool_call or tool_result kinds are returned
|
|
with the payload unchanged (no [REF:...] rewriting).
|
|
"""
|
|
with patch("src.app_controller.performance_monitor.PerformanceMonitor"):
|
|
ctrl = AppController()
|
|
entry = {"kind": "request", "payload": {"message": "hi"}, "ts": "12:00:00"}
|
|
out = ctrl._offload_entry_payload(entry)
|
|
assert out == entry
|
|
|
|
|
|
# --- Phase 6: Group 6.1 (signal handlers; Pattern 3 drain via os._exit) ---
|
|
|
|
def test_shutdown_io_pool_result_returns_ok_when_pool_shuts_down_cleanly():
|
|
"""
|
|
Pattern 3 drain: _shutdown_io_pool_result returns Result[None] with no errors
|
|
when the IO pool shuts down without raising.
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
ctrl._io_pool = MagicMock()
|
|
ctrl._io_pool.shutdown = MagicMock()
|
|
result = ctrl._shutdown_io_pool_result()
|
|
assert isinstance(result, Result)
|
|
assert result.ok is True
|
|
assert result.errors == []
|
|
|
|
|
|
def test_shutdown_io_pool_result_returns_error_when_pool_raises():
|
|
"""
|
|
Pattern 3 drain: _shutdown_io_pool_result converts OSError/RuntimeError/ValueError
|
|
to ErrorInfo(original=e) in Result.errors.
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
ctrl._io_pool = MagicMock()
|
|
ctrl._io_pool.shutdown = MagicMock(side_effect=RuntimeError("pool broken"))
|
|
result = ctrl._shutdown_io_pool_result()
|
|
assert isinstance(result, Result)
|
|
assert result.ok is False
|
|
assert len(result.errors) == 1
|
|
assert isinstance(result.errors[0], ErrorInfo)
|
|
assert "pool broken" in result.errors[0].message
|
|
assert result.errors[0].original is not None
|
|
|
|
|
|
def test_install_signal_handler_result_returns_ok_when_signal_installs():
|
|
"""
|
|
Pattern 3 drain: _install_signal_handler_result returns Result[None] on success
|
|
(no errors).
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
handler = lambda signum, frame: None
|
|
with patch("src.app_controller.signal.signal") as mock_signal:
|
|
result = ctrl._install_signal_handler_result(handler)
|
|
assert isinstance(result, Result)
|
|
assert result.ok is True
|
|
assert result.errors == []
|
|
assert mock_signal.called
|
|
|
|
|
|
def test_install_signal_handler_result_returns_error_when_signal_raises():
|
|
"""
|
|
Pattern 3 drain: _install_signal_handler_result converts ValueError/OSError
|
|
to ErrorInfo(original=e).
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
handler = lambda signum, frame: None
|
|
with patch("src.app_controller.signal.signal", side_effect=ValueError("not main thread")):
|
|
result = ctrl._install_signal_handler_result(handler)
|
|
assert isinstance(result, Result)
|
|
assert result.ok is False
|
|
assert len(result.errors) == 1
|
|
assert isinstance(result.errors[0], ErrorInfo)
|
|
assert "not main thread" in result.errors[0].message
|
|
|
|
|
|
def test_install_sigint_exit_handler_stores_error_when_signal_install_fails():
|
|
"""
|
|
Drains the Result to instance state: when the helper returns errors,
|
|
_install_sigint_exit_handler stores the first error on
|
|
ctrl._signal_handler_error for downstream consumers (e.g., sub-track 4 GUI).
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
with patch("src.app_controller.signal.signal", side_effect=ValueError("not main thread")):
|
|
_install_sigint_exit_handler(ctrl)
|
|
assert ctrl._signal_handler_error is not None
|
|
assert isinstance(ctrl._signal_handler_error, ErrorInfo)
|
|
assert "not main thread" in ctrl._signal_handler_error.message
|
|
assert ctrl._signal_handler_error.kind == ErrorKind.INTERNAL
|
|
|
|
|
|
def test_install_sigint_exit_handler_no_error_when_signal_install_succeeds():
|
|
"""
|
|
On success, _signal_handler_error stays as None.
|
|
"""
|
|
from src.app_controller import AppController
|
|
ctrl = AppController()
|
|
with patch("src.app_controller.signal.signal"):
|
|
_install_sigint_exit_handler(ctrl)
|
|
assert ctrl._signal_handler_error is None
|