From e21e49885e30d5a2728fc3a697d13c3bdf94dad9 Mon Sep 17 00:00:00 2001
From: Dominic Amato <dom@hologram.io>
Date: Fri, 3 Jan 2020 12:05:13 -0600
Subject: [PATCH] Add chunking for messeages over 512 bytes (#15)
---
Hologram/Network/Modem/Modem.py | 25 ++++++++++++++++---------
tests/Modem/test_Modem.py | 29 +++++++++++++++++++++--------
2 files changed, 37 insertions(+), 17 deletions(-)
diff --git a/Hologram/Network/Modem/Modem.py b/Hologram/Network/Modem/Modem.py
index 5b19d56..3f5d710 100644
--- a/Hologram/Network/Modem/Modem.py
+++ b/Hologram/Network/Modem/Modem.py
@@ -289,7 +289,6 @@ def connect_socket(self, host, port):
self.logger.info('Connect socket is successful')
def listen_socket(self, port):
-
at_command_val = "%d,%s" % (self.socket_identifier, port)
self.listen_socket_identifier = self.socket_identifier
ok, _ = self.set('+USOLI', at_command_val, timeout=5)
@@ -298,17 +297,25 @@ def listen_socket(self, port):
raise NetworkError('Failed to listen socket')
def write_socket(self, data):
-
self.enable_hex_mode()
- value = b'%d,%d,\"%s\"' % (self.socket_identifier,
- len(data),
- binascii.hexlify(data))
- ok, _ = self.set('+USOWR', value, timeout=10)
- if ok != ModemResult.OK:
- self.logger.error('Failed to write to socket')
- raise NetworkError('Failed to write socket')
+ hexdata = binascii.hexlify(data)
+ # We have to do it in chunks of 510 since 512 is actually too long (CMEE error)
+ # and we need 2n chars for hexified data
+ for chunk in self._chunks(hexdata, 510):
+ value = b'%d,%d,\"%s\"' % (self.socket_identifier,
+ len(binascii.unhexlify(chunk)),
+ chunk)
+ ok, _ = self.set('+USOWR', value, timeout=10)
+ if ok != ModemResult.OK:
+ self.logger.error('Failed to write to socket')
+ raise NetworkError('Failed to write socket')
self.disable_hex_mode()
+ def _chunks(self, data, n):
+ """Yield successive n-sized chunks from lst."""
+ for i in range(0, len(data), n):
+ yield data[i:i + n]
+
def read_socket(self, socket_identifier=None, payload_length=None):
if socket_identifier is None:
diff --git a/tests/Modem/test_Modem.py b/tests/Modem/test_Modem.py
index dc6cf2d..9cd540a 100644
--- a/tests/Modem/test_Modem.py
+++ b/tests/Modem/test_Modem.py
@@ -32,6 +32,9 @@ def mock_open_serial_port(modem, device_name=None):
def mock_close_serial_port(modem):
return True
+def mock_result(modem):
+ return (ModemResult.OK, None)
+
def mock_detect_usable_serial_port(modem, stop_on_first=True):
return '/dev/ttyUSB0'
@@ -54,9 +57,14 @@ def no_serial_port(monkeypatch):
def get_sms(monkeypatch):
monkeypatch.setattr(Modem, 'command', mock_command_sms)
monkeypatch.setattr(Modem, 'set', mock_set_sms)
+def override_command_result(monkeypatch):
+ monkeypatch.setattr(Modem, '_command_result', mock_result)
-# CONSTRUCTOR
+@pytest.fixture
+def override_command_result(monkeypatch):
+ monkeypatch.setattr(Modem, '_command_result', mock_result)
+# CONSTRUCTOR
def test_init_modem_no_args(no_serial_port):
modem = Modem()
@@ -80,10 +88,8 @@ def test_get_result_string(no_serial_port):
assert(modem.getResultString(-3) == 'Modem response doesn\'t match expected return value')
assert(modem.getResultString(-99) == 'Unknown response code')
-
# PROPERTIES
-
def test_get_location(no_serial_port):
modem = Modem()
with pytest.raises(NotImplementedError) as e:
@@ -99,8 +105,19 @@ def test_get_sms(no_serial_port, get_sms):
assert(res.timestamp == datetime.utcfromtimestamp(1498264009))
assert(res.message == 'Test 123')
-# DEBUGWRITE
+# WRITE SOCKET
+
+def test_socket_write_under_512(no_serial_port, override_command_result):
+ modem = Modem()
+ data = '{message:{fill}{align}{width}}'.format(message='Test-', fill='@', align='<', width=64)
+ modem.write_socket(data.encode())
+
+def test_socket_write_over_512(no_serial_port, override_command_result):
+ modem = Modem()
+ data = '{message:{fill}{align}{width}}'.format(message='Test-', fill='@', align='<', width=600)
+ modem.write_socket(data.encode())
+# DEBUGWRITE
def test_debugwrite(no_serial_port):
modem = Modem()
@@ -111,10 +128,8 @@ def test_debugwrite(no_serial_port):
modem.debugwrite('test222', hide=True)
assert(modem.debug_out == 'test') # debug_out shouldn't change since hide is enabled.
-
# MODEMWRITE
-
def test_modemwrite(no_serial_port):
modem = Modem()
assert(modem.debug_out == '')
@@ -136,7 +151,6 @@ def test_modemwrite(no_serial_port):
modem.modemwrite('test5', start=True, at=True, seteq=True, read=True, end=True)
assert(modem.debug_out == '[ATtest5=?]')
-
# COMMAND_RESULT
def test_command_result(no_serial_port):
@@ -193,7 +207,6 @@ def test_command_result(no_serial_port):
# HANDLEURC
-
# These are static methods that can be tested independently.
# We decided to wrap it all here under this test object
class TestModemProtectedStaticMethods():