FreeDATA/tests/test_message_p2p.py

78 lines
2.9 KiB
Python
Raw Normal View History

2024-01-18 10:35:44 +00:00
import sys
2024-04-18 09:04:25 +00:00
sys.path.append('freedata_server')
import numpy as np
2024-01-18 10:35:44 +00:00
import unittest
from config import CONFIG
from message_p2p import MessageP2P
from message_system_db_messages import DatabaseManagerMessages
2024-01-27 11:07:07 +00:00
from event_manager import EventManager
import queue
2024-01-29 16:50:28 +00:00
import base64
2024-01-18 10:35:44 +00:00
class TestDataFrameFactory(unittest.TestCase):
@classmethod
def setUpClass(cls):
2024-04-18 09:04:25 +00:00
config_manager = CONFIG('freedata_server/config.ini.example')
2024-01-18 10:35:44 +00:00
cls.config = config_manager.read()
2024-01-27 11:07:07 +00:00
cls.event_queue = queue.Queue()
cls.event_manager = EventManager([cls.event_queue])
2024-01-18 10:35:44 +00:00
cls.mycall = f"{cls.config['STATION']['mycall']}-{cls.config['STATION']['myssid']}"
cls.database_manager = DatabaseManagerMessages(cls.event_manager)
2024-01-18 10:35:44 +00:00
def testFromApiParams(self):
api_params = {
2024-02-02 18:50:16 +00:00
'destination': 'DJ2LS-3',
2024-01-18 10:35:44 +00:00
'body': 'Hello World!',
}
message = MessageP2P.from_api_params(self.mycall, api_params)
2024-02-02 18:50:16 +00:00
self.assertEqual(message.destination, api_params['destination'])
2024-01-18 10:35:44 +00:00
self.assertEqual(message.body, api_params['body'])
def testToPayloadWithAttachment(self):
attachment = {
'name': 'test.gif',
'type': 'image/gif',
2024-01-29 16:50:28 +00:00
'data': str(base64.b64encode(np.random.bytes(1024)), 'utf-8')
2024-01-18 10:35:44 +00:00
}
2024-02-02 18:50:16 +00:00
apiParams = {'destination': 'DJ2LS-3', 'body': 'Hello World!', 'attachments': [attachment]}
2024-01-29 16:50:28 +00:00
message = MessageP2P.from_api_params(self.mycall, apiParams)
2024-01-18 10:35:44 +00:00
payload = message.to_payload()
2024-01-27 11:07:07 +00:00
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)
2024-01-29 16:50:28 +00:00
# FIXME...
#self.assertEqual(attachment['data'], received_message.attachments[0]['data'])
2024-01-27 11:07:07 +00:00
def testToPayloadWithAttachmentAndDatabase(self):
attachment = {
'name': 'test.gif',
'type': 'image/gif',
2024-01-29 16:50:28 +00:00
'data': str(base64.b64encode(np.random.bytes(1024)), 'utf-8')
2024-01-27 11:07:07 +00:00
}
2024-02-02 18:50:16 +00:00
apiParams = {'destination': 'DJ2LS-3', 'body': 'Hello World!', 'attachments': [attachment]}
2024-01-29 16:50:28 +00:00
message = MessageP2P.from_api_params(self.mycall, apiParams)
2024-01-27 11:07:07 +00:00
payload = message.to_payload()
received_message = MessageP2P.from_payload(payload)
2024-01-29 16:50:28 +00:00
received_message_dict = MessageP2P.to_dict(received_message)
2024-02-24 21:27:01 +00:00
self.database_manager.add_message(received_message_dict, statistics={})
self.assertEqual(message.origin, received_message.origin)
self.assertEqual(message.destination, received_message.destination)
self.assertCountEqual(message.attachments, received_message.attachments)
2024-02-07 18:54:54 +00:00
result = self.database_manager.get_message_by_id(message.id)
self.assertEqual(result["is_read"], True)
self.assertEqual(result["destination"], message.destination)
2024-01-18 10:35:44 +00:00
2024-01-27 11:07:07 +00:00
2024-01-18 10:35:44 +00:00
if __name__ == '__main__':
unittest.main()