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

Tests fixed

Browse source
This commit is contained in:
Cyber MacGeddon 2026-04-16 17:18:54 +01:00
parent 7cf424a3b2
commit 9635d3eb2a
1 changed files with 77 additions and 171 deletions
Download patch file Download diff file Expand all files Collapse all files

248
tests/unit/test_base/test_flow_processor.py
View file

@ -1,62 +1,50 @@
""" """
Unit tests for trustgraph.base.flow_processor Unit tests for trustgraph.base.flow_processor
Starting small with a single test to verify basic functionality
""" """
import pytest import pytest
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from unittest import IsolatedAsyncioTestCase from unittest import IsolatedAsyncioTestCase
# Import the service under test
from trustgraph.base.flow_processor import FlowProcessor from trustgraph.base.flow_processor import FlowProcessor
# Patches needed to let AsyncProcessor.__init__ run without real
# infrastructure while still setting self.id correctly.
ASYNC_PROCESSOR_PATCHES = [
patch('trustgraph.base.async_processor.get_pubsub', return_value=MagicMock()),
patch('trustgraph.base.async_processor.ProcessorMetrics', return_value=MagicMock()),
patch('trustgraph.base.async_processor.Consumer', return_value=MagicMock()),
]
def with_async_processor_patches(func):
"""Apply all AsyncProcessor dependency patches to a test."""
for p in reversed(ASYNC_PROCESSOR_PATCHES):
func = p(func)
return func
class TestFlowProcessorSimple(IsolatedAsyncioTestCase): class TestFlowProcessorSimple(IsolatedAsyncioTestCase):
"""Test FlowProcessor base class functionality""" """Test FlowProcessor base class functionality"""
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_flow_processor_initialization_basic(self, *mocks):
async def test_flow_processor_initialization_basic(self, mock_register_config, mock_async_init):
"""Test basic FlowProcessor initialization""" """Test basic FlowProcessor initialization"""
# Arrange
def set_id(**params):
pass
mock_async_init.side_effect = set_id
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-flow-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
# Act
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
# Assert assert processor.id == 'test-flow-processor'
# Verify AsyncProcessor.__init__ was called
mock_async_init.assert_called_once()
# Verify register_config_handler was called with the correct handler
mock_register_config.assert_called_once_with(
processor.on_configure_flows, types=["processor:test-flow-processor"]
)
# Verify FlowProcessor-specific initialization
assert hasattr(processor, 'flows')
assert processor.flows == {} assert processor.flows == {}
assert hasattr(processor, 'specifications')
assert processor.specifications == [] assert processor.specifications == []
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_register_specification(self, *mocks):
async def test_register_specification(self, mock_register_config, mock_async_init):
"""Test registering a specification""" """Test registering a specification"""
# Arrange
def set_id(**params):
pass
mock_async_init.side_effect = set_id
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-flow-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
@ -66,221 +54,162 @@ class TestFlowProcessorSimple(IsolatedAsyncioTestCase):
mock_spec = MagicMock() mock_spec = MagicMock()
mock_spec.name = 'test-spec' mock_spec.name = 'test-spec'
# Act
processor.register_specification(mock_spec) processor.register_specification(mock_spec)
# Assert
assert len(processor.specifications) == 1 assert len(processor.specifications) == 1
assert processor.specifications[0] == mock_spec assert processor.specifications[0] == mock_spec
@patch('trustgraph.base.flow_processor.Flow') @patch('trustgraph.base.flow_processor.Flow')
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_start_flow(self, *mocks):
async def test_start_flow(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test starting a flow""" """Test starting a flow"""
# Arrange mock_flow_class = mocks[-1]
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor' # Set id for Flow creation
mock_flow = AsyncMock() mock_flow = AsyncMock()
mock_flow_class.return_value = mock_flow mock_flow_class.return_value = mock_flow
flow_name = 'test-flow' flow_name = 'test-flow'
flow_defn = {'config': 'test-config'} flow_defn = {'config': 'test-config'}
# Act
await processor.start_flow(flow_name, flow_defn) await processor.start_flow(flow_name, flow_defn)
# Assert
assert flow_name in processor.flows assert flow_name in processor.flows
# Verify Flow was created with correct parameters mock_flow_class.assert_called_once_with(
mock_flow_class.assert_called_once_with('test-processor', flow_name, processor, flow_defn) 'test-processor', flow_name, processor, flow_defn
# Verify the flow's start method was called )
mock_flow.start.assert_called_once() mock_flow.start.assert_called_once()
@patch('trustgraph.base.flow_processor.Flow') @patch('trustgraph.base.flow_processor.Flow')
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_stop_flow(self, *mocks):
async def test_stop_flow(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test stopping a flow""" """Test stopping a flow"""
# Arrange mock_flow_class = mocks[-1]
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor'
mock_flow = AsyncMock() mock_flow = AsyncMock()
mock_flow_class.return_value = mock_flow mock_flow_class.return_value = mock_flow
flow_name = 'test-flow'
flow_defn = {'config': 'test-config'}
# Start a flow first flow_name = 'test-flow'
await processor.start_flow(flow_name, flow_defn) await processor.start_flow(flow_name, {'config': 'test-config'})
# Act
await processor.stop_flow(flow_name) await processor.stop_flow(flow_name)
# Assert
assert flow_name not in processor.flows assert flow_name not in processor.flows
mock_flow.stop.assert_called_once() mock_flow.stop.assert_called_once()
@patch('trustgraph.base.flow_processor.Flow') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') async def test_stop_flow_not_exists(self, *mocks):
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler')
async def test_stop_flow_not_exists(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test stopping a flow that doesn't exist""" """Test stopping a flow that doesn't exist"""
# Arrange
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-flow-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
# Act - should not raise an exception
await processor.stop_flow('non-existent-flow') await processor.stop_flow('non-existent-flow')
# Assert - flows dict should still be empty
assert processor.flows == {} assert processor.flows == {}
@patch('trustgraph.base.flow_processor.Flow') @patch('trustgraph.base.flow_processor.Flow')
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_on_configure_flows_basic(self, *mocks):
async def test_on_configure_flows_basic(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test basic flow configuration handling""" """Test basic flow configuration handling"""
# Arrange mock_flow_class = mocks[-1]
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor'
mock_flow = AsyncMock() mock_flow = AsyncMock()
mock_flow_class.return_value = mock_flow mock_flow_class.return_value = mock_flow
# Configuration with flows for this processor
flow_config = {
'test-flow': {'config': 'test-config'}
}
config_data = { config_data = {
'processor:test-processor': { 'processor:test-processor': {
'test-flow': '{"config": "test-config"}' 'test-flow': '{"config": "test-config"}'
} }
} }
# Act
await processor.on_configure_flows(config_data, version=1) await processor.on_configure_flows(config_data, version=1)
# Assert
assert 'test-flow' in processor.flows assert 'test-flow' in processor.flows
mock_flow_class.assert_called_once_with('test-processor', 'test-flow', processor, {'config': 'test-config'}) mock_flow_class.assert_called_once_with(
'test-processor', 'test-flow', processor,
{'config': 'test-config'}
)
mock_flow.start.assert_called_once() mock_flow.start.assert_called_once()
@patch('trustgraph.base.flow_processor.Flow') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') async def test_on_configure_flows_no_config(self, *mocks):
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler')
async def test_on_configure_flows_no_config(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test flow configuration handling when no config exists for this processor""" """Test flow configuration handling when no config exists for this processor"""
# Arrange
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor'
# Configuration without flows for this processor
config_data = { config_data = {
'processor:other-processor': { 'processor:other-processor': {
'other-flow': '{"config": "other-config"}' 'other-flow': '{"config": "other-config"}'
} }
} }
# Act
await processor.on_configure_flows(config_data, version=1) await processor.on_configure_flows(config_data, version=1)
# Assert
assert processor.flows == {} assert processor.flows == {}
mock_flow_class.assert_not_called()
@patch('trustgraph.base.flow_processor.Flow') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') async def test_on_configure_flows_invalid_config(self, *mocks):
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler')
async def test_on_configure_flows_invalid_config(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test flow configuration handling with invalid config format""" """Test flow configuration handling with invalid config format"""
# Arrange
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor'
# Configuration without active-flow key
config_data = { config_data = {
'other-data': 'some-value' 'other-data': 'some-value'
} }
# Act
await processor.on_configure_flows(config_data, version=1) await processor.on_configure_flows(config_data, version=1)
# Assert
assert processor.flows == {} assert processor.flows == {}
mock_flow_class.assert_not_called()
@patch('trustgraph.base.flow_processor.Flow') @patch('trustgraph.base.flow_processor.Flow')
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler') async def test_on_configure_flows_start_and_stop(self, *mocks):
async def test_on_configure_flows_start_and_stop(self, mock_register_config, mock_async_init, mock_flow_class):
"""Test flow configuration handling with starting and stopping flows""" """Test flow configuration handling with starting and stopping flows"""
# Arrange mock_flow_class = mocks[-1]
mock_async_init.return_value = None
mock_register_config.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
processor.id = 'test-processor'
mock_flow1 = AsyncMock() mock_flow1 = AsyncMock()
mock_flow2 = AsyncMock() mock_flow2 = AsyncMock()
mock_flow_class.side_effect = [mock_flow1, mock_flow2] mock_flow_class.side_effect = [mock_flow1, mock_flow2]
# First configuration - start flow1
config_data1 = { config_data1 = {
'processor:test-processor': { 'processor:test-processor': {
'flow1': '{"config": "config1"}' 'flow1': '{"config": "config1"}'
@ -289,69 +218,46 @@ class TestFlowProcessorSimple(IsolatedAsyncioTestCase):
await processor.on_configure_flows(config_data1, version=1) await processor.on_configure_flows(config_data1, version=1)
# Second configuration - stop flow1, start flow2
config_data2 = { config_data2 = {
'processor:test-processor': { 'processor:test-processor': {
'flow2': '{"config": "config2"}' 'flow2': '{"config": "config2"}'
} }
} }
# Act
await processor.on_configure_flows(config_data2, version=2) await processor.on_configure_flows(config_data2, version=2)
# Assert
# flow1 should be stopped and removed
assert 'flow1' not in processor.flows assert 'flow1' not in processor.flows
mock_flow1.stop.assert_called_once() mock_flow1.stop.assert_called_once()
# flow2 should be started and added
assert 'flow2' in processor.flows assert 'flow2' in processor.flows
mock_flow2.start.assert_called_once() mock_flow2.start.assert_called_once()
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') @with_async_processor_patches
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler')
@patch('trustgraph.base.async_processor.AsyncProcessor.start') @patch('trustgraph.base.async_processor.AsyncProcessor.start')
async def test_start_calls_parent(self, mock_parent_start, mock_register_config, mock_async_init): async def test_start_calls_parent(self, mock_parent_start, *mocks):
"""Test that start() calls parent start method""" """Test that start() calls parent start method"""
# Arrange
def set_id(**params):
pass
mock_async_init.side_effect = set_id
mock_register_config.return_value = None
mock_parent_start.return_value = None mock_parent_start.return_value = None
config = { config = {
'id': 'test-flow-processor', 'id': 'test-flow-processor',
'taskgroup': AsyncMock() 'taskgroup': AsyncMock()
} }
processor = FlowProcessor(**config) processor = FlowProcessor(**config)
# Act
await processor.start() await processor.start()
# Assert
mock_parent_start.assert_called_once() mock_parent_start.assert_called_once()
@patch('trustgraph.base.async_processor.AsyncProcessor.__init__') async def test_add_args_calls_parent(self):
@patch('trustgraph.base.async_processor.AsyncProcessor.register_config_handler')
async def test_add_args_calls_parent(self, mock_register_config, mock_async_init):
"""Test that add_args() calls parent add_args method""" """Test that add_args() calls parent add_args method"""
# Arrange
def set_id(**params):
pass
mock_async_init.side_effect = set_id
mock_register_config.return_value = None
mock_parser = MagicMock() mock_parser = MagicMock()
# Act
with patch('trustgraph.base.async_processor.AsyncProcessor.add_args') as mock_parent_add_args: with patch('trustgraph.base.async_processor.AsyncProcessor.add_args') as mock_parent_add_args:
FlowProcessor.add_args(mock_parser) FlowProcessor.add_args(mock_parser)
# Assert
mock_parent_add_args.assert_called_once_with(mock_parser) mock_parent_add_args.assert_called_once_with(mock_parser)
if __name__ == '__main__': if __name__ == '__main__':
pytest.main([__file__]) pytest.main([__file__])