update
This commit is contained in:
220
test/test_micropython_basic.py
Normal file
220
test/test_micropython_basic.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""
|
||||
Micropython NATS Bridge - Basic Test Examples
|
||||
|
||||
This module demonstrates basic usage of the NATSBridge for Micropython.
|
||||
"""
|
||||
|
||||
import sys
|
||||
sys.path.insert(0, "../src")
|
||||
|
||||
from nats_bridge import MessageEnvelope, MessagePayload, smartsend, smartreceive, log_trace
|
||||
import json
|
||||
|
||||
# ============================================= 100 ============================================== #
|
||||
|
||||
|
||||
def test_text_message():
|
||||
"""Test sending and receiving text messages."""
|
||||
print("\n=== Test 1: Text Message ===")
|
||||
|
||||
# Send text message
|
||||
data = [
|
||||
("message", "Hello World", "text"),
|
||||
("greeting", "Good morning!", "text")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/text",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
size_threshold=1000000
|
||||
)
|
||||
|
||||
print("Sent envelope:")
|
||||
print(" Subject: {}".format(env.send_to))
|
||||
print(" Correlation ID: {}".format(env.correlation_id))
|
||||
print(" Payloads: {}".format(len(env.payloads)))
|
||||
|
||||
# Expected output on receiver:
|
||||
# payloads = smartreceive(msg)
|
||||
# for dataname, data, type in payloads:
|
||||
# print("Received {}: {}".format(dataname, data))
|
||||
|
||||
|
||||
def test_dictionary_message():
|
||||
"""Test sending and receiving dictionary messages."""
|
||||
print("\n=== Test 2: Dictionary Message ===")
|
||||
|
||||
# Send dictionary message
|
||||
config = {
|
||||
"step_size": 0.01,
|
||||
"iterations": 1000,
|
||||
"threshold": 0.5
|
||||
}
|
||||
|
||||
data = [
|
||||
("config", config, "dictionary")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/dictionary",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
size_threshold=1000000
|
||||
)
|
||||
|
||||
print("Sent envelope:")
|
||||
print(" Subject: {}".format(env.send_to))
|
||||
print(" Payloads: {}".format(len(env.payloads)))
|
||||
|
||||
# Expected output on receiver:
|
||||
# payloads = smartreceive(msg)
|
||||
# for dataname, data, type in payloads:
|
||||
# if type == "dictionary":
|
||||
# print("Config: {}".format(data))
|
||||
|
||||
|
||||
def test_mixed_payloads():
|
||||
"""Test sending mixed payload types in a single message."""
|
||||
print("\n=== Test 3: Mixed Payloads ===")
|
||||
|
||||
# Mixed content: text, dictionary, and binary
|
||||
image_data = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR" # PNG header (example)
|
||||
|
||||
data = [
|
||||
("message_text", "Hello!", "text"),
|
||||
("user_config", {"theme": "dark", "volume": 80}, "dictionary"),
|
||||
("user_image", image_data, "binary")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/mixed",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
size_threshold=1000000
|
||||
)
|
||||
|
||||
print("Sent envelope:")
|
||||
print(" Subject: {}".format(env.send_to))
|
||||
print(" Payloads: {}".format(len(env.payloads)))
|
||||
|
||||
# Expected output on receiver:
|
||||
# payloads = smartreceive(msg)
|
||||
# for dataname, data, type in payloads:
|
||||
# print("Received {}: {} (type: {})".format(dataname, data if type != "binary" else len(data), type))
|
||||
|
||||
|
||||
def test_large_payload():
|
||||
"""Test sending large payloads that require fileserver upload."""
|
||||
print("\n=== Test 4: Large Payload (Link Transport) ===")
|
||||
|
||||
# Create large data (> 1MB would trigger link transport)
|
||||
# For testing, we'll use a smaller size but configure threshold lower
|
||||
large_data = b"A" * 100000 # 100KB
|
||||
|
||||
data = [
|
||||
("large_data", large_data, "binary")
|
||||
]
|
||||
|
||||
# Use a lower threshold for testing
|
||||
env = smartsend(
|
||||
"/test/large",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
fileserver_url="http://localhost:8080",
|
||||
size_threshold=50000 # 50KB threshold for testing
|
||||
)
|
||||
|
||||
print("Sent envelope:")
|
||||
print(" Subject: {}".format(env.send_to))
|
||||
print(" Payloads: {}".format(len(env.payloads)))
|
||||
for p in env.payloads:
|
||||
print(" - Transport: {}, Type: {}".format(p.transport, p.type))
|
||||
|
||||
|
||||
def test_reply_to():
|
||||
"""Test sending messages with reply-to functionality."""
|
||||
print("\n=== Test 5: Reply To ===")
|
||||
|
||||
data = [
|
||||
("command", {"action": "start"}, "dictionary")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/command",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
reply_to="/test/response",
|
||||
reply_to_msg_id="reply-123",
|
||||
msg_purpose="command"
|
||||
)
|
||||
|
||||
print("Sent envelope:")
|
||||
print(" Subject: {}".format(env.send_to))
|
||||
print(" Reply To: {}".format(env.reply_to))
|
||||
print(" Reply To Msg ID: {}".format(env.reply_to_msg_id))
|
||||
|
||||
|
||||
def test_correlation_id():
|
||||
"""Test using custom correlation IDs for tracing."""
|
||||
print("\n=== Test 6: Custom Correlation ID ===")
|
||||
|
||||
custom_cid = "trace-abc123"
|
||||
data = [
|
||||
("message", "Test with correlation ID", "text")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/correlation",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
correlation_id=custom_cid
|
||||
)
|
||||
|
||||
print("Sent envelope with correlation ID: {}".format(env.correlation_id))
|
||||
print("This ID can be used to trace the message flow.")
|
||||
|
||||
|
||||
def test_multiple_payloads():
|
||||
"""Test sending multiple payloads in one message."""
|
||||
print("\n=== Test 7: Multiple Payloads ===")
|
||||
|
||||
data = [
|
||||
("text_message", "Hello", "text"),
|
||||
("json_data", {"key": "value", "number": 42}, "dictionary"),
|
||||
("table_data", b"\x01\x02\x03\x04", "binary"),
|
||||
("audio_data", b"\x00\x01\x02\x03", "binary")
|
||||
]
|
||||
|
||||
env = smartsend(
|
||||
"/test/multiple",
|
||||
data,
|
||||
nats_url="nats://localhost:4222",
|
||||
size_threshold=1000000
|
||||
)
|
||||
|
||||
print("Sent {} payloads in one message".format(len(env.payloads)))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Micropython NATS Bridge Test Suite")
|
||||
print("==================================")
|
||||
print()
|
||||
|
||||
# Run tests
|
||||
test_text_message()
|
||||
test_dictionary_message()
|
||||
test_mixed_payloads()
|
||||
test_large_payload()
|
||||
test_reply_to()
|
||||
test_correlation_id()
|
||||
test_multiple_payloads()
|
||||
|
||||
print("\n=== All tests completed ===")
|
||||
print()
|
||||
print("Note: These tests require:")
|
||||
print(" 1. A running NATS server at nats://localhost:4222")
|
||||
print(" 2. An HTTP file server at http://localhost:8080 (for large payloads)")
|
||||
print()
|
||||
print("To run the tests:")
|
||||
print(" python test_micropython_basic.py")
|
||||
Reference in New Issue
Block a user