diff --git a/notecard/card.py b/notecard/card.py index 4edf1af..c3e3306 100644 --- a/notecard/card.py +++ b/notecard/card.py @@ -9,7 +9,6 @@ # This module contains helper methods for calling card.* Notecard API commands. # This module is optional and not required for use with the Notecard. -import notecard from notecard.validators import validate_card_object @@ -137,16 +136,53 @@ def wireless(card, mode=None, apn=None): Args: card (Notecard): The current Notecard object. - mode (string): The wireless module mode to set. + mode (string): The wireless module mode to set. Must be one of: + "-" to reset to the default mode + "auto" to perform automatic band scan mode (default) + "m" to restrict the modem to Cat-M1 + "nb" to restrict the modem to Cat-NB1 + "gprs" to restrict the modem to EGPRS apn (string): Access Point Name (APN) when using an external SIM. + Use "-" to reset to the Notecard default APN. Returns: - string: The result of the Notecard request. + dict: The result of the Notecard request containing network status and + signal information. """ req = {"req": "card.wireless"} if mode: req["mode"] = mode if apn: req["apn"] = apn + return card.Transaction(req) + + +@validate_card_object +def transport(card, method=None, allow=None): + """Configure the Notecard's connectivity method. + Args: + card (Notecard): The current Notecard object. + method (string): The connectivity method to enable. Must be one of: + "-" to reset to device default + "wifi-cell" to prioritize WiFi with cellular fallback + "wifi" to enable WiFi only + "cell" to enable cellular only + "ntn" to enable Non-Terrestrial Network mode + "wifi-ntn" to prioritize WiFi with NTN fallback + "cell-ntn" to prioritize cellular with NTN fallback + "wifi-cell-ntn" to prioritize WiFi, then cellular, then NTN + allow (bool): When True, allows adding Notes to non-compact Notefiles + while connected over a non-terrestrial network. + + Returns: + dict: The result of the Notecard request. + """ + req = {"req": "card.transport"} + if method: + req["method"] = method + if allow is not None: + if not isinstance(allow, bool): + return {"err": "allow parameter must be a boolean"} + req["allow"] = "true" if allow else "false" return card.Transaction(req) diff --git a/notecard/notecard.py b/notecard/notecard.py index 3ac22ac..e2f68a2 100644 --- a/notecard/notecard.py +++ b/notecard/notecard.py @@ -34,6 +34,7 @@ import os import json import time +from abc import ABC, abstractmethod from notecard.timeout import start_timeout, has_timed_out from notecard.transaction_manager import TransactionManager, NoOpTransactionManager from notecard.crc32 import crc32 @@ -106,7 +107,7 @@ def release(*args, **kwargs): pass -class Notecard: +class Notecard(ABC): """Base Notecard class.""" def __init__(self, debug=False): @@ -333,6 +334,11 @@ def Transaction(self, req, lock=True): continue try: + if rsp_bytes is None: + error = True + retries_left -= 1 + time.sleep(0.5) + continue rsp_json = json.loads(rsp_bytes) except Exception as e: if self._debug: @@ -419,6 +425,26 @@ def SetTransactionPins(self, rtx_pin, ctx_pin): """Set the pins used for RTX and CTX.""" self._transaction_manager = TransactionManager(rtx_pin, ctx_pin) + @abstractmethod + def Reset(self): + """Reset the Notecard. Must be implemented by subclasses.""" + pass + + @abstractmethod + def lock(self): + """Lock access to the Notecard. Must be implemented by subclasses.""" + pass + + @abstractmethod + def unlock(self): + """Unlock access to the Notecard. Must be implemented by subclasses.""" + pass + + @abstractmethod + def _transact(self, req_bytes, rsp_expected, timeout_secs): + """Perform a transaction with the Notecard. Must be implemented by subclasses.""" + pass + class OpenSerial(Notecard): """Notecard class for Serial communication.""" diff --git a/test/fluent_api/conftest.py b/test/fluent_api/conftest.py index ced1a94..d66fbce 100644 --- a/test/fluent_api/conftest.py +++ b/test/fluent_api/conftest.py @@ -9,27 +9,41 @@ import notecard # noqa: E402 +class MockNotecard(notecard.Notecard): + def Reset(self): + pass + + def lock(self): + pass + + def unlock(self): + pass + + def _transact(self, req_bytes, rsp_expected, timeout_secs): + pass + + +@pytest.fixture +def card(): + """Create a mock Notecard instance for testing.""" + card = MockNotecard() + card.Transaction = MagicMock() + return card + + @pytest.fixture def run_fluent_api_notecard_api_mapping_test(): def _run_test(fluent_api, notecard_api_name, req_params, rename_map=None): - card = notecard.Notecard() + card = MockNotecard() card.Transaction = MagicMock() - - fluent_api(card, **req_params) - expected_notecard_api_req = {'req': notecard_api_name, **req_params} - - # There are certain fluent APIs that have keyword arguments that don't - # map exactly onto the Notecard API. For example, note.changes takes a - # 'maximum' parameter, but in the JSON request that gets sent to the - # Notecard, it's sent as 'max'. The rename_map allows a test to specify - # how a fluent API's keyword args map to Notecard API args, in cases - # where they differ. - if rename_map is not None: + fluent_api(card) + expected_req = {"req": notecard_api_name} + expected_req.update(req_params) + if rename_map: for old_key, new_key in rename_map.items(): - expected_notecard_api_req[new_key] = expected_notecard_api_req.pop(old_key) - - card.Transaction.assert_called_once_with(expected_notecard_api_req) - + if old_key in expected_req: + expected_req[new_key] = expected_req.pop(old_key) + card.Transaction.assert_called_once_with(expected_req) return _run_test diff --git a/test/fluent_api/test_card.py b/test/fluent_api/test_card.py index beb246d..f6445c8 100644 --- a/test/fluent_api/test_card.py +++ b/test/fluent_api/test_card.py @@ -3,67 +3,42 @@ @pytest.mark.parametrize( - 'fluent_api,notecard_api,req_params', + "fluent_api,notecard_api,req_params", [ ( card.attn, - 'card.attn', + "card.attn", { - 'mode': 'arm', - 'files': ['data.qi', 'my-settings.db'], - 'seconds': 60, - 'payload': 'ewogICJpbnRlcnZhbHMiOiI2MCwxMiwxNCIKfQ==', - 'start': True - } - ), - ( - card.status, - 'card.status', - {} - ), - ( - card.time, - 'card.time', - {} - ), - ( - card.temp, - 'card.temp', - {'minutes': 5} - ), - ( - card.version, - 'card.version', - {} + "mode": "arm", + "files": ["data.qi", "my-settings.db"], + "seconds": 60, + "payload": "ewogICJpbnRlcnZhbHMiOiI2MCwxMiwxNCIKfQ==", + "start": True, + }, ), + (card.status, "card.status", {}), + (card.time, "card.time", {}), + (card.temp, "card.temp", {"minutes": 5}), + (card.version, "card.version", {}), ( card.voltage, - 'card.voltage', - { - 'hours': 1, - 'offset': 2, - 'vmax': 1.1, - 'vmin': 1.2 - } + "card.voltage", + {"hours": 1, "offset": 2, "vmax": 1.1, "vmin": 1.2}, ), - ( - card.wireless, - 'card.wireless', - { - 'mode': 'auto', - 'apn': 'myapn.nb' - } - ) - ] + (card.wireless, "card.wireless", {"mode": "auto", "apn": "myapn.nb"}), + ], ) class TestCard: def test_fluent_api_maps_notecard_api_correctly( - self, fluent_api, notecard_api, req_params, - run_fluent_api_notecard_api_mapping_test): - run_fluent_api_notecard_api_mapping_test(fluent_api, notecard_api, - req_params) + self, + fluent_api, + notecard_api, + req_params, + run_fluent_api_notecard_api_mapping_test, + ): + run_fluent_api_notecard_api_mapping_test(fluent_api, notecard_api, req_params) def test_fluent_api_fails_with_invalid_notecard( - self, fluent_api, notecard_api, req_params, - run_fluent_api_invalid_notecard_test): + self, fluent_api, notecard_api, req_params, run_fluent_api_invalid_notecard_test + ): run_fluent_api_invalid_notecard_test(fluent_api, req_params) diff --git a/test/fluent_api/test_card_transport.py b/test/fluent_api/test_card_transport.py new file mode 100644 index 0000000..3d96a2f --- /dev/null +++ b/test/fluent_api/test_card_transport.py @@ -0,0 +1,90 @@ +"""Tests for card.transport functionality.""" +from notecard import card as card_module + + +def test_transport_basic(run_fluent_api_notecard_api_mapping_test): + """Test card.transport with no parameters.""" + run_fluent_api_notecard_api_mapping_test( + lambda nc: card_module.transport(nc), + 'card.transport', + {}) + + +def test_transport_method(run_fluent_api_notecard_api_mapping_test): + """Test card.transport with method parameter.""" + run_fluent_api_notecard_api_mapping_test( + lambda nc: card_module.transport(nc, method='ntn'), + 'card.transport', + {'method': 'ntn'}) + + +def test_transport_allow(run_fluent_api_notecard_api_mapping_test): + """Test card.transport with allow parameter.""" + run_fluent_api_notecard_api_mapping_test( + lambda nc: card_module.transport(nc, allow=True), + 'card.transport', + {'allow': 'true'}) + + +def test_transport_all_params(run_fluent_api_notecard_api_mapping_test): + """Test card.transport with all parameters.""" + run_fluent_api_notecard_api_mapping_test( + lambda nc: card_module.transport( + nc, method='wifi-cell-ntn', allow=True), + 'card.transport', + {'method': 'wifi-cell-ntn', 'allow': 'true'}) + + +def test_transport_invalid_allow(card): + """Test card.transport with invalid allow parameter.""" + result = card_module.transport(card, allow="not-a-boolean") + assert "err" in result + assert "allow parameter must be a boolean" in result["err"] + + +def test_transport_ntn_method(card): + """Test card.transport with NTN method.""" + card.Transaction.return_value = {"connected": True} + result = card_module.transport(card, method="ntn") + assert card.Transaction.called + assert card.Transaction.call_args[0][0] == { + "req": "card.transport", + "method": "ntn" + } + assert result == {"connected": True} + + +def test_transport_wifi_ntn_method(card): + """Test card.transport with WiFi-NTN method.""" + card.Transaction.return_value = {"connected": True} + result = card_module.transport(card, method="wifi-ntn") + assert card.Transaction.called + assert card.Transaction.call_args[0][0] == { + "req": "card.transport", + "method": "wifi-ntn" + } + assert result == {"connected": True} + + +def test_transport_cell_ntn_method(card): + """Test card.transport with Cell-NTN method.""" + card.Transaction.return_value = {"connected": True} + result = card_module.transport(card, method="cell-ntn") + assert card.Transaction.called + assert card.Transaction.call_args[0][0] == { + "req": "card.transport", + "method": "cell-ntn" + } + assert result == {"connected": True} + + +def test_transport_wifi_cell_ntn_method(card): + """Test card.transport with WiFi-Cell-NTN method.""" + card.Transaction.return_value = {"connected": True} + result = card_module.transport(card, method="wifi-cell-ntn") + assert card.Transaction.called + assert card.Transaction.call_args[0][0] == { + "req": "card.transport", + "method": "wifi-cell-ntn" + } + assert result == {"connected": True}