151 lines
6.6 KiB
Python
151 lines
6.6 KiB
Python
import unittest
|
|
from unittest.mock import patch, Mock
|
|
import requests
|
|
import json
|
|
import os
|
|
|
|
from components import ArchiveSynchronizer
|
|
|
|
|
|
class TestArchiveSynchronizer(unittest.TestCase):
|
|
def setUp(self):
|
|
self.synchronizer = ArchiveSynchronizer(config=Mock(), name=None)
|
|
self.synchronizer.archive_endpoint = "https://dev.r5portal.it/api/st-ten-save/"
|
|
self.synchronizer.status_endpoint = "https://dev.r5portal.it/api/device-info-update/"
|
|
self.synchronizer.download_endpoint = "https://dev.r5portal.it/media/uploads/"
|
|
self.synchronizer.machine_id = "st-ten-1"
|
|
self.synchronizer.log = Mock()
|
|
|
|
@patch('requests.Session')
|
|
@patch('os.makedirs')
|
|
@patch('builtins.open', new_callable=unittest.mock.mock_open)
|
|
def test_remote_fetch_success(self, mock_open, mock_makedirs, mock_session):
|
|
remote_path = "img-001.png"
|
|
local_path = "config/warning_images"
|
|
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'Test content'
|
|
mock_session.return_value.__enter__.return_value.get.return_value = mock_response
|
|
|
|
result = self.synchronizer.remote_fetch(remote_path, local_path)
|
|
|
|
self.assertEqual(result, {"downloaded_file": f"{local_path}/{remote_path}"})
|
|
mock_open.assert_called_with(f"{local_path}/{remote_path}", "wb")
|
|
mock_open().write.assert_called_with(b'Test content')
|
|
|
|
@patch('requests.Session')
|
|
def test_remote_fetch_not_found(self, mock_session):
|
|
remote_path = "img-001.png"
|
|
local_path = "config/warning_images"
|
|
|
|
mock_response = Mock()
|
|
mock_response.status_code = 404
|
|
mock_session.return_value.__enter__.return_value.get.return_value = mock_response
|
|
|
|
result = self.synchronizer.remote_fetch(remote_path, local_path)
|
|
self.assertEqual(result, {"error": "File not found"})
|
|
|
|
@patch('requests.Session')
|
|
def test_remote_fetch_forbidden(self, mock_session):
|
|
remote_path = "img-001.png"
|
|
local_path = "config/warning_images"
|
|
|
|
mock_response = Mock()
|
|
mock_response.status_code = 403
|
|
mock_session.return_value.__enter__.return_value.get.return_value = mock_response
|
|
|
|
result = self.synchronizer.remote_fetch(remote_path, local_path)
|
|
self.assertEqual(result, {"error": "Access forbidden or not logged in"})
|
|
|
|
@patch('requests.Session')
|
|
@patch('requests.Session.get', side_effect=requests.ConnectionError("Connection error"))
|
|
def test_remote_fetch_connection_error(self, mock_get, mock_session):
|
|
remote_path = "img-001.png"
|
|
local_path = "config/warning_images"
|
|
|
|
result = self.synchronizer.remote_fetch(remote_path, local_path)
|
|
self.assertEqual(result, {"error": "Connection error"})
|
|
self.synchronizer.log.error.assert_called_with(
|
|
"Connection error occurred while fetching the file: Connection error")
|
|
|
|
@patch('requests.Session')
|
|
@patch('requests.Session.get', side_effect=requests.Timeout("Timeout error"))
|
|
def test_remote_fetch_timeout_error(self, mock_get, mock_session):
|
|
remote_path = "img-001.png"
|
|
local_path = "config/warning_images"
|
|
|
|
result = self.synchronizer.remote_fetch(remote_path, local_path)
|
|
self.assertEqual(result, {"error": "Timeout error"})
|
|
self.synchronizer.log.error.assert_called_with("Timeout error occurred while fetching the file: Timeout error")
|
|
|
|
@patch('requests.Session')
|
|
def test_parse_response_and_execute_success(self, mock_session):
|
|
mock_response = Mock()
|
|
mock_response.json.return_value = {
|
|
"STATUS": "OK",
|
|
"ACTIONS_TO_DO": [
|
|
{"remote_path": "img-001.png", "local_path": "config/warning_images"}
|
|
]
|
|
}
|
|
|
|
with patch.object(self.synchronizer, 'remote_fetch',
|
|
return_value={"downloaded_file": "config/warning_images/img-001.png"}) as mock_rf:
|
|
self.synchronizer.parse_response_and_execute(mock_response)
|
|
mock_rf.assert_called_once_with(remote_path="img-001.png", local_path="config/warning_images")
|
|
|
|
def test_parse_response_and_execute_json_error(self):
|
|
mock_response = Mock()
|
|
mock_response.json.side_effect = json.JSONDecodeError("Expecting value", "doc", 0)
|
|
|
|
self.synchronizer.parse_response_and_execute(mock_response)
|
|
self.synchronizer.log.error.assert_called_with("Failed to decode JSON response")
|
|
|
|
@patch('requests.Session')
|
|
@patch.object(ArchiveSynchronizer, 'parse_response_and_execute')
|
|
def test_update_machine_status_success(self, mock_parse_response, mock_session):
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_session.return_value.__enter__.return_value.post.return_value = mock_response
|
|
|
|
self.synchronizer.machine_status = "running"
|
|
result = self.synchronizer.update_machine_status()
|
|
self.assertTrue(result)
|
|
mock_parse_response.assert_called_once_with(mock_response)
|
|
self.synchronizer.log.info.assert_called_with("Status: running: Machine Status Updated Successfully")
|
|
|
|
@patch('requests.Session')
|
|
def test_update_machine_status_bad_status_response(self, mock_session):
|
|
mock_response = Mock()
|
|
mock_response.status_code = 500
|
|
mock_session.return_value.__enter__.return_value.post.return_value = mock_response
|
|
|
|
self.synchronizer.machine_status = "idle"
|
|
result = self.synchronizer.update_machine_status()
|
|
self.assertFalse(result)
|
|
self.synchronizer.log.warning.assert_called_with(
|
|
"Status: idle: failed to update machine status: bad status response: 500: no response"
|
|
)
|
|
|
|
@patch('requests.Session', side_effect=requests.ConnectionError("Connection error"))
|
|
def test_update_machine_status_connection_error(self, mock_session):
|
|
self.synchronizer.machine_status = "offline"
|
|
result = self.synchronizer.update_machine_status()
|
|
self.assertFalse(result)
|
|
self.synchronizer.log.warning.assert_called_with(
|
|
"Status: offline: failed to update machine status, archive_endpoint might be unreachable: Connection error"
|
|
)
|
|
|
|
@patch('requests.Session', side_effect=requests.Timeout("Timeout error"))
|
|
def test_update_machine_status_timeout_error(self, mock_session):
|
|
self.synchronizer.machine_status = "offline"
|
|
result = self.synchronizer.update_machine_status()
|
|
self.assertFalse(result)
|
|
self.synchronizer.log.warning.assert_called_with(
|
|
"Status: offline: failed to update machine status, archive_endpoint might be unreachable: Timeout error"
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|