128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
"""
|
|
Integration tests for ntfy topics.
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
class TestNtfyTopics:
|
|
def test_cc_cmd_topic_exists_and_writable(self):
|
|
"""Verify C&C CMD topic exists and accepts messages."""
|
|
from engine.config import NTFY_CC_CMD_TOPIC
|
|
|
|
topic_url = NTFY_CC_CMD_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 CMD topic: {e}") from e
|
|
|
|
def test_cc_resp_topic_exists_and_writable(self):
|
|
"""Verify C&C RESP topic exists and accepts messages."""
|
|
from engine.config import NTFY_CC_RESP_TOPIC
|
|
|
|
topic_url = NTFY_CC_RESP_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 RESP 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_cmd_topic_readable(self):
|
|
"""Verify we can read messages from C&C CMD topic."""
|
|
from engine.config import NTFY_CC_CMD_TOPIC
|
|
|
|
test_message = f"integration_test_{int(time.time())}"
|
|
topic_url = NTFY_CC_CMD_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 CMD topic: {e}") from e
|
|
|
|
time.sleep(1)
|
|
|
|
poll_url = f"{NTFY_CC_CMD_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 CMD topic: {e}") from e
|
|
|
|
def test_topics_are_different(self):
|
|
"""Verify C&C CMD/RESP and message topics are different."""
|
|
from engine.config import NTFY_CC_CMD_TOPIC, NTFY_CC_RESP_TOPIC, NTFY_TOPIC
|
|
|
|
assert NTFY_CC_CMD_TOPIC != NTFY_TOPIC
|
|
assert NTFY_CC_RESP_TOPIC != NTFY_TOPIC
|
|
assert NTFY_CC_CMD_TOPIC != NTFY_CC_RESP_TOPIC
|
|
assert "_cc_cmd" in NTFY_CC_CMD_TOPIC
|
|
assert "_cc_resp" in NTFY_CC_RESP_TOPIC
|