apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-22 11:41:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Rev gateway tests

Browse source
This commit is contained in:
Cyber MacGeddon 2025-07-12 22:00:37 +01:00
parent 200506e555
commit 40253998fc
1 changed files with 22 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

44
tests/unit/test_rev_gateway/test_rev_gateway_service.py
View file

@ -119,46 +119,46 @@ class TestReverseGateway:
@patch('trustgraph.rev_gateway.service.ConfigReceiver')
@patch('trustgraph.rev_gateway.service.MessageDispatcher')
@patch('pulsar.Client')
@patch('aiohttp.ClientSession')
@pytest.mark.asyncio
async def test_reverse_gateway_connect_success(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver):
async def test_reverse_gateway_connect_success(self, mock_session_class, mock_pulsar_client, mock_dispatcher, mock_config_receiver):
"""Test ReverseGateway successful connection"""
mock_client_instance = MagicMock()
mock_pulsar_client.return_value = mock_client_instance
mock_session = AsyncMock()
mock_ws = AsyncMock()
mock_session.ws_connect.return_value = mock_ws
mock_session_class.return_value = mock_session
gateway = ReverseGateway()
with patch('aiohttp.ClientSession') as mock_session_class:
mock_session = AsyncMock()
mock_ws = AsyncMock()
mock_session.ws_connect.return_value = mock_ws
mock_session_class.return_value = mock_session
result = await gateway.connect()
assert result is True
assert gateway.session == mock_session
assert gateway.ws == mock_ws
mock_session.ws_connect.assert_called_once_with(gateway.url)
result = await gateway.connect()
assert result is True
assert gateway.session == mock_session
assert gateway.ws == mock_ws
mock_session.ws_connect.assert_called_once_with(gateway.url)
@patch('trustgraph.rev_gateway.service.ConfigReceiver')
@patch('trustgraph.rev_gateway.service.MessageDispatcher')
@patch('pulsar.Client')
@patch('aiohttp.ClientSession')
@pytest.mark.asyncio
async def test_reverse_gateway_connect_failure(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver):
async def test_reverse_gateway_connect_failure(self, mock_session_class, mock_pulsar_client, mock_dispatcher, mock_config_receiver):
"""Test ReverseGateway connection failure"""
mock_client_instance = MagicMock()
mock_pulsar_client.return_value = mock_client_instance
mock_session = AsyncMock()
mock_session.ws_connect.side_effect = Exception("Connection failed")
mock_session_class.return_value = mock_session
gateway = ReverseGateway()
with patch('aiohttp.ClientSession') as mock_session_class:
mock_session = AsyncMock()
mock_session.ws_connect.side_effect = Exception("Connection failed")
mock_session_class.return_value = mock_session
result = await gateway.connect()
assert result is False
result = await gateway.connect()
assert result is False
@patch('trustgraph.rev_gateway.service.ConfigReceiver')
@patch('trustgraph.rev_gateway.service.MessageDispatcher')