FreeDATA/tests/test_message_p2p.py
2024-01-27 12:07:07 +01:00

69 lines
2.7 KiB
Python
Executable file

import sys
sys.path.append('modem')
import numpy as np
import unittest
from config import CONFIG
from message_p2p import MessageP2P
from message_system_db_manager import DatabaseManager
from event_manager import EventManager
import queue
class TestDataFrameFactory(unittest.TestCase):
@classmethod
def setUpClass(cls):
config_manager = CONFIG('modem/config.ini.example')
cls.config = config_manager.read()
cls.event_queue = queue.Queue()
cls.event_manager = EventManager([cls.event_queue])
cls.mycall = f"{cls.config['STATION']['mycall']}-{cls.config['STATION']['myssid']}"
cls.database_manager = DatabaseManager(cls.event_manager, uri='sqlite:///:memory:')
def testFromApiParams(self):
api_params = {
'dxcall': 'DJ2LS-3',
'body': 'Hello World!',
}
message = MessageP2P.from_api_params(self.mycall, api_params)
self.assertEqual(message.destination, api_params['dxcall'])
self.assertEqual(message.body, api_params['body'])
def testToPayloadWithAttachment(self):
attachment = {
'name': 'test.gif',
'type': 'image/gif',
'data': np.random.bytes(1024)
}
message = MessageP2P(self.mycall, 'DJ2LS-3', 'Hello World!', [attachment])
payload = message.to_payload()
received_message = MessageP2P.from_payload(payload)
self.assertEqual(message.origin, received_message.origin)
self.assertEqual(message.destination, received_message.destination)
self.assertCountEqual(message.attachments, received_message.attachments)
self.assertEqual(attachment['data'], received_message.attachments[0]['data'])
def testToPayloadWithAttachmentAndDatabase(self):
attachment = {
'name': 'test.gif',
'type': 'image/gif',
'data': np.random.bytes(1024)
}
message = MessageP2P(self.mycall, 'DJ2LS-3', 'Hello World!', [attachment])
payload = message.to_payload()
received_message = MessageP2P.from_payload(payload)
received_message_dict = MessageP2P.to_dict(received_message, received=True)
self.database_manager.add_message(received_message_dict)
self.assertEqual(message.origin, received_message.origin)
self.assertEqual(message.destination, received_message.destination)
self.assertCountEqual(message.attachments, received_message.attachments)
self.assertEqual(attachment['data'], received_message.attachments[0]['data'])
result = self.database_manager.get_all_messages()
self.assertEqual(result[0]["destination"], message.destination)
if __name__ == '__main__':
unittest.main()