Mathieu MoranandCopilot
3a6f12ba9d
Add cellular-first ntfy delivery with WiFi fallback and safe SIM868 HTTP header framing. Co-authored-by: Copilot <[email protected]>
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
import unittest
|
|
from contextlib import nullcontext
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from modules import app_cellular
|
|
|
|
|
|
class CellularHttpTest(unittest.TestCase):
|
|
def test_rejects_at_command_injection_before_modem_access(self):
|
|
with (
|
|
patch(
|
|
"modules.app_cellular.is_configured",
|
|
return_value=True,
|
|
),
|
|
patch("modules.app_cellular._ensure_gprs_context") as ensure,
|
|
self.assertRaises(app_cellular.CellularError),
|
|
):
|
|
app_cellular.http_post(
|
|
'https://example.test/"\r\nAT+RESET',
|
|
b"body",
|
|
{},
|
|
)
|
|
|
|
ensure.assert_not_called()
|
|
|
|
@patch("modules.app_cellular.http_post")
|
|
def test_json_wrapper_uses_json_content_type(self, post):
|
|
post.return_value = (200, b"ok")
|
|
|
|
result = app_cellular.http_post_json(
|
|
"https://example.test/upload",
|
|
b"{}",
|
|
{"X-Api-Key": "key"},
|
|
timeout=30,
|
|
)
|
|
|
|
self.assertEqual(result, (200, b"ok"))
|
|
post.assert_called_once_with(
|
|
"https://example.test/upload",
|
|
b"{}",
|
|
{"X-Api-Key": "key"},
|
|
content_type="application/json",
|
|
timeout=30,
|
|
)
|
|
|
|
@patch("modules.app_cellular._read_until", return_value="OK")
|
|
@patch("modules.app_cellular._ensure_gprs_context")
|
|
@patch("modules.app_cellular.is_configured", return_value=True)
|
|
@patch("modules.app_cellular._at")
|
|
def test_multiple_headers_use_literal_crlf_escape_sequence(
|
|
self,
|
|
at,
|
|
_is_configured,
|
|
_ensure_context,
|
|
_read_until,
|
|
):
|
|
serial = MagicMock()
|
|
at.side_effect = lambda _serial, command, **_kwargs: (
|
|
"+HTTPACTION: 1,200,0"
|
|
if command == "AT+HTTPACTION=1"
|
|
else "OK"
|
|
)
|
|
with (
|
|
patch(
|
|
"modules.app_cellular.app_gps._serial_lock",
|
|
nullcontext(),
|
|
),
|
|
patch(
|
|
"modules.app_cellular.app_gps._get_serial",
|
|
return_value=serial,
|
|
),
|
|
):
|
|
status, _body = app_cellular.http_post(
|
|
"https://example.test/topic",
|
|
b"message",
|
|
{"Title": "Alert", "Priority": "high"},
|
|
content_type="text/plain",
|
|
)
|
|
|
|
self.assertEqual(status, 200)
|
|
user_data_commands = [
|
|
call.args[1]
|
|
for call in at.call_args_list
|
|
if 'AT+HTTPPARA="USERDATA"' in call.args[1]
|
|
]
|
|
self.assertEqual(
|
|
user_data_commands,
|
|
[
|
|
'AT+HTTPPARA="USERDATA","'
|
|
r'Title: Alert\r\nPriority: high\r\n"'
|
|
],
|
|
)
|
|
self.assertNotIn("\r", user_data_commands[0])
|
|
self.assertNotIn("\n", user_data_commands[0])
|
|
|
|
@patch("modules.app_cellular._ensure_gprs_context")
|
|
@patch("modules.app_cellular.is_configured", return_value=True)
|
|
def test_rejects_headers_over_modem_limit(
|
|
self,
|
|
_is_configured,
|
|
ensure_context,
|
|
):
|
|
with self.assertRaises(app_cellular.CellularError):
|
|
app_cellular.http_post(
|
|
"https://example.test/topic",
|
|
b"message",
|
|
{"X-Large": "x" * 1024},
|
|
)
|
|
|
|
ensure_context.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|