From 3b98a1e8c03dce8e8451e2502dd302e7fbf657e6 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 14:11:03 -0500 Subject: [PATCH 01/12] Add ability to log to file --- .pre-commit-config.yaml | 2 +- pyproject.toml | 1 + pysquared/boot/__init__.py | 3 ++ pysquared/boot/storage.py | 49 ++++++++++++++++++++++++++++++++ pysquared/boot_functions.py | 56 ------------------------------------- pysquared/logger.py | 42 +++++++++++++--------------- pysquared/sd_card.py | 18 +++--------- 7 files changed, 78 insertions(+), 93 deletions(-) create mode 100644 pysquared/boot/__init__.py create mode 100644 pysquared/boot/storage.py delete mode 100644 pysquared/boot_functions.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index cb3fbf5c..01b58816 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,7 +22,7 @@ repos: language: pygrep types: [python] files: ^pysquared/ - exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py + exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/sd_card\.py - repo: local hooks: diff --git a/pyproject.toml b/pyproject.toml index 25ab6489..84aeae29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ docs = [ [tool.setuptools] packages = [ "pysquared", + "pysquared.boot", "pysquared.config", "pysquared.hardware", "pysquared.hardware.imu", diff --git a/pysquared/boot/__init__.py b/pysquared/boot/__init__.py new file mode 100644 index 00000000..84519555 --- /dev/null +++ b/pysquared/boot/__init__.py @@ -0,0 +1,3 @@ +""" +This module provides utilities that can run during the boot process by adding them to boot.py. +""" diff --git a/pysquared/boot/storage.py b/pysquared/boot/storage.py new file mode 100644 index 00000000..5b884663 --- /dev/null +++ b/pysquared/boot/storage.py @@ -0,0 +1,49 @@ +"""File includes utilities for managing storage during the boot process.""" + +import os +import time + +import storage + + +def mkdir( + path: str, + storage_action_delay: float = 0.02, +) -> None: + """ + Create directories on internal storage during boot. + + In CircuitPython the internal storage is not writable by default. In order to mount + any external storage (such as an SD Card) the drive must be remounted in read/write mode. + This function handles the necessary steps to safely create a directory on the internal + storage during boot. + + Args: + mount_point: Path to mount point + storage_action_delay: Delay after storage actions to ensure stability + + Usage: + ```python + from pysquared.boot.storage import mkdir + mkdir("/sd") + ``` + """ + try: + storage.disable_usb_drive() + time.sleep(storage_action_delay) + print("Disabled USB drive") + + storage.remount("/", False) + time.sleep(storage_action_delay) + print("Remounted root filesystem") + + try: + os.mkdir(path) + print(f"Mount point {path} created.") + except OSError: + print(f"Mount point {path} already exists.") + + finally: + storage.enable_usb_drive() + time.sleep(storage_action_delay) + print("Enabled USB drive") diff --git a/pysquared/boot_functions.py b/pysquared/boot_functions.py deleted file mode 100644 index fdd1cd35..00000000 --- a/pysquared/boot_functions.py +++ /dev/null @@ -1,56 +0,0 @@ -"""Module that provides function relevant on boot""" - -import os -import time - -import storage - - -def set_mount_points( - mount_points=[ - "/sd", - ], - wait_time=0.02, - max_attempt=5, -) -> None: - """ - Mounts the specified mount points. - - Args: - mount_points ([String]): List of mount points - wait_time (float): Time to wait between mount attempts - max_attempts (int): Amount of attempts before failure - """ - mount_points = [ - "/sd", - ] - - wait_time = 0.02 - - storage.disable_usb_drive() - print("Disabling USB drive") - time.sleep(wait_time) - - storage.mount("/", False) - print("Remounting root filesystem") - time.sleep(wait_time) - - attempts = 0 - while attempts < max_attempt: - attempts += 1 - try: - for path in mount_points: - try: - os.mkdir(path) - print(f"Mount point {path} created.") - except OSError: - print(f"Mount point {path} already exists.") - except Exception as e: - print(f"Error creating mount point {path}: {e}") - time.sleep(wait_time) - continue - - break - - storage.enable_usb_drive() - print("Enabling USB drive") diff --git a/pysquared/logger.py b/pysquared/logger.py index 8799abd4..99fa8b45 100644 --- a/pysquared/logger.py +++ b/pysquared/logger.py @@ -18,8 +18,6 @@ import traceback from collections import OrderedDict -import adafruit_pathlib - from .nvm.counter import Counter @@ -82,28 +80,32 @@ class Logger: def __init__( self, error_counter: Counter, - sd_path: adafruit_pathlib.Path = None, - # sd_card: SDCardManager = None, log_level: int = LogLevel.NOTSET, + log_dir: str | None = None, colorized: bool = False, ) -> None: """ Initializes the Logger instance. Args: - error_counter (Counter): Counter for error occurrences. - log_level (int): Initial log level. - colorized (bool): Whether to colorize output. + error_counter: Counter for error occurrences. + log_level: Initial log level. + log_dir: Directory to save log files. If None, logs are not saved to a file. + colorized: Whether to colorize output. """ self._error_counter: Counter = error_counter - self.sd_path: adafruit_pathlib.Path = sd_path self._log_level: int = log_level - self.colorized: bool = colorized - - try: - self.sd_path = self.sd_path / "logs" - except TypeError as e: - print(f"path not set: {e}") + self._log_dir: str | None = log_dir + self._colorized: bool = colorized + + if log_dir is not None: + try: + # Octal number 0o040000 is the stat mode indicating the file being stat'd is a directory + _is_directory: int = 0o040000 + if os.stat(log_dir)[0] != _is_directory: + raise ValueError("Logging path must be a directory.") + except OSError as e: + raise ValueError("Invalid logging path.") from e def _can_print_this_level(self, level_value: int) -> bool: """ @@ -176,16 +178,12 @@ def _log(self, level: str, level_value: int, message: str, **kwargs) -> None: ) if self._can_print_this_level(level_value): - # Write to sd card if mounted - if self.path: - if "logs" not in self.sd_path.iterdir(): - print("/sd/logs does not exist, creating...") - os.mkdir("/sd/logs") - - with open("/sd/logs/activity.log", "a") as f: + if self._log_dir is not None: + file = self._log_dir + os.sep + "activity.log" + with open(file, "a") as f: f.write(json_output + "\n") - if self.colorized: + if self._colorized: json_output = json_output.replace( f'"level": "{level}"', f'"level": "{LogColors[level]}"' ) diff --git a/pysquared/sd_card.py b/pysquared/sd_card.py index 9484730f..a69a0edf 100644 --- a/pysquared/sd_card.py +++ b/pysquared/sd_card.py @@ -1,18 +1,12 @@ """This module provides a SD Card class to manipulate the sd card filesystem""" -# import os - import sdcardio import storage +from busio import SPI +from microcontroller import Pin from .hardware.exception import HardwareInitializationError -try: - from busio import SPI - from microcontroller import Pin -except ImportError: - pass - class SDCardManager: """Class providing various functionalities related to USB and SD card operations.""" @@ -26,13 +20,9 @@ def __init__( baudrate: int = 400000, mount_path: str = "/sd", ) -> None: - self.mounted = False - self.mount_path = mount_path - try: sd = sdcardio.SDCard(spi_bus, chip_select, baudrate) - vfs = storage.VfsFat(sd) - storage.mount(vfs, self.mount_path) - self.mounted = True + vfs = storage.VfsFat(sd) # type: ignore # Issue: https://github.com/adafruit/Adafruit_CircuitPython_Typing/issues/51 + storage.mount(vfs, mount_path) except (OSError, ValueError) as e: raise HardwareInitializationError("Failed to initialize SD Card") from e From ec2583ea053100c271383cd8d392de36b2df9376 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 14:51:51 -0500 Subject: [PATCH 02/12] Add boot.filesystem tests --- mocks/circuitpython/storage.py | 26 +++ pysquared/beacon.py | 4 +- pysquared/boot/{storage.py => filesystem.py} | 11 +- tests/unit/boot/test_filesystem.py | 177 +++++++++++++++++++ 4 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 mocks/circuitpython/storage.py rename pysquared/boot/{storage.py => filesystem.py} (80%) create mode 100644 tests/unit/boot/test_filesystem.py diff --git a/mocks/circuitpython/storage.py b/mocks/circuitpython/storage.py new file mode 100644 index 00000000..beaf59ff --- /dev/null +++ b/mocks/circuitpython/storage.py @@ -0,0 +1,26 @@ +"""Mock for the CircuitPython storage module. + +This module provides a mock implementation of the CircuitPython storage module +for testing purposes. It allows for simulating the behavior of the storage +module without the need for actual hardware. +""" + + +def disable_usb_drive() -> None: + """A mock function to disable the USB drive.""" + pass + + +def enable_usb_drive() -> None: + """A mock function to enable the USB drive.""" + pass + + +def remount(path: str, readonly: bool) -> None: + """A mock function to remount the filesystem. + + Args: + path: The path to remount. + readonly: Whether to mount as read-only. + """ + pass diff --git a/pysquared/beacon.py b/pysquared/beacon.py index 2caaafc9..0896210f 100644 --- a/pysquared/beacon.py +++ b/pysquared/beacon.py @@ -19,9 +19,9 @@ from collections import OrderedDict try: - from mocks.circuitpython.microcontroller import Processor -except ImportError: from microcontroller import Processor +except ImportError: + from mocks.circuitpython.microcontroller import Processor from .hardware.radio.packetizer.packet_manager import PacketManager from .logger import Logger diff --git a/pysquared/boot/storage.py b/pysquared/boot/filesystem.py similarity index 80% rename from pysquared/boot/storage.py rename to pysquared/boot/filesystem.py index 5b884663..0593035c 100644 --- a/pysquared/boot/storage.py +++ b/pysquared/boot/filesystem.py @@ -1,9 +1,12 @@ -"""File includes utilities for managing storage during the boot process.""" +"""File includes utilities for managing the filesystem during the boot process.""" import os import time -import storage +try: + import storage +except ImportError: + import mocks.circuitpython.storage as storage def mkdir( @@ -24,13 +27,13 @@ def mkdir( Usage: ```python - from pysquared.boot.storage import mkdir + from pysquared.boot.filesystem import mkdir mkdir("/sd") ``` """ try: storage.disable_usb_drive() - time.sleep(storage_action_delay) + time.sleep(seconds=storage_action_delay) print("Disabled USB drive") storage.remount("/", False) diff --git a/tests/unit/boot/test_filesystem.py b/tests/unit/boot/test_filesystem.py new file mode 100644 index 00000000..4a9ae616 --- /dev/null +++ b/tests/unit/boot/test_filesystem.py @@ -0,0 +1,177 @@ +"""Unit tests for the storage module. + +This module contains unit tests for the `storage` module, which provides utilities +for managing storage during the boot process. The tests focus on verifying that +the correct sequence of storage operations is performed with appropriate delays. +""" + +# pyright: reportAttributeAccessIssue=false + +import os +import shutil +import tempfile +from unittest.mock import MagicMock, patch + +import pytest + +from pysquared.boot.filesystem import mkdir + + +@pytest.fixture(autouse=True) +def test_dir(): + """Sets up a temporary directory for testing and cleans it up afterwards.""" + temp_dir = tempfile.mkdtemp() + + test_dir = os.path.join(temp_dir, "mytestdir") + yield test_dir + + shutil.rmtree(temp_dir, ignore_errors=True) + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +def test_mkdir_success( + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, + capsys: pytest.CaptureFixture[str], +) -> None: + """Test the mkdir function for successful directory creation.""" + mkdir(test_dir, 0.02) + + # Verify each expected call was made + mock_storage.disable_usb_drive.assert_called_once() + mock_storage.remount.assert_called_once_with("/", False) + mock_storage.enable_usb_drive.assert_called_once() + + assert mock_sleep.call_count == 3 + mock_sleep.assert_called_with(0.02) + + # Verify correct print messages + captured = capsys.readouterr() + assert "Disabled USB drive" in captured.out + assert "Remounted root filesystem" in captured.out + assert f"Mount point {test_dir} created." in captured.out + assert "Enabled USB drive" in captured.out + + # directory exists + assert os.path.exists(test_dir) + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +@patch("pysquared.boot.filesystem.os.mkdir") +def test_mkdir_directory_already_exists( + mock_os_mkdir: MagicMock, + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, + capsys: pytest.CaptureFixture[str], +) -> None: + """Test the mkdir function when directory already exists.""" + mock_os_mkdir.side_effect = OSError("Directory already exists") + + mkdir(test_dir, 0.02) + + # Verify all storage operations still performed + mock_storage.disable_usb_drive.assert_called_once() + mock_storage.remount.assert_called_once_with("/", False) + mock_os_mkdir.assert_called_once_with(test_dir) + mock_storage.enable_usb_drive.assert_called_once() + + # Verify correct print messages + captured = capsys.readouterr() + assert "Disabled USB drive" in captured.out + assert "Remounted root filesystem" in captured.out + assert f"Mount point {test_dir} already exists." in captured.out + assert "Enabled USB drive" in captured.out + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +@patch("pysquared.boot.filesystem.os.mkdir") +def test_mkdir_custom_delay( + mock_os_mkdir: MagicMock, + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, +) -> None: + """Test the mkdir function with custom storage action delay.""" + custom_delay = 0.05 + mkdir(test_dir, custom_delay) + + # Verify sleep was called with custom delay + assert mock_sleep.call_count == 3 + mock_sleep.assert_called_with(custom_delay) + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +@patch("pysquared.boot.filesystem.os.mkdir") +def test_mkdir_default_delay( + mock_os_mkdir: MagicMock, + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, +) -> None: + """Test the mkdir function with default storage action delay.""" + mkdir(test_dir) # No delay parameter provided + + # Verify sleep was called with default delay + assert mock_sleep.call_count == 3 + mock_sleep.assert_called_with(0.02) + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +@patch("pysquared.boot.filesystem.os.mkdir") +def test_mkdir_storage_disable_exception( + mock_os_mkdir: MagicMock, + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, +) -> None: + """Test mkdir function when storage.disable_usb_drive() raises an exception.""" + mock_storage.disable_usb_drive.side_effect = Exception("USB disable failed") + + with pytest.raises(Exception, match="USB disable failed"): + mkdir(test_dir, 0.02) + + # Verify that enable_usb_drive is still called in finally block + mock_storage.enable_usb_drive.assert_called_once() + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +def test_mkdir_storage_remount_exception( + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, +) -> None: + """Test mkdir function when storage.remount() raises an exception.""" + mock_storage.remount.side_effect = Exception("Remount failed") + + with pytest.raises(Exception, match="Remount failed"): + mkdir(test_dir, 0.02) + + # Verify that enable_usb_drive is still called in finally block + mock_storage.enable_usb_drive.assert_called_once() + + +@patch("pysquared.boot.filesystem.storage") +@patch("pysquared.boot.filesystem.time.sleep") +@patch("pysquared.boot.filesystem.os.mkdir") +def test_mkdir_mkdir_exception_not_oserror( + mock_os_mkdir: MagicMock, + mock_sleep: MagicMock, + mock_storage: MagicMock, + test_dir: str, +) -> None: + """Test mkdir function when os.mkdir() raises a non-OSError exception.""" + mock_os_mkdir.side_effect = ValueError("Invalid path") + + with pytest.raises(ValueError, match="Invalid path"): + mkdir(test_dir, 0.02) + + # Verify that enable_usb_drive is still called in finally block + mock_storage.enable_usb_drive.assert_called_once() From 4614fffce76c10934c474b3d104f31e33a9cd99a Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 16:09:57 -0500 Subject: [PATCH 03/12] Add tests for logger.py --- pysquared/logger.py | 22 ++++------ tests/unit/test_logger.py | 86 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 92 insertions(+), 16 deletions(-) diff --git a/pysquared/logger.py b/pysquared/logger.py index 99fa8b45..5bbca1c4 100644 --- a/pysquared/logger.py +++ b/pysquared/logger.py @@ -101,9 +101,12 @@ def __init__( if log_dir is not None: try: # Octal number 0o040000 is the stat mode indicating the file being stat'd is a directory - _is_directory: int = 0o040000 - if os.stat(log_dir)[0] != _is_directory: - raise ValueError("Logging path must be a directory.") + directory_mode: int = 0o040000 + st_mode = os.stat(log_dir)[0] + if st_mode != directory_mode: + raise ValueError( + f"Logging path must be a directory, received {st_mode}." + ) except OSError as e: raise ValueError("Invalid logging path.") from e @@ -164,18 +167,7 @@ def _log(self, level: str, level_value: int, message: str, **kwargs) -> None: json_order.update(kwargs) - try: - json_output = json.dumps(json_order) - except TypeError as e: - json_output = json.dumps( - OrderedDict( - [ - ("time", asctime), - ("level", "ERROR"), - ("msg", f"Failed to serialize log message: {e}"), - ] - ), - ) + json_output = json.dumps(json_order) if self._can_print_this_level(level_value): if self._log_dir is not None: diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index 152f15d6..a5a9330b 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -4,7 +4,12 @@ functionality with different severity levels, colorized output, and error counting. """ -from unittest.mock import MagicMock +# pyright: reportAttributeAccessIssue=false + +import os +import shutil +import tempfile +from unittest.mock import MagicMock, patch import pytest from microcontroller import Pin @@ -13,6 +18,14 @@ from pysquared.nvm.counter import Counter +@pytest.fixture +def temp_dir(): + """Sets up a temporary directory for testing and cleans it up afterwards.""" + temp_dir = tempfile.mkdtemp() + yield temp_dir + shutil.rmtree(temp_dir, ignore_errors=True) + + @pytest.fixture def logger(): """Provides a Logger instance for testing without colorization.""" @@ -27,6 +40,13 @@ def logger_color(): return Logger(error_counter=count, colorized=True) +@pytest.fixture +def logger_file(): + """Provides a Logger instance for testing with file logging.""" + count = MagicMock(spec=Counter) + return Logger(error_counter=count, log_dir=temp_dir()) + + def test_debug_log(capsys, logger): """Tests logging a debug message without colorization. @@ -282,3 +302,67 @@ def test_invalid_json_type_pin(capsys, logger): logger.debug("Initializing watchdog", pin=mock_pin) captured = capsys.readouterr() assert "TypeError" not in captured.out + + +@patch("pysquared.logger.os.stat") +def test_logger_init_with_valid_directory( + mock_os_stat: MagicMock, +): + """Tests Logger initialization with a valid directory for log_dir.""" + count = MagicMock(spec=Counter) + + mock_os_stat.return_value = [0o040000] + + logger = Logger(error_counter=count, log_dir="placeholder") + assert logger._log_dir == "placeholder" + + +@patch("pysquared.logger.os.stat") +def test_logger_init_with_not_a_directory( + mock_os_stat: MagicMock, +): + """Tests Logger initialization with filesystem object that is not a directory for log_dir.""" + count = MagicMock(spec=Counter) + + mock_os_stat.return_value = [1234] + + with pytest.raises(ValueError): + _ = Logger(error_counter=count, log_dir="placeholder") + + +@patch("pysquared.logger.os.stat") +def test_logger_init_with_invalid_directory( + mock_os_stat: MagicMock, +): + """Tests Logger initialization with invalid directory for log_dir.""" + count = MagicMock(spec=Counter) + + with pytest.raises(ValueError): + mock_os_stat.side_effect = OSError("Stat failed") + _ = Logger(error_counter=count, log_dir="placeholder") + + +@patch("pysquared.logger.os.stat") +def test_log_to_file(mock_os_stat: MagicMock): + """Tests logging messages to a file.""" + count = MagicMock(spec=Counter) + + mock_os_stat.return_value = [0o040000] + + with tempfile.TemporaryDirectory() as temp_dir: + log_path = os.path.join(temp_dir, "poke") + os.mkdir(log_path) + logger = Logger(error_counter=count, log_dir=log_path) + logger.info("Aaron Siemsen rocks") + + with open(os.path.join(log_path, "activity.log"), "r") as f: + contents = f.read() + assert "Aaron Siemsen rocks" in contents + + +def test_get_error_count(): + """Tests retrieving the error count from the Logger.""" + count = MagicMock(spec=Counter) + count.get.return_value = 5 + logger = Logger(error_counter=count) + assert logger.get_error_count() == 5 From f0cb4006f448aa265054ee40e48df7a4925cfc3e Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 16:27:33 -0500 Subject: [PATCH 04/12] Clean up test_logger --- tests/unit/test_logger.py | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index a5a9330b..8abf3f1e 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -7,7 +7,6 @@ # pyright: reportAttributeAccessIssue=false import os -import shutil import tempfile from unittest.mock import MagicMock, patch @@ -18,14 +17,6 @@ from pysquared.nvm.counter import Counter -@pytest.fixture -def temp_dir(): - """Sets up a temporary directory for testing and cleans it up afterwards.""" - temp_dir = tempfile.mkdtemp() - yield temp_dir - shutil.rmtree(temp_dir, ignore_errors=True) - - @pytest.fixture def logger(): """Provides a Logger instance for testing without colorization.""" @@ -40,13 +31,6 @@ def logger_color(): return Logger(error_counter=count, colorized=True) -@pytest.fixture -def logger_file(): - """Provides a Logger instance for testing with file logging.""" - count = MagicMock(spec=Counter) - return Logger(error_counter=count, log_dir=temp_dir()) - - def test_debug_log(capsys, logger): """Tests logging a debug message without colorization. From f600b4a22219d759579e4a2a5cd3eae3bda615d5 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 16:34:13 -0500 Subject: [PATCH 05/12] Fix call to time.sleep --- pysquared/boot/filesystem.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pysquared/boot/filesystem.py b/pysquared/boot/filesystem.py index 0593035c..c8b8ad55 100644 --- a/pysquared/boot/filesystem.py +++ b/pysquared/boot/filesystem.py @@ -33,7 +33,7 @@ def mkdir( """ try: storage.disable_usb_drive() - time.sleep(seconds=storage_action_delay) + time.sleep(storage_action_delay) print("Disabled USB drive") storage.remount("/", False) From 62f5a688eb781ca283b8818686235ffe81495184 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 17:10:09 -0500 Subject: [PATCH 06/12] Initial sd_card test --- mocks/circuitpython/storage.py | 25 +++++++++++++++++++++++++ pysquared/sd_card.py | 11 ++++++++--- tests/unit/boot/test_filesystem.py | 8 ++++---- 3 files changed, 37 insertions(+), 7 deletions(-) diff --git a/mocks/circuitpython/storage.py b/mocks/circuitpython/storage.py index beaf59ff..ef3c8fb0 100644 --- a/mocks/circuitpython/storage.py +++ b/mocks/circuitpython/storage.py @@ -5,6 +5,23 @@ module without the need for actual hardware. """ +from __future__ import annotations + +from circuitpython_typing import BlockDevice + +# from storage import VfsFat + + +def mount(filesystem: VfsFat, mount_path: str, *, readonly: bool = False) -> None: + """A mock function to mount the filesystem. + + Args: + filesystem: The filesystem to mount. + mount_path: Where to mount the filesystem. + readonly: True when the filesystem should be readonly to CircuitPython. + """ + pass + def disable_usb_drive() -> None: """A mock function to disable the USB drive.""" @@ -24,3 +41,11 @@ def remount(path: str, readonly: bool) -> None: readonly: Whether to mount as read-only. """ pass + + +class VfsFat: + """A mock class representing a FAT filesystem.""" + + def __init__(self, block_device: BlockDevice) -> None: + """Initializes the mock FAT filesystem.""" + pass diff --git a/pysquared/sd_card.py b/pysquared/sd_card.py index a69a0edf..07d3d216 100644 --- a/pysquared/sd_card.py +++ b/pysquared/sd_card.py @@ -1,10 +1,15 @@ """This module provides a SD Card class to manipulate the sd card filesystem""" -import sdcardio -import storage from busio import SPI from microcontroller import Pin +try: + import sdcardio + import storage +except ImportError: + import mocks.circuitpython.sdcardio as sdcardio + import mocks.circuitpython.storage as storage + from .hardware.exception import HardwareInitializationError @@ -24,5 +29,5 @@ def __init__( sd = sdcardio.SDCard(spi_bus, chip_select, baudrate) vfs = storage.VfsFat(sd) # type: ignore # Issue: https://github.com/adafruit/Adafruit_CircuitPython_Typing/issues/51 storage.mount(vfs, mount_path) - except (OSError, ValueError) as e: + except Exception as e: raise HardwareInitializationError("Failed to initialize SD Card") from e diff --git a/tests/unit/boot/test_filesystem.py b/tests/unit/boot/test_filesystem.py index 4a9ae616..35118653 100644 --- a/tests/unit/boot/test_filesystem.py +++ b/tests/unit/boot/test_filesystem.py @@ -1,8 +1,8 @@ -"""Unit tests for the storage module. +"""Unit tests for the filesystem class. -This module contains unit tests for the `storage` module, which provides utilities -for managing storage during the boot process. The tests focus on verifying that -the correct sequence of storage operations is performed with appropriate delays. +This file contains unit tests for the `filesystem` class, which provides utilities +for managing the filesystem during the boot process. The tests focus on verifying that +the correct sequence of filesystem operations. """ # pyright: reportAttributeAccessIssue=false From c9840657471cfcdb3d4bb67b0ad3f3214df505bb Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:05:21 -0500 Subject: [PATCH 07/12] Add tests for SD Card --- mocks/circuitpython/storage.py | 6 +- pyproject.toml | 23 +-- pysquared/sd_card.py | 33 ---- tests/unit/hardware/test_burnwire.py | 227 --------------------------- 4 files changed, 14 insertions(+), 275 deletions(-) delete mode 100644 pysquared/sd_card.py delete mode 100644 tests/unit/hardware/test_burnwire.py diff --git a/mocks/circuitpython/storage.py b/mocks/circuitpython/storage.py index ef3c8fb0..27d0ea62 100644 --- a/mocks/circuitpython/storage.py +++ b/mocks/circuitpython/storage.py @@ -9,8 +9,6 @@ from circuitpython_typing import BlockDevice -# from storage import VfsFat - def mount(filesystem: VfsFat, mount_path: str, *, readonly: bool = False) -> None: """A mock function to mount the filesystem. @@ -44,8 +42,8 @@ def remount(path: str, readonly: bool) -> None: class VfsFat: - """A mock class representing a FAT filesystem.""" + """A mock class representing a VfsFat filesystem.""" def __init__(self, block_device: BlockDevice) -> None: - """Initializes the mock FAT filesystem.""" + """Initializes the mock VfsFat filesystem.""" pass diff --git a/pyproject.toml b/pyproject.toml index 84aeae29..976ff0f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,25 +47,26 @@ packages = [ "pysquared", "pysquared.boot", "pysquared.config", - "pysquared.hardware", - "pysquared.hardware.imu", + "pysquared.hardware.burnwire.manager", + "pysquared.hardware.burnwire", "pysquared.hardware.imu.manager", - "pysquared.hardware.magnetometer", + "pysquared.hardware.imu", + "pysquared.hardware.light_sensor.manager", "pysquared.hardware.magnetometer.manager", - "pysquared.hardware.radio", + "pysquared.hardware.magnetometer", + "pysquared.hardware.power_monitor.manager", + "pysquared.hardware.power_monitor", "pysquared.hardware.radio.manager", "pysquared.hardware.radio.packetizer", - "pysquared.hardware.power_monitor", - "pysquared.hardware.power_monitor.manager", - "pysquared.hardware.temperature_sensor", + "pysquared.hardware.radio", + "pysquared.hardware.sd_card", "pysquared.hardware.temperature_sensor.manager", - "pysquared.hardware.burnwire", - "pysquared.hardware.burnwire.manager", - "pysquared.hardware.light_sensor.manager", + "pysquared.hardware.temperature_sensor", + "pysquared.hardware", "pysquared.nvm", "pysquared.protos", - "pysquared.rtc", "pysquared.rtc.manager", + "pysquared.rtc", "pysquared.sensor_reading" ] diff --git a/pysquared/sd_card.py b/pysquared/sd_card.py deleted file mode 100644 index 07d3d216..00000000 --- a/pysquared/sd_card.py +++ /dev/null @@ -1,33 +0,0 @@ -"""This module provides a SD Card class to manipulate the sd card filesystem""" - -from busio import SPI -from microcontroller import Pin - -try: - import sdcardio - import storage -except ImportError: - import mocks.circuitpython.sdcardio as sdcardio - import mocks.circuitpython.storage as storage - -from .hardware.exception import HardwareInitializationError - - -class SDCardManager: - """Class providing various functionalities related to USB and SD card operations.""" - - """Initializing class, remounting storage, and initializing SD card""" - - def __init__( - self, - spi_bus: SPI, - chip_select: Pin, - baudrate: int = 400000, - mount_path: str = "/sd", - ) -> None: - try: - sd = sdcardio.SDCard(spi_bus, chip_select, baudrate) - vfs = storage.VfsFat(sd) # type: ignore # Issue: https://github.com/adafruit/Adafruit_CircuitPython_Typing/issues/51 - storage.mount(vfs, mount_path) - except Exception as e: - raise HardwareInitializationError("Failed to initialize SD Card") from e diff --git a/tests/unit/hardware/test_burnwire.py b/tests/unit/hardware/test_burnwire.py deleted file mode 100644 index 89d6c0e8..00000000 --- a/tests/unit/hardware/test_burnwire.py +++ /dev/null @@ -1,227 +0,0 @@ -"""Unit tests for the BurnwireManager class. - -This module contains unit tests for the `BurnwireManager` class, which controls -the activation of burnwires. The tests cover initialization, successful burn -operations, error handling, and cleanup procedures. -""" - -# pyright: reportAttributeAccessIssue=false, reportOptionalMemberAccess=false, reportFunctionMemberAccess=false - -from unittest.mock import ANY, MagicMock, patch - -import pytest -from digitalio import DigitalInOut - -from pysquared.hardware.burnwire.manager.burnwire import BurnwireManager -from pysquared.logger import Logger - - -@pytest.fixture -def mock_logger(): - """Mocks the Logger class.""" - return MagicMock(spec=Logger) - - -@pytest.fixture -def mock_enable_burn(): - """Mocks the DigitalInOut pin for enabling the burnwire.""" - return MagicMock(spec=DigitalInOut) - - -@pytest.fixture -def mock_fire_burn(): - """Mocks the DigitalInOut pin for firing the burnwire.""" - return MagicMock(spec=DigitalInOut) - - -@pytest.fixture -def burnwire_manager(mock_logger, mock_enable_burn, mock_fire_burn): - """Provides a BurnwireManager instance for testing.""" - return BurnwireManager( - logger=mock_logger, - enable_burn=mock_enable_burn, - fire_burn=mock_fire_burn, - enable_logic=True, - ) - - -def test_burnwire_initialization_default_logic( - mock_logger, mock_enable_burn, mock_fire_burn -): - """Tests burnwire initialization with default enable_logic=True. - - Args: - mock_logger: Mocked Logger instance. - mock_enable_burn: Mocked enable_burn pin. - mock_fire_burn: Mocked fire_burn pin. - """ - manager = BurnwireManager(mock_logger, mock_enable_burn, mock_fire_burn) - assert manager._enable_logic is True - assert manager.number_of_attempts == 0 - - -def test_burnwire_initialization_inverted_logic( - mock_logger, mock_enable_burn, mock_fire_burn -): - """Tests burnwire initialization with enable_logic=False. - - Args: - mock_logger: Mocked Logger instance. - mock_enable_burn: Mocked enable_burn pin. - mock_fire_burn: Mocked fire_burn pin. - """ - manager = BurnwireManager( - mock_logger, mock_enable_burn, mock_fire_burn, enable_logic=False - ) - assert manager._enable_logic is False - assert manager.number_of_attempts == 0 - - -def test_successful_burn(burnwire_manager): - """Tests a successful burnwire activation. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - with patch("time.sleep") as mock_sleep: - result = burnwire_manager.burn(timeout_duration=1.0) - - mock_sleep.assert_any_call(0.1) # Verify stabilization delay - mock_sleep.assert_any_call(1.0) # Verify burn duration - - # Verify final safe state - assert burnwire_manager._fire_burn.value == (not burnwire_manager._enable_logic) - assert burnwire_manager._enable_burn.value == ( - not burnwire_manager._enable_logic - ) - - assert result is True - assert burnwire_manager.number_of_attempts == 1 - - -def test_burn_error_handling(burnwire_manager): - """Tests error handling during burnwire activation. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - # Mock the enable_burn pin to raise an exception when setting value - type(burnwire_manager._enable_burn).value = property( - fset=MagicMock(side_effect=RuntimeError("Hardware failure")) - ) - - result = burnwire_manager.burn() - - assert result is False - - # Verify critical log call about burn failure - assert burnwire_manager._log.critical.call_count == 2 - calls = [call[0][0] for call in burnwire_manager._log.critical.call_args_list] - assert any("Failed!" in msg for msg in calls) - - -def test_cleanup_on_error(burnwire_manager): - """Tests that cleanup occurs even when an error happens during burn. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - with patch("time.sleep") as mock_sleep: - mock_sleep.side_effect = RuntimeError("Unexpected error") - - result = burnwire_manager.burn() - - assert result is False - # Verify pins are set to safe state even after error - assert burnwire_manager._fire_burn.value == (not burnwire_manager._enable_logic) - assert burnwire_manager._enable_burn.value == ( - not burnwire_manager._enable_logic - ) - burnwire_manager._log.debug.assert_any_call("Burnwire Safed") - - -def test_attempt_burn_exception_handling(burnwire_manager): - """Tests that _attempt_burn properly handles and propagates exceptions. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - # Mock the enable_burn pin to raise an exception when setting value - type(burnwire_manager._enable_burn).value = property( - fset=MagicMock(side_effect=RuntimeError("Hardware failure")) - ) - - with pytest.raises(RuntimeError) as exc_info: - burnwire_manager._attempt_burn() - - assert "Failed to set fire_burn pin" in str(exc_info.value) - - -def test_burn_keyboard_interrupt(burnwire_manager): - """Tests that a KeyboardInterrupt during burn is handled and logged, including in _attempt_burn. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - # Patch _attempt_burn to raise KeyboardInterrupt - with patch.object(burnwire_manager, "_attempt_burn", side_effect=KeyboardInterrupt): - result = burnwire_manager.burn(timeout_duration=1.0) - assert result is False - # Check that the log contains the interruption message from burn() - found = any( - "Burn Attempt Interrupted after" in str(call[0][0]) - for call in burnwire_manager._log.debug.call_args_list - ) - assert found - - # Patch _enable to raise KeyboardInterrupt as if from inside _attempt_burn - with patch.object(burnwire_manager, "_enable", side_effect=KeyboardInterrupt): - with patch.object(burnwire_manager._log, "warning") as mock_warning: - with pytest.raises(KeyboardInterrupt): - burnwire_manager._attempt_burn() - # Should log the warning from _attempt_burn - mock_warning.assert_called_once() - assert "Interrupted" in mock_warning.call_args[0][0] - - -def test_enable_fire_burn_pin_error(burnwire_manager): - """Tests that a RuntimeError is raised if setting fire_burn pin fails in _enable. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - # Allow enable_burn to succeed - burnwire_manager._enable_burn.value = burnwire_manager._enable_logic - # Make fire_burn raise an exception when set - type(burnwire_manager._fire_burn).value = property( - fset=MagicMock(side_effect=Exception("fire_burn failure")) - ) - with pytest.raises(RuntimeError) as exc_info: - burnwire_manager._enable() - assert str(exc_info.value) == "Failed to set fire_burn pin" - - -def test_disable_cleanup_critical_log(burnwire_manager): - """Tests that a critical log is made if _disable fails during cleanup and no prior error occurred. - - Args: - burnwire_manager: BurnwireManager instance for testing. - """ - # Patch _enable to succeed - with patch.object(burnwire_manager, "_enable", return_value=None): - # Patch time.sleep to avoid delay - with patch("time.sleep"): - # Patch _disable to raise an Exception - with patch.object( - burnwire_manager, "_disable", side_effect=Exception("disable failed") - ): - # Patch _log.critical to track calls - with patch.object(burnwire_manager._log, "critical") as mock_critical: - # Patch _fire_burn and _enable_burn to allow value assignment - burnwire_manager._fire_burn.value = False - burnwire_manager._enable_burn.value = False - # The error variable should be None, so critical should be called - burnwire_manager._attempt_burn() - mock_critical.assert_called_with( - "Failed to safe burnwire pins!", ANY - ) From 1636f848c85c3f6d76b58f1c5ce50745ad331e64 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:10:24 -0500 Subject: [PATCH 08/12] Readd removed files --- .pre-commit-config.yaml | 2 +- mocks/circuitpython/sdcardio.py | 20 ++ pysquared/hardware/sd_card/__init__.py | 3 + pysquared/hardware/sd_card/sd_card.py | 28 +++ tests/unit/hardware/burnwire/test_burnwire.py | 227 ++++++++++++++++++ tests/unit/hardware/sd_card/test_sd_card.py | 72 ++++++ 6 files changed, 351 insertions(+), 1 deletion(-) create mode 100644 mocks/circuitpython/sdcardio.py create mode 100644 pysquared/hardware/sd_card/__init__.py create mode 100644 pysquared/hardware/sd_card/sd_card.py create mode 100644 tests/unit/hardware/burnwire/test_burnwire.py create mode 100644 tests/unit/hardware/sd_card/test_sd_card.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 01b58816..aad6be86 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,7 +22,7 @@ repos: language: pygrep types: [python] files: ^pysquared/ - exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/sd_card\.py + exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/hardware/sd_card/sd_card\.py - repo: local hooks: diff --git a/mocks/circuitpython/sdcardio.py b/mocks/circuitpython/sdcardio.py new file mode 100644 index 00000000..51ad18a4 --- /dev/null +++ b/mocks/circuitpython/sdcardio.py @@ -0,0 +1,20 @@ +"""Mock for the CircuitPython sdcardio module. + +This module provides a mock implementation of the CircuitPython sdcardio module +for testing purposes. It allows for simulating the behavior of the sdcardio +module without the need for actual hardware. +""" + + +class SDCard: + """A mock class representing an SD card.""" + + def __init__(self, spi, cs, baudrate): + """Initializes the mock SD card. + + Args: + spi: The SPI bus. + cs: The chip select pin. + baudrate: The communication speed. + """ + pass diff --git a/pysquared/hardware/sd_card/__init__.py b/pysquared/hardware/sd_card/__init__.py new file mode 100644 index 00000000..ff1ea6e4 --- /dev/null +++ b/pysquared/hardware/sd_card/__init__.py @@ -0,0 +1,3 @@ +""" +This module provides an interface for controlling SD cards. +""" diff --git a/pysquared/hardware/sd_card/sd_card.py b/pysquared/hardware/sd_card/sd_card.py new file mode 100644 index 00000000..ae7b4292 --- /dev/null +++ b/pysquared/hardware/sd_card/sd_card.py @@ -0,0 +1,28 @@ +"""This module provides a SD Card class to manipulate the sd card filesystem""" + +import sdcardio +import storage +from busio import SPI +from microcontroller import Pin + +from ..exception import HardwareInitializationError + + +class SDCardManager: + """Class providing various functionalities related to USB and SD card operations.""" + + """Initializing class, remounting storage, and initializing SD card""" + + def __init__( + self, + spi_bus: SPI, + chip_select: Pin, + baudrate: int = 400000, + mount_path: str = "/sd", + ) -> None: + try: + sd = sdcardio.SDCard(spi_bus, chip_select, baudrate) + vfs = storage.VfsFat(sd) # type: ignore # Issue: https://github.com/adafruit/Adafruit_CircuitPython_Typing/issues/51 + storage.mount(vfs, mount_path) + except Exception as e: + raise HardwareInitializationError("Failed to initialize SD Card") from e diff --git a/tests/unit/hardware/burnwire/test_burnwire.py b/tests/unit/hardware/burnwire/test_burnwire.py new file mode 100644 index 00000000..89d6c0e8 --- /dev/null +++ b/tests/unit/hardware/burnwire/test_burnwire.py @@ -0,0 +1,227 @@ +"""Unit tests for the BurnwireManager class. + +This module contains unit tests for the `BurnwireManager` class, which controls +the activation of burnwires. The tests cover initialization, successful burn +operations, error handling, and cleanup procedures. +""" + +# pyright: reportAttributeAccessIssue=false, reportOptionalMemberAccess=false, reportFunctionMemberAccess=false + +from unittest.mock import ANY, MagicMock, patch + +import pytest +from digitalio import DigitalInOut + +from pysquared.hardware.burnwire.manager.burnwire import BurnwireManager +from pysquared.logger import Logger + + +@pytest.fixture +def mock_logger(): + """Mocks the Logger class.""" + return MagicMock(spec=Logger) + + +@pytest.fixture +def mock_enable_burn(): + """Mocks the DigitalInOut pin for enabling the burnwire.""" + return MagicMock(spec=DigitalInOut) + + +@pytest.fixture +def mock_fire_burn(): + """Mocks the DigitalInOut pin for firing the burnwire.""" + return MagicMock(spec=DigitalInOut) + + +@pytest.fixture +def burnwire_manager(mock_logger, mock_enable_burn, mock_fire_burn): + """Provides a BurnwireManager instance for testing.""" + return BurnwireManager( + logger=mock_logger, + enable_burn=mock_enable_burn, + fire_burn=mock_fire_burn, + enable_logic=True, + ) + + +def test_burnwire_initialization_default_logic( + mock_logger, mock_enable_burn, mock_fire_burn +): + """Tests burnwire initialization with default enable_logic=True. + + Args: + mock_logger: Mocked Logger instance. + mock_enable_burn: Mocked enable_burn pin. + mock_fire_burn: Mocked fire_burn pin. + """ + manager = BurnwireManager(mock_logger, mock_enable_burn, mock_fire_burn) + assert manager._enable_logic is True + assert manager.number_of_attempts == 0 + + +def test_burnwire_initialization_inverted_logic( + mock_logger, mock_enable_burn, mock_fire_burn +): + """Tests burnwire initialization with enable_logic=False. + + Args: + mock_logger: Mocked Logger instance. + mock_enable_burn: Mocked enable_burn pin. + mock_fire_burn: Mocked fire_burn pin. + """ + manager = BurnwireManager( + mock_logger, mock_enable_burn, mock_fire_burn, enable_logic=False + ) + assert manager._enable_logic is False + assert manager.number_of_attempts == 0 + + +def test_successful_burn(burnwire_manager): + """Tests a successful burnwire activation. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + with patch("time.sleep") as mock_sleep: + result = burnwire_manager.burn(timeout_duration=1.0) + + mock_sleep.assert_any_call(0.1) # Verify stabilization delay + mock_sleep.assert_any_call(1.0) # Verify burn duration + + # Verify final safe state + assert burnwire_manager._fire_burn.value == (not burnwire_manager._enable_logic) + assert burnwire_manager._enable_burn.value == ( + not burnwire_manager._enable_logic + ) + + assert result is True + assert burnwire_manager.number_of_attempts == 1 + + +def test_burn_error_handling(burnwire_manager): + """Tests error handling during burnwire activation. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + # Mock the enable_burn pin to raise an exception when setting value + type(burnwire_manager._enable_burn).value = property( + fset=MagicMock(side_effect=RuntimeError("Hardware failure")) + ) + + result = burnwire_manager.burn() + + assert result is False + + # Verify critical log call about burn failure + assert burnwire_manager._log.critical.call_count == 2 + calls = [call[0][0] for call in burnwire_manager._log.critical.call_args_list] + assert any("Failed!" in msg for msg in calls) + + +def test_cleanup_on_error(burnwire_manager): + """Tests that cleanup occurs even when an error happens during burn. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + with patch("time.sleep") as mock_sleep: + mock_sleep.side_effect = RuntimeError("Unexpected error") + + result = burnwire_manager.burn() + + assert result is False + # Verify pins are set to safe state even after error + assert burnwire_manager._fire_burn.value == (not burnwire_manager._enable_logic) + assert burnwire_manager._enable_burn.value == ( + not burnwire_manager._enable_logic + ) + burnwire_manager._log.debug.assert_any_call("Burnwire Safed") + + +def test_attempt_burn_exception_handling(burnwire_manager): + """Tests that _attempt_burn properly handles and propagates exceptions. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + # Mock the enable_burn pin to raise an exception when setting value + type(burnwire_manager._enable_burn).value = property( + fset=MagicMock(side_effect=RuntimeError("Hardware failure")) + ) + + with pytest.raises(RuntimeError) as exc_info: + burnwire_manager._attempt_burn() + + assert "Failed to set fire_burn pin" in str(exc_info.value) + + +def test_burn_keyboard_interrupt(burnwire_manager): + """Tests that a KeyboardInterrupt during burn is handled and logged, including in _attempt_burn. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + # Patch _attempt_burn to raise KeyboardInterrupt + with patch.object(burnwire_manager, "_attempt_burn", side_effect=KeyboardInterrupt): + result = burnwire_manager.burn(timeout_duration=1.0) + assert result is False + # Check that the log contains the interruption message from burn() + found = any( + "Burn Attempt Interrupted after" in str(call[0][0]) + for call in burnwire_manager._log.debug.call_args_list + ) + assert found + + # Patch _enable to raise KeyboardInterrupt as if from inside _attempt_burn + with patch.object(burnwire_manager, "_enable", side_effect=KeyboardInterrupt): + with patch.object(burnwire_manager._log, "warning") as mock_warning: + with pytest.raises(KeyboardInterrupt): + burnwire_manager._attempt_burn() + # Should log the warning from _attempt_burn + mock_warning.assert_called_once() + assert "Interrupted" in mock_warning.call_args[0][0] + + +def test_enable_fire_burn_pin_error(burnwire_manager): + """Tests that a RuntimeError is raised if setting fire_burn pin fails in _enable. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + # Allow enable_burn to succeed + burnwire_manager._enable_burn.value = burnwire_manager._enable_logic + # Make fire_burn raise an exception when set + type(burnwire_manager._fire_burn).value = property( + fset=MagicMock(side_effect=Exception("fire_burn failure")) + ) + with pytest.raises(RuntimeError) as exc_info: + burnwire_manager._enable() + assert str(exc_info.value) == "Failed to set fire_burn pin" + + +def test_disable_cleanup_critical_log(burnwire_manager): + """Tests that a critical log is made if _disable fails during cleanup and no prior error occurred. + + Args: + burnwire_manager: BurnwireManager instance for testing. + """ + # Patch _enable to succeed + with patch.object(burnwire_manager, "_enable", return_value=None): + # Patch time.sleep to avoid delay + with patch("time.sleep"): + # Patch _disable to raise an Exception + with patch.object( + burnwire_manager, "_disable", side_effect=Exception("disable failed") + ): + # Patch _log.critical to track calls + with patch.object(burnwire_manager._log, "critical") as mock_critical: + # Patch _fire_burn and _enable_burn to allow value assignment + burnwire_manager._fire_burn.value = False + burnwire_manager._enable_burn.value = False + # The error variable should be None, so critical should be called + burnwire_manager._attempt_burn() + mock_critical.assert_called_with( + "Failed to safe burnwire pins!", ANY + ) diff --git a/tests/unit/hardware/sd_card/test_sd_card.py b/tests/unit/hardware/sd_card/test_sd_card.py new file mode 100644 index 00000000..a32c9a69 --- /dev/null +++ b/tests/unit/hardware/sd_card/test_sd_card.py @@ -0,0 +1,72 @@ +"""Unit tests for the filesystem class. + +This file contains unit tests for the `filesystem` class, which provides utilities +for managing the filesystem during the boot process. The tests focus on verifying that +the correct sequence of filesystem operations. +""" + +# pyright: reportAttributeAccessIssue=false + +import sys +from unittest.mock import MagicMock, patch + +import pytest +from busio import SPI +from microcontroller import Pin + +from pysquared.hardware.exception import HardwareInitializationError + +sys.modules["storage"] = MagicMock() +sys.modules["sdcardio"] = MagicMock() + +from pysquared.hardware.sd_card.sd_card import SDCardManager # noqa: E402 + + +@patch("pysquared.hardware.sd_card.sd_card.storage") +@patch("pysquared.hardware.sd_card.sd_card.sdcardio") +def test_sdcard_init_success( + mock_sdcardio: MagicMock, + mock_storage: MagicMock, +) -> None: + """Test SD Card successful initialization.""" + mock_sd_card = MagicMock() + mock_sdcardio.SDCard.return_value = mock_sd_card + + mock_block_device = MagicMock() + mock_storage.VfsFat.return_value = mock_block_device + + spi = MagicMock(spec=SPI) + cs = MagicMock(spec=Pin) + baudrate = 400000 + mount_path = "/sd" + _ = SDCardManager( + spi_bus=spi, + chip_select=cs, + baudrate=baudrate, + mount_path=mount_path, + ) + + # Verify each expected call was made + mock_sdcardio.SDCard.assert_called_once_with(spi, cs, baudrate) + mock_storage.VfsFat.assert_called_once_with(mock_sd_card) + mock_storage.mount.assert_called_once_with(mock_block_device, mount_path) + + +@patch("pysquared.hardware.sd_card.sd_card.storage") +@patch("pysquared.hardware.sd_card.sd_card.sdcardio") +def test_sdcard_init_error( + mock_sdcardio: MagicMock, + mock_storage: MagicMock, +) -> None: + """Test SD Card failing initialization""" + mock_sdcardio.SDCard.side_effect = Exception("Evan (SD) Ortiz") + + with pytest.raises( + HardwareInitializationError, match="Failed to initialize SD Card" + ): + _ = SDCardManager( + spi_bus=MagicMock(spec=SPI), + chip_select=MagicMock(spec=Pin), + baudrate=400000, + mount_path="/sd", + ) From 8f35601b20cd94f9073d439c4bd67ec8901be019 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:13:29 -0500 Subject: [PATCH 09/12] Remove unused mocks --- mocks/circuitpython/sdcardio.py | 20 -------------------- mocks/circuitpython/storage.py | 21 --------------------- 2 files changed, 41 deletions(-) delete mode 100644 mocks/circuitpython/sdcardio.py diff --git a/mocks/circuitpython/sdcardio.py b/mocks/circuitpython/sdcardio.py deleted file mode 100644 index 51ad18a4..00000000 --- a/mocks/circuitpython/sdcardio.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Mock for the CircuitPython sdcardio module. - -This module provides a mock implementation of the CircuitPython sdcardio module -for testing purposes. It allows for simulating the behavior of the sdcardio -module without the need for actual hardware. -""" - - -class SDCard: - """A mock class representing an SD card.""" - - def __init__(self, spi, cs, baudrate): - """Initializes the mock SD card. - - Args: - spi: The SPI bus. - cs: The chip select pin. - baudrate: The communication speed. - """ - pass diff --git a/mocks/circuitpython/storage.py b/mocks/circuitpython/storage.py index 27d0ea62..e618f1aa 100644 --- a/mocks/circuitpython/storage.py +++ b/mocks/circuitpython/storage.py @@ -7,19 +7,6 @@ from __future__ import annotations -from circuitpython_typing import BlockDevice - - -def mount(filesystem: VfsFat, mount_path: str, *, readonly: bool = False) -> None: - """A mock function to mount the filesystem. - - Args: - filesystem: The filesystem to mount. - mount_path: Where to mount the filesystem. - readonly: True when the filesystem should be readonly to CircuitPython. - """ - pass - def disable_usb_drive() -> None: """A mock function to disable the USB drive.""" @@ -39,11 +26,3 @@ def remount(path: str, readonly: bool) -> None: readonly: Whether to mount as read-only. """ pass - - -class VfsFat: - """A mock class representing a VfsFat filesystem.""" - - def __init__(self, block_device: BlockDevice) -> None: - """Initializes the mock VfsFat filesystem.""" - pass From 5ab4ef664f2661ca67585d7f5979a91e16aca529 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:14:24 -0500 Subject: [PATCH 10/12] Remove annotation import --- mocks/circuitpython/storage.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/mocks/circuitpython/storage.py b/mocks/circuitpython/storage.py index e618f1aa..beaf59ff 100644 --- a/mocks/circuitpython/storage.py +++ b/mocks/circuitpython/storage.py @@ -5,8 +5,6 @@ module without the need for actual hardware. """ -from __future__ import annotations - def disable_usb_drive() -> None: """A mock function to disable the USB drive.""" From 04b55a37e0dbc14cf6d83c904511034021d74d67 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:38:39 -0500 Subject: [PATCH 11/12] Reorganize --- .pre-commit-config.yaml | 2 +- pyproject.toml | 1 + pysquared/hardware/sd_card/{ => manager}/__init__.py | 0 pysquared/hardware/sd_card/{ => manager}/sd_card.py | 2 +- .../hardware/burnwire/{ => manager}/test_burnwire.py | 0 .../hardware/sd_card/{ => manager}/test_sd_card.py | 10 +++++----- 6 files changed, 8 insertions(+), 7 deletions(-) rename pysquared/hardware/sd_card/{ => manager}/__init__.py (100%) rename pysquared/hardware/sd_card/{ => manager}/sd_card.py (94%) rename tests/unit/hardware/burnwire/{ => manager}/test_burnwire.py (100%) rename tests/unit/hardware/sd_card/{ => manager}/test_sd_card.py (85%) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index aad6be86..e27d4199 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,7 +22,7 @@ repos: language: pygrep types: [python] files: ^pysquared/ - exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/hardware/sd_card/sd_card\.py + exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/hardware/sd_card/manager/sd_card\.py - repo: local hooks: diff --git a/pyproject.toml b/pyproject.toml index 976ff0f9..ee3ed6da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,7 @@ packages = [ "pysquared.hardware.radio.manager", "pysquared.hardware.radio.packetizer", "pysquared.hardware.radio", + "pysquared.hardware.sd_card.manager", "pysquared.hardware.sd_card", "pysquared.hardware.temperature_sensor.manager", "pysquared.hardware.temperature_sensor", diff --git a/pysquared/hardware/sd_card/__init__.py b/pysquared/hardware/sd_card/manager/__init__.py similarity index 100% rename from pysquared/hardware/sd_card/__init__.py rename to pysquared/hardware/sd_card/manager/__init__.py diff --git a/pysquared/hardware/sd_card/sd_card.py b/pysquared/hardware/sd_card/manager/sd_card.py similarity index 94% rename from pysquared/hardware/sd_card/sd_card.py rename to pysquared/hardware/sd_card/manager/sd_card.py index ae7b4292..adced9ae 100644 --- a/pysquared/hardware/sd_card/sd_card.py +++ b/pysquared/hardware/sd_card/manager/sd_card.py @@ -5,7 +5,7 @@ from busio import SPI from microcontroller import Pin -from ..exception import HardwareInitializationError +from ...exception import HardwareInitializationError class SDCardManager: diff --git a/tests/unit/hardware/burnwire/test_burnwire.py b/tests/unit/hardware/burnwire/manager/test_burnwire.py similarity index 100% rename from tests/unit/hardware/burnwire/test_burnwire.py rename to tests/unit/hardware/burnwire/manager/test_burnwire.py diff --git a/tests/unit/hardware/sd_card/test_sd_card.py b/tests/unit/hardware/sd_card/manager/test_sd_card.py similarity index 85% rename from tests/unit/hardware/sd_card/test_sd_card.py rename to tests/unit/hardware/sd_card/manager/test_sd_card.py index a32c9a69..a2322d63 100644 --- a/tests/unit/hardware/sd_card/test_sd_card.py +++ b/tests/unit/hardware/sd_card/manager/test_sd_card.py @@ -19,11 +19,11 @@ sys.modules["storage"] = MagicMock() sys.modules["sdcardio"] = MagicMock() -from pysquared.hardware.sd_card.sd_card import SDCardManager # noqa: E402 +from pysquared.hardware.sd_card.manager.sd_card import SDCardManager # noqa: E402 -@patch("pysquared.hardware.sd_card.sd_card.storage") -@patch("pysquared.hardware.sd_card.sd_card.sdcardio") +@patch("pysquared.hardware.sd_card.manager.sd_card.storage") +@patch("pysquared.hardware.sd_card.manager.sd_card.sdcardio") def test_sdcard_init_success( mock_sdcardio: MagicMock, mock_storage: MagicMock, @@ -52,8 +52,8 @@ def test_sdcard_init_success( mock_storage.mount.assert_called_once_with(mock_block_device, mount_path) -@patch("pysquared.hardware.sd_card.sd_card.storage") -@patch("pysquared.hardware.sd_card.sd_card.sdcardio") +@patch("pysquared.hardware.sd_card.manager.sd_card.storage") +@patch("pysquared.hardware.sd_card.manager.sd_card.sdcardio") def test_sdcard_init_error( mock_sdcardio: MagicMock, mock_storage: MagicMock, From 358960ad44032c53f56d7fa25d1b625c0516f311 Mon Sep 17 00:00:00 2001 From: Nate Gay Date: Mon, 1 Sep 2025 18:52:02 -0500 Subject: [PATCH 12/12] Use setter instead of constructor arg to set log dir --- pysquared/logger.py | 40 ++++++++++++++++++++++++--------------- tests/unit/test_logger.py | 18 +++++++++++++----- 2 files changed, 38 insertions(+), 20 deletions(-) diff --git a/pysquared/logger.py b/pysquared/logger.py index 5bbca1c4..20e08678 100644 --- a/pysquared/logger.py +++ b/pysquared/logger.py @@ -77,11 +77,12 @@ class LogLevel: class Logger: """Handles logging messages with different severity levels.""" + _log_dir: str | None = None + def __init__( self, error_counter: Counter, log_level: int = LogLevel.NOTSET, - log_dir: str | None = None, colorized: bool = False, ) -> None: """ @@ -90,26 +91,12 @@ def __init__( Args: error_counter: Counter for error occurrences. log_level: Initial log level. - log_dir: Directory to save log files. If None, logs are not saved to a file. colorized: Whether to colorize output. """ self._error_counter: Counter = error_counter self._log_level: int = log_level - self._log_dir: str | None = log_dir self._colorized: bool = colorized - if log_dir is not None: - try: - # Octal number 0o040000 is the stat mode indicating the file being stat'd is a directory - directory_mode: int = 0o040000 - st_mode = os.stat(log_dir)[0] - if st_mode != directory_mode: - raise ValueError( - f"Logging path must be a directory, received {st_mode}." - ) - except OSError as e: - raise ValueError("Invalid logging path.") from e - def _can_print_this_level(self, level_value: int) -> bool: """ Checks if the message should be printed based on the log level. @@ -246,3 +233,26 @@ def get_error_count(self) -> int: int: The number of errors logged. """ return self._error_counter.get() + + def set_log_dir(self, log_dir: str) -> None: + """ + Sets the log directory for file logging. + + Args: + log_dir (str): Directory to save log files. + + Raises: + ValueError: If the provided path is not a valid directory. + """ + try: + # Octal number 0o040000 is the stat mode indicating the file being stat'd is a directory + directory_mode: int = 0o040000 + st_mode = os.stat(log_dir)[0] + if st_mode != directory_mode: + raise ValueError( + f"Logging path must be a directory, received {st_mode}." + ) + except OSError as e: + raise ValueError("Invalid logging path.") from e + + self._log_dir = log_dir diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index 8abf3f1e..1dc790f0 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -297,7 +297,8 @@ def test_logger_init_with_valid_directory( mock_os_stat.return_value = [0o040000] - logger = Logger(error_counter=count, log_dir="placeholder") + logger = Logger(error_counter=count) + logger.set_log_dir("placeholder") assert logger._log_dir == "placeholder" @@ -310,8 +311,10 @@ def test_logger_init_with_not_a_directory( mock_os_stat.return_value = [1234] + logger = Logger(error_counter=count) + with pytest.raises(ValueError): - _ = Logger(error_counter=count, log_dir="placeholder") + logger.set_log_dir("placeholder") @patch("pysquared.logger.os.stat") @@ -321,9 +324,12 @@ def test_logger_init_with_invalid_directory( """Tests Logger initialization with invalid directory for log_dir.""" count = MagicMock(spec=Counter) + mock_os_stat.side_effect = OSError("Stat failed") + + logger = Logger(error_counter=count) + with pytest.raises(ValueError): - mock_os_stat.side_effect = OSError("Stat failed") - _ = Logger(error_counter=count, log_dir="placeholder") + logger.set_log_dir("placeholder") @patch("pysquared.logger.os.stat") @@ -333,10 +339,12 @@ def test_log_to_file(mock_os_stat: MagicMock): mock_os_stat.return_value = [0o040000] + logger = Logger(error_counter=count) + with tempfile.TemporaryDirectory() as temp_dir: log_path = os.path.join(temp_dir, "poke") os.mkdir(log_path) - logger = Logger(error_counter=count, log_dir=log_path) + logger.set_log_dir(log_path) logger.info("Aaron Siemsen rocks") with open(os.path.join(log_path, "activity.log"), "r") as f: