David Gwilliam
648326cd37
- Rewrite cmdline to send commands via ntfy and wait for response - Add NtfyResponsePoller class for serial-port-like interface - Add integration tests for ntfy topics (test read/write) - Add NTFY_CC_TOPIC export to config
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""
|
|
Integration tests for ntfy topics.
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
class TestNtfyTopics:
|
|
def test_cc_topic_exists_and_writable(self):
|
|
"""Verify C&C topic exists and accepts messages."""
|
|
from engine.config import NTFY_CC_TOPIC
|
|
|
|
topic_url = NTFY_CC_TOPIC.replace("/json", "")
|
|
test_message = f"test_{int(time.time())}"
|
|
|
|
req = urllib.request.Request(
|
|
topic_url,
|
|
data=test_message.encode("utf-8"),
|
|
headers={
|
|
"User-Agent": "mainline-test/0.1",
|
|
"Content-Type": "text/plain",
|
|
},
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
assert resp.status == 200
|
|
except Exception as e:
|
|
raise AssertionError(f"Failed to write to C&C topic: {e}") from e
|
|
|
|
def test_message_topic_exists_and_writable(self):
|
|
"""Verify message topic exists and accepts messages."""
|
|
from engine.config import NTFY_TOPIC
|
|
|
|
topic_url = NTFY_TOPIC.replace("/json", "")
|
|
test_message = f"test_{int(time.time())}"
|
|
|
|
req = urllib.request.Request(
|
|
topic_url,
|
|
data=test_message.encode("utf-8"),
|
|
headers={
|
|
"User-Agent": "mainline-test/0.1",
|
|
"Content-Type": "text/plain",
|
|
},
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
assert resp.status == 200
|
|
except Exception as e:
|
|
raise AssertionError(f"Failed to write to message topic: {e}") from e
|
|
|
|
def test_cc_topic_readable(self):
|
|
"""Verify we can read messages from C&C topic."""
|
|
from engine.config import NTFY_CC_TOPIC
|
|
|
|
test_message = f"integration_test_{int(time.time())}"
|
|
topic_url = NTFY_CC_TOPIC.replace("/json", "")
|
|
|
|
req = urllib.request.Request(
|
|
topic_url,
|
|
data=test_message.encode("utf-8"),
|
|
headers={
|
|
"User-Agent": "mainline-test/0.1",
|
|
"Content-Type": "text/plain",
|
|
},
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
urllib.request.urlopen(req, timeout=10)
|
|
except Exception as e:
|
|
raise AssertionError(f"Failed to write to C&C topic: {e}") from e
|
|
|
|
time.sleep(1)
|
|
|
|
poll_url = f"{NTFY_CC_TOPIC}?poll=1&limit=1"
|
|
req = urllib.request.Request(
|
|
poll_url,
|
|
headers={"User-Agent": "mainline-test/0.1"},
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
if body.strip():
|
|
data = json.loads(body.split("\n")[0])
|
|
assert isinstance(data, dict)
|
|
except Exception as e:
|
|
raise AssertionError(f"Failed to read from C&C topic: {e}") from e
|
|
|
|
def test_topics_are_different(self):
|
|
"""Verify C&C and message topics are different."""
|
|
from engine.config import NTFY_CC_TOPIC, NTFY_TOPIC
|
|
|
|
assert NTFY_CC_TOPIC != NTFY_TOPIC
|
|
assert "_cc" in NTFY_CC_TOPIC
|