""" Python Dictionary Receiver Test Tests the smartreceive function with dictionary payloads """ import asyncio import sys import os import json # Add parent directory to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from natsbridge import smartreceive, DEFAULT_BROKER_URL, DEFAULT_FILESERVER_URL TEST_BROKER_URL = os.environ.get('NATS_URL', 'nats://localhost:4222') TEST_FILESERVER_URL = os.environ.get('FILESERVER_URL', 'http://localhost:8080') async def run_test(): print('=== Python Dictionary Receiver Test ===\n') # Create a mock NATS message with dictionary payloads import base64 simple_dict = {'key1': 'value1', 'key2': 'value2'} nested_dict = {'outer': {'inner': 'value', 'number': 42}} array_dict = {'items': [1, 2, 3, 'four', 'five']} mixed_dict = {'string': 'text', 'number': 123, 'boolean': True, 'null_val': None} test_data = { 'correlation_id': 'py-receiver-dict-' + str(asyncio.get_event_loop().time() * 1000000), 'msg_id': 'msg-' + str(asyncio.get_event_loop().time() * 1000000), 'timestamp': asyncio.get_event_loop().time().isoformat(), 'send_to': '/test/dictionary', 'msg_purpose': 'test', 'sender_name': 'py-dict-test', 'sender_id': 'sender-' + str(asyncio.get_event_loop().time() * 1000000), 'receiver_name': 'py-receiver', 'receiver_id': 'receiver-' + str(asyncio.get_event_loop().time() * 1000000), 'reply_to': '', 'reply_to_msg_id': '', 'broker_url': TEST_BROKER_URL, 'metadata': {}, 'payloads': [ { 'id': 'payload-1', 'dataname': 'simple_dict', 'payload_type': 'dictionary', 'transport': 'direct', 'encoding': 'base64', 'size': len(json.dumps(simple_dict).encode('utf8')), 'data': base64.b64encode(json.dumps(simple_dict).encode('utf8')).decode('ascii'), 'metadata': {'payload_bytes': len(json.dumps(simple_dict).encode('utf8'))} }, { 'id': 'payload-2', 'dataname': 'nested_dict', 'payload_type': 'dictionary', 'transport': 'direct', 'encoding': 'base64', 'size': len(json.dumps(nested_dict).encode('utf8')), 'data': base64.b64encode(json.dumps(nested_dict).encode('utf8')).decode('ascii'), 'metadata': {'payload_bytes': len(json.dumps(nested_dict).encode('utf8'))} }, { 'id': 'payload-3', 'dataname': 'array_dict', 'payload_type': 'dictionary', 'transport': 'direct', 'encoding': 'base64', 'size': len(json.dumps(array_dict).encode('utf8')), 'data': base64.b64encode(json.dumps(array_dict).encode('utf8')).decode('ascii'), 'metadata': {'payload_bytes': len(json.dumps(array_dict).encode('utf8'))} }, { 'id': 'payload-4', 'dataname': 'mixed_dict', 'payload_type': 'dictionary', 'transport': 'direct', 'encoding': 'base64', 'size': len(json.dumps(mixed_dict).encode('utf8')), 'data': base64.b64encode(json.dumps(mixed_dict).encode('utf8')).decode('ascii'), 'metadata': {'payload_bytes': len(json.dumps(mixed_dict).encode('utf8'))} } ] } mock_msg = { 'payload': json.dumps(test_data) } print('Mock Message Created:') print(f' Correlation ID: {test_data["correlation_id"]}') print(f' Payloads: {len(test_data["payloads"])}') print(f' Payload types: {", ".join(p["payload_type"] for p in test_data["payloads"])}\n') try: # Receive and process the message print('Receiving and processing message...') env = await smartreceive( mock_msg, max_retries=3, base_delay=100, max_delay=1000 ) print('\n=== Received Envelope ===') print(f'Correlation ID: {env["correlation_id"]}') print(f'Message ID: {env["msg_id"]}') print(f'Timestamp: {env["timestamp"]}') print(f'Subject: {env["send_to"]}') print(f'Payloads: {len(env["payloads"])}\n') # Validate received data print('=== Validation ===') passed = True if not env.get('correlation_id'): print('❌ correlation_id is missing') passed = False else: print('✅ correlation_id present') if len(env['payloads']) != 4: print(f'❌ Expected 4 payloads, got {len(env["payloads"])}') passed = False else: print('✅ Correct number of payloads') # Expected data expected_data = [ ('simple_dict', simple_dict, 'dictionary'), ('nested_dict', nested_dict, 'dictionary'), ('array_dict', array_dict, 'dictionary'), ('mixed_dict', mixed_dict, 'dictionary') ] for i in range(len(env['payloads'])): payload = env['payloads'][i] expected = expected_data[i] if payload[0] != expected[0]: print(f"❌ Payload {i + 1}: Expected dataname '{expected[0]}', got '{payload[0]}'") passed = False else: print(f'✅ Payload {i + 1}: Correct dataname') if payload[2] != expected[2]: print(f"❌ Payload {i + 1}: Expected type '{expected[2]}', got '{payload[2]}'") passed = False else: print(f'✅ Payload {i + 1}: Correct type') data_match = json.dumps(payload[1], sort_keys=True) == json.dumps(expected[1], sort_keys=True) if not data_match: print(f'❌ Payload {i + 1}: Data mismatch') print(f' Expected: {json.dumps(expected[1], sort_keys=True)}') print(f' Got: {json.dumps(payload[1], sort_keys=True)}') passed = False else: print(f'✅ Payload {i + 1}: Data correctly deserialized') # Test round-trip with receive print('\n=== Round-trip Test ===') round_trip_data = { 'correlation_id': 'roundtrip-' + str(asyncio.get_event_loop().time() * 1000000), 'msg_id': 'msg-' + str(asyncio.get_event_loop().time() * 1000000), 'timestamp': asyncio.get_event_loop().time().isoformat(), 'send_to': '/test/dictionary', 'msg_purpose': 'test', 'sender_name': 'py-test', 'sender_id': 'sender-' + str(asyncio.get_event_loop().time() * 1000000), 'receiver_name': 'py-receiver', 'receiver_id': 'receiver-' + str(asyncio.get_event_loop().time() * 1000000), 'reply_to': '', 'reply_to_msg_id': '', 'broker_url': TEST_BROKER_URL, 'metadata': {}, 'payloads': [ { 'id': 'payload-rt', 'dataname': 'roundtrip', 'payload_type': 'dictionary', 'transport': 'direct', 'encoding': 'base64', 'size': len(json.dumps({'test': 'data', 'nested': {'a': 1, 'b': 2}}).encode('utf8')), 'data': base64.b64encode(json.dumps({'test': 'data', 'nested': {'a': 1, 'b': 2}}).encode('utf8')).decode('ascii'), 'metadata': {'payload_bytes': len(json.dumps({'test': 'data', 'nested': {'a': 1, 'b': 2}}).encode('utf8'))} } ] } mock_rt_msg = {'payload': json.dumps(round_trip_data)} rt_env = await smartreceive(mock_rt_msg) if rt_env['payloads'][0][0] == 'roundtrip' and rt_env['payloads'][0][2] == 'dictionary': print('✅ Round-trip test successful') else: print('❌ Round-trip test failed') passed = False # Final result print('\n=== Test Result ===') if passed: print('✅ ALL TESTS PASSED') sys.exit(0) else: print('❌ SOME TESTS FAILED') sys.exit(1) except Exception as e: print(f'❌ Test failed with error: {e}') import traceback traceback.print_exc() sys.exit(1) if __name__ == '__main__': asyncio.run(run_test())