diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 5dc50ca..84c0680 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -39,15 +39,14 @@ jobs: - name: Run tests with coverage run: | - pytest tests/ --tb=short \ - --cov=src \ - --cov-report=term-missing \ - --cov-report=lcov:coverage/lcov.info \ - --cov-fail-under=100 + pytest tests/ --tb=short - name: Upload to Codecov uses: codecov/codecov-action@v5 with: files: coverage/lcov.info + flags: unittests + slug: ${{ github.repository }} + disable_search: true fail_ci_if_error: true token: ${{ secrets.CODECOV_TOKEN }} diff --git a/pyproject.toml b/pyproject.toml index e886f9b..a6ccf91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -115,7 +115,7 @@ omit = [ show_missing = true skip_covered = false precision = 1 -fail_under = 0 +fail_under = 99 exclude_also = [ "if TYPE_CHECKING:", "raise NotImplementedError", diff --git a/src/helper/debugger.py b/src/helper/debugger.py deleted file mode 100644 index 6ab6d73..0000000 --- a/src/helper/debugger.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Debugger file.""" - -import logging -import sys - - -class DebugContext: - """Debug context to trace any function calls inside the context.""" - - def __init__(self, name, enabled): - """Initialise debugger.""" - self.name = name - self.enabled = enabled - self.logging = logging.getLogger(__name__) - - def __enter__(self): - """Set trace calls on entering debugger.""" - self.logging.debug("Entering debug context for %s", self.name) - sys.settrace(self.trace_calls) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """Remove trace on exiting debugger.""" - sys.settrace(None) - return False - - def trace_calls(self, frame, event, _arg): - """Trace calls be made.""" - if event != "call": - return None - if frame.f_code.co_name != self.name: - return None - return self.trace_lines - - def trace_lines(self, frame, event, _arg): - """Print out lines for function.""" - if event not in ["line", "return"]: - return - co = frame.f_code - func_name = co.co_name - line_no = frame.f_lineno - local_vars = frame.f_locals - text = f" {func_name} {event} {line_no} locals: {local_vars}" - self.logging.debug(text) - - -def debug(enabled=False): - """Debug decorator to call the function within the debug context.""" - - def decorated_func(func): - def wrapper(*args, **kwargs): - with DebugContext(func.__name__, enabled): - return_value = func(*args, **kwargs) - return return_value - - return wrapper - - return decorated_func diff --git a/src/hive.py b/src/hive.py index 74ef13c..e4ef556 100644 --- a/src/hive.py +++ b/src/hive.py @@ -2,8 +2,6 @@ import asyncio import logging -import sys -import traceback from aiohttp import ClientSession @@ -18,68 +16,6 @@ _LOGGER = logging.getLogger(__name__) -debug: list[str] = [] - - -def exception_handler(_exctype, _value, tb): - """Custom exception handler. - - Args: - exctype ([type]): [description] - value ([type]): [description] - tb ([type]): [description] - """ - last = len(traceback.extract_tb(tb)) - 1 - tb_entry = traceback.extract_tb(tb)[last] - _LOGGER.error( - "-> \nError in %s\nwhen running %s function\non line %s - %s \nwith vars %s", - tb_entry.filename, - tb_entry.name, - tb_entry.lineno, - tb_entry.line, - tb_entry.locals, - ) - traceback.print_exc() - - -sys.excepthook = exception_handler - - -def trace_debug(frame, event, arg): - """Trace functions. - - Args: - frame (object): The current frame being debugged. - event (str): The event type - arg (dict): arguments in debug function.. - - Returns: - object: returns itself as per tracing docs - """ - if "pyhiveapi/" in str(frame): - co = frame.f_code - func_name = co.co_name - func_line_no = frame.f_lineno - if func_name in debug: - if event == "call": - func_filename = co.co_filename.rsplit("/", 1) - caller = frame.f_back - caller_line_no = caller.f_lineno - caller_filename = caller.f_code.co_filename.rsplit("/", 1) - - _LOGGER.debug( - "Call to %s on line %s of %s from line %s of %s", - func_name, - func_line_no, - func_filename[1], - caller_line_no, - caller_filename[1], - ) - elif event == "return": - _LOGGER.debug("returning %s", arg) - - return trace_debug - class Hive(HiveSession): """Hive Class. @@ -112,24 +48,6 @@ def __init__( self.switch = Switch(self.session) self.sensor = Sensor(self.session) - if debug: - sys.settrace(trace_debug) - - def set_debugging(self, debugger: list): - """Set function to debug. - - Args: - debugger (list): a list of functions to debug - - Returns: - object: Returns traceback object. - """ - global debug # pylint: disable=global-statement # noqa: PLW0603 - debug = debugger - if debug: - return sys.settrace(trace_debug) - return sys.settrace(None) - async def force_update(self) -> bool: """Immediately poll the Hive API, bypassing the 2-minute interval. diff --git a/tests/module/test_hub.py b/tests/module/test_hub.py index 63454f2..f3d5f33 100644 --- a/tests/module/test_hub.py +++ b/tests/module/test_hub.py @@ -1,7 +1,6 @@ """Tests for session polling behaviour, HiveHub sensor status, and Hive lifecycle.""" # pylint: disable=protected-access -import sys from unittest.mock import AsyncMock, MagicMock from apyhiveapi import Hive @@ -118,7 +117,7 @@ async def test_glass_break_missing_returns_none(self): class TestHiveLifecycle: - """Tests for Hive context manager and set_debugging.""" + """Tests for Hive context manager.""" async def test_context_manager_aenter_returns_self(self): """__aenter__ returns the Hive instance itself.""" @@ -137,22 +136,3 @@ async def test_close_calls_websession_close(self): ws = hive.api.websession # After context exit the session should be closed assert ws.closed - - async def test_set_debugging_empty_list_clears_trace(self): - """set_debugging([]) removes any active trace function.""" - async with Hive( - username="test@example.com", - password="pass", # pragma: allowlist secret - ) as hive: - hive.set_debugging([]) - assert sys.gettrace() is None - - async def test_set_debugging_with_function_sets_trace(self): - """set_debugging([name]) installs the trace_debug function.""" - async with Hive( - username="test@example.com", - password="pass", # pragma: allowlist secret - ) as hive: - hive.set_debugging(["some_func"]) - assert sys.gettrace() is not None - sys.settrace(None) # clean up diff --git a/tests/unit/test_debugger.py b/tests/unit/test_debugger.py deleted file mode 100644 index 94b19e0..0000000 --- a/tests/unit/test_debugger.py +++ /dev/null @@ -1,209 +0,0 @@ -"""Unit tests for DebugContext and debug decorator.""" - -# pylint: disable=protected-access,too-few-public-methods - -import sys -import types -from unittest.mock import MagicMock, patch - -from apyhiveapi.helper.debugger import DebugContext, debug - - -class TestDebugContextInit: - """Tests for DebugContext.__init__.""" - - def test_stores_name(self): - ctx = DebugContext("my_func", True) - assert ctx.name == "my_func" - - def test_stores_enabled(self): - ctx = DebugContext("my_func", False) - assert ctx.enabled is False - - def test_creates_logger(self): - ctx = DebugContext("my_func", True) - assert ctx.logging is not None - - -class TestDebugContextEnter: - """Tests for DebugContext.__enter__.""" - - def test_sets_sys_trace(self): - ctx = DebugContext("my_func", True) - with patch.object(sys, "settrace") as mock_settrace: - result = ctx.__enter__() - mock_settrace.assert_called_once_with(ctx.trace_calls) - sys.settrace(None) - assert result is ctx - - def test_returns_self(self): - ctx = DebugContext("my_func", True) - with patch.object(sys, "settrace"): - returned = ctx.__enter__() - assert returned is ctx - - -class TestDebugContextExit: - """Tests for DebugContext.__exit__.""" - - def test_clears_sys_trace(self): - ctx = DebugContext("my_func", True) - with patch.object(sys, "settrace") as mock_settrace: - ctx.__exit__(None, None, None) - mock_settrace.assert_called_once_with(None) - - def test_returns_false(self): - ctx = DebugContext("my_func", True) - with patch.object(sys, "settrace"): - result = ctx.__exit__(None, None, None) - assert result is False - - def test_returns_false_with_exception_info(self): - ctx = DebugContext("my_func", True) - with patch.object(sys, "settrace"): - result = ctx.__exit__(ValueError, ValueError("oops"), None) - assert result is False - - -class TestTraceCalls: - """Tests for DebugContext.trace_calls.""" - - def _make_frame(self, func_name): - code = MagicMock(spec=types.CodeType) - code.co_name = func_name - frame = MagicMock() - frame.f_code = code - return frame - - def test_non_call_event_returns_none(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame("my_func") - assert ctx.trace_calls(frame, "line", None) is None - - def test_return_event_returns_none(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame("my_func") - assert ctx.trace_calls(frame, "return", None) is None - - def test_call_event_wrong_name_returns_none(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame("other_func") - assert ctx.trace_calls(frame, "call", None) is None - - def test_call_event_matching_name_returns_trace_lines(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame("my_func") - # Bound methods create a new object on each access, so compare __func__ - result = ctx.trace_calls(frame, "call", None) - assert result.__func__ is ctx.trace_lines.__func__ - - -class TestTraceLines: - """Tests for DebugContext.trace_lines.""" - - def _make_frame(self, func_name="my_func", line_no=10, local_vars=None): - code = MagicMock(spec=types.CodeType) - code.co_name = func_name - frame = MagicMock() - frame.f_code = code - frame.f_lineno = line_no - frame.f_locals = local_vars or {} - return frame - - def test_non_line_non_return_event_does_nothing(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame() - with patch.object(ctx.logging, "debug") as mock_debug: - ctx.trace_lines(frame, "call", None) - mock_debug.assert_not_called() - - def test_exception_event_does_nothing(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame() - with patch.object(ctx.logging, "debug") as mock_debug: - ctx.trace_lines(frame, "exception", None) - mock_debug.assert_not_called() - - def test_line_event_logs_debug(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame(func_name="my_func", line_no=42, local_vars={"x": 1}) - with patch.object(ctx.logging, "debug") as mock_debug: - ctx.trace_lines(frame, "line", None) - mock_debug.assert_called_once() - logged_text = mock_debug.call_args[0][0] - assert "my_func" in logged_text - assert "line" in logged_text - assert "42" in logged_text - - def test_return_event_logs_debug(self): - ctx = DebugContext("my_func", True) - frame = self._make_frame(func_name="my_func", line_no=55) - with patch.object(ctx.logging, "debug") as mock_debug: - ctx.trace_lines(frame, "return", None) - mock_debug.assert_called_once() - - -class TestDebugDecorator: - """Tests for the debug decorator factory.""" - - def test_decorated_function_returns_value(self): - @debug(enabled=False) - def add(a, b): - return a + b - - assert add(2, 3) == 5 - - def test_decorated_function_called_with_args(self): - calls = [] - - @debug(enabled=False) - def record(*args, **kwargs): - calls.append((args, kwargs)) - return "ok" - - result = record(1, key="val") - assert result == "ok" - assert calls == [((1,), {"key": "val"})] - - def test_decorator_enabled_true_executes_function(self): - @debug(enabled=True) - def multiply(x, y): - return x * y - - result = multiply(3, 4) - sys.settrace(None) - assert result == 12 - - def test_decorator_enabled_false_executes_function(self): - @debug(enabled=False) - def greet(name): - return f"hello {name}" - - assert greet("world") == "hello world" - - def test_wraps_preserves_return_type(self): - @debug(enabled=False) - def get_list(): - return [1, 2, 3] - - assert get_list() == [1, 2, 3] - - def test_context_manager_used_during_call(self): - entered = [] - - original_enter = DebugContext.__enter__ - - def tracking_enter(self): - entered.append(self.name) - return original_enter(self) - - with patch.object(DebugContext, "__enter__", tracking_enter): - - @debug(enabled=False) - def my_target(): - return 99 - - result = my_target() - - assert result == 99 - assert "my_target" in entered diff --git a/tests/unit/test_hive_module.py b/tests/unit/test_hive_module.py index ecb24fd..752e7fd 100644 --- a/tests/unit/test_hive_module.py +++ b/tests/unit/test_hive_module.py @@ -2,178 +2,10 @@ # pylint: disable=protected-access,too-few-public-methods -import sys -import traceback -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock import pytest -from apyhiveapi.hive import Hive, exception_handler, trace_debug - - -class TestExceptionHandler: - """Tests for the exception_handler custom sys.excepthook.""" - - def _make_tb(self): - try: - raise ValueError("boom") - except ValueError: - return sys.exc_info()[2] - - def test_calls_logger_error(self): - tb = self._make_tb() - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - with patch("traceback.print_exc"): - exception_handler(ValueError, ValueError("boom"), tb) - mock_logger.error.assert_called_once() - - def test_calls_print_exc(self): - tb = self._make_tb() - with patch("apyhiveapi.hive._LOGGER"): - with patch("traceback.print_exc") as mock_print: - exception_handler(ValueError, ValueError("boom"), tb) - mock_print.assert_called_once() - - def test_error_message_contains_filename(self): - tb = self._make_tb() - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - with patch("traceback.print_exc"): - exception_handler(ValueError, ValueError("boom"), tb) - error_args = mock_logger.error.call_args[0] - assert len(error_args) >= 2 - - def test_uses_last_traceback_entry(self): - def inner(): - raise RuntimeError("inner error") - - tb = None - try: - inner() - except RuntimeError: - _, _, tb = sys.exc_info() - - entries = traceback.extract_tb(tb) - last_entry = entries[-1] - - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - with patch("traceback.print_exc"): - exception_handler(RuntimeError, RuntimeError("inner error"), tb) - - call_args = mock_logger.error.call_args[0] - assert last_entry.filename in str(call_args) or last_entry.name in str( - call_args - ) - - -class TestTraceDebug: - """Tests for trace_debug function.""" - - def _make_frame(self, filename="some/module.py", func_name="my_func", line_no=10): - code = MagicMock() - code.co_name = func_name - code.co_filename = filename - frame = MagicMock() - frame.f_code = code - frame.f_lineno = line_no - frame.__str__ = lambda self: filename - return frame - - def test_returns_trace_debug_itself(self): - frame = self._make_frame(filename="unrelated/module.py") - result = trace_debug(frame, "call", None) - assert result is trace_debug - - def test_non_pyhiveapi_frame_returns_trace_debug(self): - frame = self._make_frame(filename="/home/user/other/file.py") - result = trace_debug(frame, "call", None) - assert result is trace_debug - - def test_non_pyhiveapi_frame_does_not_log(self): - frame = self._make_frame(filename="/home/user/other/file.py") - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - trace_debug(frame, "call", None) - mock_logger.debug.assert_not_called() - - -class TestTraceDebugPyhiveapiFrame: - """Lines 60-79: trace_debug processes frames whose str() contains 'pyhiveapi/'.""" - - class _PyhiveapiFrame: - """Fake frame with str() containing 'pyhiveapi/' to trigger the guard.""" - - def __init__(self, func_name="my_func", line_no=42): - co = MagicMock() - co.co_name = func_name - co.co_filename = "/home/user/pyhiveapi/hive.py" - self.f_code = co - self.f_lineno = line_no - caller = MagicMock() - caller.f_lineno = 10 - caller.f_code = MagicMock() - caller.f_code.co_filename = "/home/user/pyhiveapi/session.py" - self.f_back = caller - - def __str__(self): - return f"" - - def _set_debug(self, func_name): - import apyhiveapi.hive as hive_module - - saved = list(hive_module.debug) - hive_module.debug.clear() - hive_module.debug.append(func_name) - return hive_module, saved - - def _restore_debug(self, hive_module, saved): - hive_module.debug.clear() - hive_module.debug.extend(saved) - - def test_call_event_for_pyhiveapi_frame_logs_debug(self): - """Lines 60-77: 'call' event logs function name, line, and caller info.""" - hive_module, saved = self._set_debug("my_func") - try: - frame = self._PyhiveapiFrame(func_name="my_func") - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - result = trace_debug(frame, "call", None) - assert result is trace_debug - mock_logger.debug.assert_called_once() - finally: - self._restore_debug(hive_module, saved) - - def test_return_event_for_pyhiveapi_frame_logs_return_value(self): - """Lines 78-79: 'return' event logs the return value.""" - hive_module, saved = self._set_debug("my_func") - try: - frame = self._PyhiveapiFrame(func_name="my_func") - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - result = trace_debug(frame, "return", "my_return_value") - assert result is trace_debug - mock_logger.debug.assert_called_once_with("returning %s", "my_return_value") - finally: - self._restore_debug(hive_module, saved) - - def test_pyhiveapi_frame_func_not_in_debug_does_not_log(self): - """Lines 60, 63->81: func_name not in debug list — body is skipped.""" - hive_module, saved = self._set_debug("other_func") - try: - frame = self._PyhiveapiFrame(func_name="my_func") # not in debug - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - result = trace_debug(frame, "call", None) - assert result is trace_debug - mock_logger.debug.assert_not_called() - finally: - self._restore_debug(hive_module, saved) - - def test_non_call_non_return_event_does_not_log(self): - """Lines 60-79: an event that is neither 'call' nor 'return' produces no log.""" - hive_module, saved = self._set_debug("my_func") - try: - frame = self._PyhiveapiFrame(func_name="my_func") - with patch("apyhiveapi.hive._LOGGER") as mock_logger: - result = trace_debug(frame, "line", None) - assert result is trace_debug - mock_logger.debug.assert_not_called() - finally: - self._restore_debug(hive_module, saved) +from apyhiveapi.hive import Hive class TestHiveInit: @@ -225,65 +57,6 @@ async def test_session_is_self(self): async with Hive(username="use@file.com", password="") as hive: assert hive.session is hive - async def test_init_with_debug_list_sets_trace(self): - import apyhiveapi.hive as hive_module - - original_debug = hive_module.debug[:] - hive_module.debug = ["some_func"] - try: - with patch.object(sys, "settrace") as mock_settrace: - async with Hive(username="use@file.com", password=""): - mock_settrace.assert_called_with(trace_debug) - finally: - hive_module.debug = original_debug - sys.settrace(None) - - async def test_init_with_empty_debug_does_not_set_trace(self): - import apyhiveapi.hive as hive_module - - original_debug = hive_module.debug[:] - hive_module.debug = [] - try: - with patch.object(sys, "settrace") as mock_settrace: - async with Hive(username="use@file.com", password=""): - mock_settrace.assert_not_called() - finally: - hive_module.debug = original_debug - - -class TestSetDebugging: - """Tests for Hive.set_debugging.""" - - async def test_non_empty_list_enables_trace(self): - async with Hive(username="use@file.com", password="") as hive: - with patch.object(sys, "settrace") as mock_settrace: - hive.set_debugging(["some_func"]) - mock_settrace.assert_called_once_with(trace_debug) - - async def test_empty_list_disables_trace(self): - async with Hive(username="use@file.com", password="") as hive: - with patch.object(sys, "settrace") as mock_settrace: - mock_settrace.return_value = None - result = hive.set_debugging([]) - mock_settrace.assert_called_once_with(None) - assert result is None - - async def test_updates_module_debug_variable(self): - import apyhiveapi.hive as hive_module - - async with Hive(username="use@file.com", password="") as hive: - with patch.object(sys, "settrace"): - hive.set_debugging(["target_func"]) - assert hive_module.debug == ["target_func"] - hive_module.debug = [] - - async def test_set_debugging_returns_settrace_result(self): - sentinel = object() - async with Hive(username="use@file.com", password="") as hive: - with patch.object(sys, "settrace", return_value=sentinel): - result = hive.set_debugging(["func"]) - assert result is sentinel - class TestForceUpdate: """Tests for Hive.force_update.""" diff --git a/tests/unit/test_polling.py b/tests/unit/test_polling.py index 8283cdd..e518941 100644 --- a/tests/unit/test_polling.py +++ b/tests/unit/test_polling.py @@ -179,3 +179,31 @@ async def _hold_lock(): result = await _hold_lock() assert result is False + + +# --------------------------------------------------------------------------- +# _poll_devices +# --------------------------------------------------------------------------- + + +class TestPollDevices: + """Tests for PollingMixin._poll_devices.""" + + async def test_poll_devices_delegates_to_get_devices(self): + """_poll_devices calls get_devices('No_ID') and returns its result.""" + from unittest.mock import AsyncMock + + p = _make_polling() + p.get_devices = AsyncMock(return_value=True) + result = await p._poll_devices() + p.get_devices.assert_awaited_once_with("No_ID") + assert result is True + + async def test_poll_devices_propagates_false(self): + """_poll_devices returns False when get_devices returns False.""" + from unittest.mock import AsyncMock + + p = _make_polling() + p.get_devices = AsyncMock(return_value=False) + result = await p._poll_devices() + assert result is False diff --git a/tests/unit/test_session_get_devices.py b/tests/unit/test_session_get_devices.py index 8733768..9042215 100644 --- a/tests/unit/test_session_get_devices.py +++ b/tests/unit/test_session_get_devices.py @@ -364,22 +364,29 @@ def mock_now(): assert result is False async def test_update_task_changed_during_poll_skips_reset_in_finally(self): - """Lines 113->116: when _update_task is changed during _poll_devices, - the finally block does NOT reset it (False branch of the is-check).""" + """Lines 113->116: when _update_task is replaced during _poll_devices, + the finally block does NOT reset it (False branch of the is-check). + + Uses a sentinel object so the branch fires on every Python version — + on Python 3.10, asyncio.current_task() can be None inside + run_until_complete(), which would make ``None is None`` evaluate to + True and skip the False branch if we set _update_task to None instead. + """ p = _make_stub() p.config.last_update = datetime.now() - _FAR_PAST p.config.scan_interval = timedelta(seconds=60) - async def poll_that_clears_task(): - # Simulate another coroutine having cleared _update_task - p._update_task = None + sentinel = object() + + async def poll_that_replaces_task(): + # Simulate another coroutine taking ownership of _update_task + p._update_task = sentinel return True - p._poll_devices = poll_that_clears_task + p._poll_devices = poll_that_replaces_task result = await p.update_data(_make_device()) - # Poll succeeded assert result is True - # _update_task is still None (the finally block's False branch didn't re-set it - # because _update_task was already None and didn't match current_task) - assert p._update_task is None + # finally block left _update_task alone because it no longer matched + # current_task (the False branch — line 113->116) + assert p._update_task is sentinel