Stanislav Hejny
226 lines
8.5 KiB
Python
226 lines
8.5 KiB
Python
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from aiohttp import ClientResponse
|
|
|
|
from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.model.model import DataMessage, DataResponse
|
|
|
|
|
|
class TestCleverThisServiceAdapter(unittest.TestCase):
|
|
def setUp(self):
|
|
# Mock AMQConfiguration
|
|
self.mock_amq_config = AMQConfiguration("./application.properties.local")
|
|
|
|
# Mock ClientSession
|
|
self.mock_session = AsyncMock()
|
|
|
|
# Create adapter instance
|
|
self.adapter = CleverThisServiceAdapter(
|
|
amq_configuration=self.mock_amq_config, routes=[], session=self.mock_session
|
|
)
|
|
|
|
def test_init(self):
|
|
self.assertEqual(self.adapter.service_host, "localhost")
|
|
self.assertEqual(self.adapter.service_port, 8080)
|
|
self.assertFalse(self.adapter.require_authenticated_user)
|
|
|
|
def test_get_content_type(self):
|
|
test_cases = [
|
|
({"Content-Type": ["application/json"]}, "application/json"),
|
|
({"content-type": ["text/xml"]}, "text/xml"),
|
|
({"CONTENT-TYPE": ["text/plain; charset=utf-8"]}, "text/plain"),
|
|
({}, "application/json"),
|
|
({"Other-Header": ["value"]}, "application/json"),
|
|
]
|
|
|
|
for headers, expected in test_cases:
|
|
with self.subTest(headers=headers):
|
|
result = self.adapter.get_content_type(headers)
|
|
self.assertEqual(result, expected)
|
|
|
|
async def test_get_current_user(self):
|
|
# Base implementation should return None
|
|
result = await self.adapter.get_current_user("test_token")
|
|
self.assertIsNone(result)
|
|
|
|
@patch.object(CleverThisServiceAdapter, "get_current_user")
|
|
async def test_on_message_unauthenticated(self, mock_get_user):
|
|
mock_message = MagicMock(spec=AMQMessage)
|
|
mock_message.headers = {}
|
|
mock_message.path = "/some/path"
|
|
mock_message.method = "GET"
|
|
mock_message.id = "123"
|
|
mock_message.trace_info = {}
|
|
|
|
# Test when authentication is required but not provided
|
|
response = await self.adapter.on_message(mock_message)
|
|
self.assertEqual(response.response_code(), "401")
|
|
mock_get_user.assert_not_called()
|
|
|
|
@patch.object(CleverThisServiceAdapter, "get_current_user")
|
|
@patch.object(CleverThisServiceAdapter, "get")
|
|
async def test_on_message_authenticated(self, mock_get, mock_get_user):
|
|
mock_message = MagicMock(spec=AMQMessage)
|
|
mock_message.headers = {"Authorization": ["Bearer valid_token"]}
|
|
mock_message.path = "/some/path"
|
|
mock_message.method = "GET"
|
|
mock_message.id = "123"
|
|
mock_message.trace_info = {}
|
|
|
|
mock_user = MagicMock()
|
|
mock_get_user.return_value = mock_user
|
|
|
|
mock_response = MagicMock(spec=DataResponse)
|
|
mock_get.return_value = mock_response
|
|
|
|
response = await self.adapter.on_message(mock_message)
|
|
|
|
self.assertEqual(response, mock_response)
|
|
mock_get_user.assert_called_once_with("valid_token")
|
|
mock_get.assert_called_once_with(mock_message, mock_user)
|
|
|
|
@patch.object(CleverThisServiceAdapter, "get_current_user")
|
|
async def test_on_message_login_path(self, mock_get_user):
|
|
mock_message = MagicMock(spec=AMQMessage)
|
|
mock_message.headers = {}
|
|
mock_message.path = "/login"
|
|
mock_message.method = "POST"
|
|
mock_message.id = "123"
|
|
mock_message.trace_info = {}
|
|
|
|
mock_response = MagicMock(spec=DataResponse)
|
|
with patch.object(self.adapter, "post", return_value=mock_response) as mock_post:
|
|
response = await self.adapter.on_message(mock_message)
|
|
self.assertEqual(response, mock_response)
|
|
mock_get_user.assert_not_called()
|
|
mock_post.assert_called_once()
|
|
|
|
@patch.object(CleverThisServiceAdapter, "get_current_user")
|
|
async def test_on_message_all_methods(self, mock_get_user):
|
|
mock_user = MagicMock()
|
|
mock_get_user.return_value = mock_user
|
|
|
|
mock_response = MagicMock(spec=DataResponse)
|
|
|
|
methods = ["GET", "PUT", "POST", "DELETE", "HEAD"]
|
|
for method in methods:
|
|
with self.subTest(method=method):
|
|
mock_message = MagicMock(spec=AMQMessage)
|
|
mock_message.headers = {"Authorization": ["Bearer valid_token"]}
|
|
mock_message.path = "/some/path"
|
|
mock_message.method = method
|
|
mock_message.id = "123"
|
|
mock_message.trace_info = {}
|
|
|
|
with patch.object(
|
|
self.adapter, method.lower(), return_value=mock_response
|
|
) as mock_method:
|
|
response = await self.adapter.on_message(mock_message)
|
|
self.assertEqual(response, mock_response)
|
|
mock_method.assert_called_once()
|
|
|
|
async def test_get_with_session(self):
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.path = "/api/test"
|
|
mock_message.headers = {"X-Test": "value"}
|
|
mock_message.trace_info = {"trace-id": "123"}
|
|
mock_message.id = "msg123"
|
|
|
|
mock_response = MagicMock(spec=ClientResponse)
|
|
mock_response.status = 200
|
|
mock_response.content_type = "application/json"
|
|
mock_response.read = AsyncMock(return_value=b'{"result": "success"}')
|
|
|
|
self.mock_session.get.return_value = mock_response
|
|
|
|
response = await self.adapter.get(mock_message, None)
|
|
|
|
self.assertEqual(response.response_code(), "200")
|
|
self.mock_session.get.assert_called_once_with(
|
|
"http://testhost:8080/api/test",
|
|
headers={"X-Test": "value", "trace-id": "123"},
|
|
)
|
|
|
|
async def test_get_without_session(self):
|
|
self.adapter.session = None
|
|
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.path = "/api/test"
|
|
mock_message.id = "msg123"
|
|
mock_message.trace_info = {}
|
|
|
|
response = await self.adapter.get(mock_message, None)
|
|
|
|
self.assertEqual(response.response_code(), "404")
|
|
|
|
async def test_put_with_form_handling(self):
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.path = "/api/test"
|
|
mock_message.headers = {}
|
|
mock_message.trace_info = {}
|
|
mock_message.id = "msg123"
|
|
mock_message.body = b"test=value"
|
|
|
|
mock_response = MagicMock(spec=ClientResponse)
|
|
mock_response.status = 200
|
|
mock_response.content_type = "application/json"
|
|
mock_response.read = AsyncMock(return_value=b'{"result": "success"}')
|
|
|
|
self.mock_session.put.return_value = mock_response
|
|
|
|
with patch.object(
|
|
self.adapter,
|
|
"_authorize_forward_request",
|
|
wraps=self.adapter._authorize_forward_request,
|
|
) as mock_handler:
|
|
response = await self.adapter.put(mock_message, None)
|
|
|
|
self.assertEqual(response.response_code(), "200")
|
|
mock_handler.assert_called_once()
|
|
|
|
async def test_post_with_form_handling(self):
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.path = "/api/test"
|
|
mock_message.headers = {}
|
|
mock_message.trace_info = {}
|
|
mock_message.id = "msg123"
|
|
mock_message.body = b"test=value"
|
|
|
|
mock_response = MagicMock(spec=ClientResponse)
|
|
mock_response.status = 201
|
|
mock_response.content_type = "application/json"
|
|
mock_response.read = AsyncMock(return_value=b'{"result": "created"}')
|
|
|
|
self.mock_session.post.return_value = mock_response
|
|
|
|
with patch.object(
|
|
self.adapter,
|
|
"_authorize_forward_request",
|
|
wraps=self.adapter._authorize_forward_request,
|
|
) as mock_handler:
|
|
response = await self.adapter.post(mock_message, None)
|
|
|
|
self.assertEqual(response.response_code(), "201")
|
|
mock_handler.assert_called_once()
|
|
|
|
async def test_head_not_implemented(self):
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.id = "msg123"
|
|
|
|
with self.assertRaises(NotImplementedError):
|
|
await self.adapter.head(mock_message, None)
|
|
|
|
async def test_delete_not_implemented(self):
|
|
mock_message = MagicMock(spec=DataMessage)
|
|
mock_message.id = "msg123"
|
|
|
|
# Assuming delete is not implemented in base class
|
|
with self.assertRaises(NotImplementedError):
|
|
await self.adapter.delete(mock_message, None)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|