NATSBridge/test/test_micropython_basic.py

#!/usr/bin/env python3
"""
Basic functionality test for nats_bridge.py
Tests the core classes and functions without NATS connection
"""

import sys
import os

# Add src to path for import
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))

from nats_bridge import (
    MessagePayload,
    MessageEnvelope,
    smartsend,
    smartreceive,
    log_trace,
    generate_uuid,
    get_timestamp,
    _serialize_data,
    _deserialize_data
)
import json


def test_message_payload():
    """Test MessagePayload class"""
    print("\n=== Testing MessagePayload ===")
    
    # Test direct transport with text
    payload1 = MessagePayload(
        data="Hello World",
        msg_type="text",
        id="test-id-1",
        dataname="message",
        transport="direct",
        encoding="base64",
        size=11
    )
    
    assert payload1.id == "test-id-1"
    assert payload1.dataname == "message"
    assert payload1.type == "text"
    assert payload1.transport == "direct"
    assert payload1.encoding == "base64"
    assert payload1.size == 11
    print("  [PASS] MessagePayload with text data")
    
    # Test link transport with URL
    payload2 = MessagePayload(
        data="http://example.com/file.txt",
        msg_type="binary",
        id="test-id-2",
        dataname="file",
        transport="link",
        encoding="none",
        size=1000
    )
    
    assert payload2.transport == "link"
    assert payload2.data == "http://example.com/file.txt"
    print("  [PASS] MessagePayload with link transport")
    
    # Test to_dict method
    payload_dict = payload1.to_dict()
    assert "id" in payload_dict
    assert "dataname" in payload_dict
    assert "type" in payload_dict
    assert "transport" in payload_dict
    assert "data" in payload_dict
    print("  [PASS] MessagePayload.to_dict() method")


def test_message_envelope():
    """Test MessageEnvelope class"""
    print("\n=== Testing MessageEnvelope ===")
    
    # Create payloads
    payload1 = MessagePayload("Hello", "text", id="p1", dataname="msg1")
    payload2 = MessagePayload("http://example.com/file", "binary", id="p2", dataname="file", transport="link")
    
    # Create envelope
    env = MessageEnvelope(
        send_to="/test/subject",
        payloads=[payload1, payload2],
        correlation_id="test-correlation-id",
        msg_id="test-msg-id",
        msg_purpose="chat",
        sender_name="test_sender",
        receiver_name="test_receiver",
        reply_to="/test/reply"
    )
    
    assert env.send_to == "/test/subject"
    assert env.correlation_id == "test-correlation-id"
    assert env.msg_id == "test-msg-id"
    assert env.msg_purpose == "chat"
    assert len(env.payloads) == 2
    print("  [PASS] MessageEnvelope creation")
    
    # Test to_json method
    json_str = env.to_json()
    json_data = json.loads(json_str)
    assert json_data["sendTo"] == "/test/subject"
    assert json_data["correlationId"] == "test-correlation-id"
    assert json_data["msgPurpose"] == "chat"
    assert len(json_data["payloads"]) == 2
    print("  [PASS] MessageEnvelope.to_json() method")


def test_serialize_data():
    """Test _serialize_data function"""
    print("\n=== Testing _serialize_data ===")
    
    # Test text serialization
    text_bytes = _serialize_data("Hello", "text")
    assert isinstance(text_bytes, bytes)
    assert text_bytes == b"Hello"
    print("  [PASS] Text serialization")
    
    # Test dictionary serialization
    dict_data = {"key": "value", "number": 42}
    dict_bytes = _serialize_data(dict_data, "dictionary")
    assert isinstance(dict_bytes, bytes)
    parsed = json.loads(dict_bytes.decode('utf-8'))
    assert parsed["key"] == "value"
    print("  [PASS] Dictionary serialization")
    
    # Test binary serialization
    binary_data = b"\x00\x01\x02"
    binary_bytes = _serialize_data(binary_data, "binary")
    assert binary_bytes == b"\x00\x01\x02"
    print("  [PASS] Binary serialization")
    
    # Test image serialization
    image_data = bytes([1, 2, 3, 4, 5])
    image_bytes = _serialize_data(image_data, "image")
    assert image_bytes == image_data
    print("  [PASS] Image serialization")


def test_deserialize_data():
    """Test _deserialize_data function"""
    print("\n=== Testing _deserialize_data ===")
    
    # Test text deserialization
    text_bytes = b"Hello"
    text_data = _deserialize_data(text_bytes, "text", "test-correlation-id")
    assert text_data == "Hello"
    print("  [PASS] Text deserialization")
    
    # Test dictionary deserialization
    dict_bytes = b'{"key": "value"}'
    dict_data = _deserialize_data(dict_bytes, "dictionary", "test-correlation-id")
    assert dict_data == {"key": "value"}
    print("  [PASS] Dictionary deserialization")
    
    # Test binary deserialization
    binary_data = b"\x00\x01\x02"
    binary_result = _deserialize_data(binary_data, "binary", "test-correlation-id")
    assert binary_result == b"\x00\x01\x02"
    print("  [PASS] Binary deserialization")


def test_utilities():
    """Test utility functions"""
    print("\n=== Testing Utility Functions ===")
    
    # Test generate_uuid
    uuid1 = generate_uuid()
    uuid2 = generate_uuid()
    assert uuid1 != uuid2
    print(f"  [PASS] generate_uuid() - generated: {uuid1}")
    
    # Test get_timestamp
    timestamp = get_timestamp()
    assert "T" in timestamp
    print(f"  [PASS] get_timestamp() - generated: {timestamp}")


def main():
    """Run all tests"""
    print("=" * 60)
    print("NATSBridge Python/Micropython - Basic Functionality Tests")
    print("=" * 60)
    
    try:
        test_message_payload()
        test_message_envelope()
        test_serialize_data()
        test_deserialize_data()
        test_utilities()
        
        print("\n" + "=" * 60)
        print("ALL TESTS PASSED!")
        print("=" * 60)
        
    except Exception as e:
        print(f"\n[FAIL] Test failed with error: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()