63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import importlib.util
|
|
import json
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
MODULE_PATH = PROJECT_ROOT / "jwebot_edge_delivery.py"
|
|
SPEC = importlib.util.spec_from_file_location("jwebot_edge_delivery", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
sys.modules[SPEC.name] = MODULE
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class DeliveryHelperTests(unittest.TestCase):
|
|
def test_submit_gui_sender_uses_unix_socket_protocol(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
socket_path = Path(temporary) / "gui.sock"
|
|
received = {}
|
|
|
|
def server():
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener:
|
|
listener.bind(str(socket_path))
|
|
listener.listen(1)
|
|
connection, _ = listener.accept()
|
|
with connection:
|
|
raw = connection.recv(4096).split(b"\n", 1)[0]
|
|
received.update(json.loads(raw.decode("utf-8")))
|
|
connection.sendall(b'{"ok":true,"state":"sent"}\n')
|
|
|
|
thread = threading.Thread(target=server, daemon=True)
|
|
thread.start()
|
|
for _ in range(200):
|
|
if socket_path.exists():
|
|
break
|
|
MODULE.time.sleep(0.01)
|
|
|
|
with mock.patch.object(MODULE, "GUI_SENDER_SOCKET", socket_path):
|
|
response = MODULE.submit_gui_sender(
|
|
{
|
|
"op": "send",
|
|
"event_id": "event-1",
|
|
"group_fingerprint": "group-a",
|
|
"reply": "reply",
|
|
"source": "delivery-helper",
|
|
}
|
|
)
|
|
|
|
self.assertEqual(response["state"], "sent")
|
|
self.assertEqual(received["op"], "send")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|