SIENTIAPDE-1005

Full coverage

Refactor code structure and improve logging configuration; update tests for activities and connectors

.coveragerc

@@ -0,0 +1,4 @@
+# .coveragerc
+[run]
+omit =
+    scouter/worker/*

.gitignore

@@ -32,4 +32,8 @@ __pycache__/
 *.tmp
 *.bak
 *.old
-.secret
+.secret
+
+# Ignorar coverage
+htmlcov/
+.coverage

scouter/activities/gates.py

@@ -96,9 +96,6 @@ class Gates(BaseActivity):
                 # Get the latest timestamp
                 latest_timestamp = group['timestamp'].max()
 
-                if group.empty:
-                    continue
-
                 aggr_value = self.apply_aggregation(group, aggr_function)
 
                 if aggr_value == 'continue':

scouter/activities/redis.py

@@ -64,11 +64,7 @@ class Redis(BaseActivity):
         for _, row in data.iterrows():
             value = row['value']
 
-            if value is None:
+            data_hold[row['name']] = value
-                data_hold[row['name']] = np.nan
-
-            else:
-                data_hold[row['name']] = value
 
         data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \
             datetime.now().strftime("%Y-%m-%d %H:%M:%S")

scouter/utils/connectors_config.py

@@ -0,0 +1,28 @@
+from os import getenv
+
+
+def build_postgres_config():
+    return {
+        'host': getenv('POSTGRES_HOST', 'localhost'),
+        'port': int(getenv('POSTGRES_PORT', '5432')),
+        'user': getenv('POSTGRES_USER', 'sientia'),
+        'password': getenv('POSTGRES_PASSWORD', 'sientia'),
+        'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
+        'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
+        'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
+    }
+
+
+def build_kafka_config():
+    return {
+        'bootstrap_servers': getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
+        'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')),
+        'group_id': 'scouter-group'
+    }
+
+
+def build_redis_config():
+    return {
+        'host': getenv('REDIS_HOST', 'localhost'),
+        'port': int(getenv('REDIS_PORT', '6379')),
+    }

scouter/utils/logger.py

@@ -0,0 +1,22 @@
+from os import getenv
+import logging
+import sys
+
+
+def get_logger(name: str):
+    log_level = getenv('LOG_LEVEL', 'INFO').upper()
+
+    logger = logging.getLogger(name)
+    logger.setLevel(log_level)
+    stream_handler = logging.StreamHandler(sys.stdout)
+    stream_handler.setLevel(log_level)
+
+    stream_handler.setFormatter(
+        logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        )
+    )
+
+    logger.addHandler(stream_handler)
+
+    return logger

scouter/worker/worker.py

@@ -3,7 +3,6 @@ from temporalio.worker import Worker
 
 with workflow.unsafe.imports_passed_through():
     import os
-    import logging
     from sientia_do.notifications.handlers import NotificationHandler
     from scouter.activities.activities import Activities
     from scouter.workflow.scouter import Scouter
@@ -11,53 +10,18 @@ with workflow.unsafe.imports_passed_through():
     from scouter.workflow.fake_data import FakeData
     from scouter.activities.faker import Faker
     import asyncio
-    import sys
+    from scouter.utils.logger import get_logger
-
+    from scouter.utils.connectors_config import (
-
+        build_postgres_config,
-def build_postgres_config():
+        build_kafka_config,
-    return {
+        build_redis_config
-        'host': os.getenv('POSTGRES_HOST', 'localhost'),
+    )
-        'port': int(os.getenv('POSTGRES_PORT', '5432')),
-        'user': os.getenv('POSTGRES_USER', 'sientia'),
-        'password': os.getenv('POSTGRES_PASSWORD', 'sientia'),
-        'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'),
-        'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '5')),
-        'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '20'))
-    }
-
-
-def build_kafka_config():
-    return {
-        'bootstrap_servers': os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
-        'polling_time': int(os.getenv('KAFKA_POLLING_TIME', '1000')),
-        'group_id': 'scouter-group'
-    }
-
-
-def build_redis_config():
-    return {
-        'host': os.getenv('REDIS_HOST', 'localhost'),
-        'port': int(os.getenv('REDIS_PORT', '6379')),
-    }
 
 
 async def main():
-
-    log_level = os.getenv('LOG_LEVEL', 'INFO').upper()
-
     host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
-    logger = logging.getLogger(__name__)
-    logger.setLevel(log_level)
-    stream_handler = logging.StreamHandler(sys.stdout)
-    stream_handler.setLevel(log_level)
 
-    stream_handler.setFormatter(
+    logger = get_logger(__name__)
-        logging.Formatter(
-            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-        )
-    )
-
-    logger.addHandler(stream_handler)
 
     logger.info('Starting Worker...')
 

tests/activities/test_activities.py

@@ -0,0 +1,143 @@
+from unittest.mock import patch, MagicMock, ANY
+from scouter.activities.activities import Activities
+from scouter.activities.postgres import Postgres
+from scouter.activities.redis import Redis
+from scouter.activities.kafka import Kafka
+from scouter.activities.gates import Gates
+
+
+@patch('scouter.activities.activities.Postgres.__init__')
+@patch('scouter.activities.activities.Redis.__init__')
+@patch('scouter.activities.activities.Kafka.__init__')
+@patch('scouter.activities.activities.Gates.__init__')
+def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgres_init):
+
+    postgres_config = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10
+    }
+
+    redis_config = {
+        'host': 'localhost',
+        'port': 6379
+    }
+
+    kafka_config = {
+        'bootstrap_servers': 'localhost:9092',
+        'polling_time': 1000,
+        'group_id': 'test-group'
+    }
+
+    logger = MagicMock()
+    notification_handler = MagicMock()
+
+    activities = Activities(
+        postgres_config=postgres_config,
+        redis_config=redis_config,
+        kafka_config=kafka_config,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    assert isinstance(activities, Activities)
+    assert isinstance(activities, Postgres)
+    assert isinstance(activities, Redis)
+    assert isinstance(activities, Kafka)
+    assert isinstance(activities, Gates)
+
+    mock_postgres_init.assert_called_once_with(
+        ANY,
+        host=postgres_config['host'],
+        port=postgres_config['port'],
+        user=postgres_config['user'],
+        password=postgres_config['password'],
+        dbname=postgres_config['dbname'],
+        min_connections=postgres_config['min_connections'],
+        max_connections=postgres_config['max_connections'],
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_redis_init.assert_called_once_with(
+        ANY,
+        host=redis_config['host'],
+        port=redis_config['port'],
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_kafka_init.assert_called_once_with(
+        ANY,
+        bootstrap_servers=kafka_config['bootstrap_servers'],
+        polling_time=kafka_config['polling_time'],
+        group_id=kafka_config['group_id'],
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    mock_gates_init.assert_called_once_with(
+        ANY,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+
+@patch('scouter.activities.activities.Postgres.__init__')
+@patch('scouter.activities.activities.Redis.__init__')
+@patch('scouter.activities.activities.Kafka.__init__')
+def test_prepare_activity(_mock_kafka_init,
+                          _mock_redis_init, _mock_postgres_init):
+    postgres_config = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10
+    }
+
+    redis_config = {
+        'host': 'localhost',
+        'port': 6379
+    }
+
+    kafka_config = {
+        'bootstrap_servers': 'localhost:9092',
+        'polling_time': 1000,
+        'group_id': 'test-group'
+    }
+
+    logger = MagicMock()
+    notification_handler = MagicMock()
+
+    activities = Activities(
+        postgres_config=postgres_config,
+        redis_config=redis_config,
+        kafka_config=kafka_config,
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    input_data = {
+        'workflow_name': 'test-workflow-name',
+        'schedule_name': 'test-schedule-name',
+        'model_name': 'test-model-name',
+        'model_id': 'test-model-id'
+    }
+
+    activities.prepare_activity(input_data)
+
+    assert activities.notification_handler.base_notification.pipeline_name == input_data[
+        'workflow_name']
+    assert activities.notification_handler.base_notification.schedule_name == input_data[
+        'schedule_name']
+    assert activities.notification_handler.base_notification.model_name == input_data[
+        'model_name']
+    assert activities.notification_handler.base_notification.model_id == input_data[
+        'model_id']

tests/activities/test_faker.py

@@ -1,8 +1,8 @@
-import pytest
-from unittest.mock import MagicMock, patch, call
-from scouter.activities.faker import Faker
 from logging import Logger
+from unittest.mock import MagicMock, patch, call
+import pytest
 from sientia_do.notifications.handlers import NotificationHandler
+from scouter.activities.faker import Faker
 
 
 @pytest.fixture
@@ -39,12 +39,14 @@ async def test_faker_init(faker_instance, mock_kafka_producer):
 
 
 @pytest.mark.asyncio
-async def test_generate_and_send_data_default_count(faker_instance, mock_kafka_producer, mock_datetime):
+async def test_generate_and_send_data_default_count(faker_instance,
+                                                    mock_kafka_producer, mock_datetime):
     """Test generating data with default message count"""
     # Mock random.choice to control the output
     with patch('random.choice') as mock_choice, \
             patch('random.uniform', return_value=42.5), \
-            patch('random.randint', return_value=3):
+            patch('random.randint', return_value=3), \
+            patch('random.random', return_value=0.5):
 
         # Setup mock for tag and name selection
         mock_choice.side_effect = [
@@ -61,14 +63,27 @@ async def test_generate_and_send_data_default_count(faker_instance, mock_kafka_p
         mock_kafka_producer.flush.assert_called_once()
 
         # Verify the message format
-        expected_data = {
+        expected_data = [{
             'tag': 'ns=1;i=1001',
             'name': 'Temperature Sensor',
             'timestamp': '2025-05-14 14:54:24',
             'value': 42.5
-        }
+        }, {
-        mock_kafka_producer.send.assert_any_call(
+            'tag': 'ns=1;i=1002',
-            'test_topic', value=expected_data)
+            'name': 'Vibration Meter',
+            'timestamp': '2025-05-14 14:54:24',
+            'value': 42.5
+        }, {
+            'tag': 'ns=1;i=1003',
+            'name': 'Pressure Gauge',
+            'timestamp': '2025-05-14 14:54:24',
+            'value': 42.5
+        }]
+        mock_kafka_producer.send.assert_has_calls([
+            call('test_topic', value=expected_data[0]),
+            call('test_topic', value=expected_data[1]),
+            call('test_topic', value=expected_data[2])
+        ])
 
 
 @pytest.mark.asyncio
@@ -106,3 +121,22 @@ async def test_generate_and_send_data_random_values(faker_instance, mock_kafka_p
     assert call_args['tag'] in faker_instance.tags
     assert 'value' in call_args
     assert 0 <= call_args['value'] <= 100
+
+
+@pytest.mark.asyncio
+@patch('scouter.activities.faker.random.random', return_value=0.05)
+async def test_generate_and_send_data_generate_null_values(
+        _random_mock,
+        faker_instance,
+        mock_kafka_producer):
+    # Call the method
+    await faker_instance.generate_and_send_data({'topic': 'test_topic',
+                                                 'num_messages': 1})
+
+    # Get the call arguments
+    call_args = mock_kafka_producer.send.call_args[1]['value']
+
+    assert call_args['value'] is None
+    assert call_args['tag'] in faker_instance.tags
+    assert 'name' in call_args
+    assert 'timestamp' in call_args

tests/activities/test_gates.py

@@ -1,4 +1,4 @@
-from unittest.mock import Mock, patch, MagicMock
+from unittest.mock import Mock, patch, MagicMock, ANY
 import numpy as np
 import pandas as pd
 import pytest
@@ -288,3 +288,74 @@ async def test_aggregate_data(gates_fixture):
     # Verify
     assert result == expected_result
     gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_aggregate_data_with_continue(gates_fixture):
+
+    gates_fixture.apply_aggregation = MagicMock(return_value='continue')
+
+    input_data = {
+        'data': [
+            {'tag': 'tag1', 'name': 'name1', 'value': 1.0, 'timestamp': '2023-01-01'},
+            {'tag': 'tag1', 'name': 'name1', 'value': 2.0, 'timestamp': '2023-01-02'},
+            {'tag': 'tag1', 'name': 'name1', 'value': 3.0, 'timestamp': '2023-01-03'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 4.0, 'timestamp': '2023-01-01'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 5.0, 'timestamp': '2023-01-02'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 6.0, 'timestamp': '2023-01-03'},
+            {'tag': 'tag1', 'name': 'name1',
+                'value': None, 'timestamp': '2023-01-04'},
+        ],
+        'model_tags': {
+            'name1': {'aggr_function': 'avg'},
+            'name2': {'aggr_function': 'max'},
+        }
+    }
+
+    # Expected result
+    expected_result = {}
+
+    # Execute
+    result = await gates_fixture.aggregate_data(input_data)
+
+    # Verify
+    assert result == expected_result
+    gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_aggregate_data_raise_exception(gates_fixture):
+
+    gates_fixture.apply_aggregation = MagicMock(
+        side_effect=Exception("Test exception"))
+
+    input_data = {
+        'data': [
+            {'tag': 'tag1', 'name': 'name1', 'value': 1.0, 'timestamp': '2023-01-01'},
+            {'tag': 'tag1', 'name': 'name1', 'value': 2.0, 'timestamp': '2023-01-02'},
+            {'tag': 'tag1', 'name': 'name1', 'value': 3.0, 'timestamp': '2023-01-03'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 4.0, 'timestamp': '2023-01-01'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 5.0, 'timestamp': '2023-01-02'},
+            {'tag': 'tag2', 'name': 'name2', 'value': 6.0, 'timestamp': '2023-01-03'},
+            {'tag': 'tag1', 'name': 'name1',
+                'value': None, 'timestamp': '2023-01-04'},
+        ],
+        'model_tags': {
+            'name1': {'aggr_function': 'avg'},
+            'name2': {'aggr_function': 'max'},
+        }
+    }
+
+    try:
+        await gates_fixture.aggregate_data(input_data)
+    except Exception as e:
+        assert str(e) == "Test exception"
+        gates_fixture.notification_handler.build_and_send_notification.assert_called_once_with(
+            notification_id="AGGREGATION_ISSUES",
+            message="Error aggregating data: Test exception",
+            block="aggregate_data",
+            level=NotificationLevel.ERROR,
+            attachment_content=ANY
+        )
+    else:
+        assert False

tests/activities/test_kafka.py

@@ -63,3 +63,20 @@ async def test_load_from_kafka(kafka):
     kafka.kafka_connector.subscribe.assert_called_once_with(["test-topic"])
 
     kafka.kafka_connector.poll.assert_called_once_with(timeout_ms=1000)
+
+
+@mark.asyncio
+async def test_load_from_kafka_empty(kafka):
+    input_data = {"topic": "test-topic"}
+
+    kafka.kafka_connector.poll.return_value = MagicMock(
+        items=MagicMock(return_value=[])
+    )
+
+    result = await kafka.load_from_kafka(input_data)
+
+    assert result == {}
+
+    kafka.kafka_connector.subscribe.assert_called_once_with(["test-topic"])
+
+    kafka.kafka_connector.poll.assert_called_once_with(timeout_ms=1000)

tests/utils/test_connectors_config.py

@@ -0,0 +1,109 @@
+import os
+from unittest.mock import patch
+import pytest
+from scouter.utils.connectors_config import (
+    build_postgres_config,
+    build_kafka_config,
+    build_redis_config
+)
+
+
+@pytest.fixture
+def mock_env_vars():
+    with patch.dict(os.environ, {}, clear=True):
+        yield
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_postgres_config_defaults():
+    """Test that build_postgres_config returns default values when no env vars are set"""
+    config = build_postgres_config()
+
+    assert config == {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'sientia',
+        'password': 'sientia',
+        'dbname': 'sientia',
+        'min_connections': 5,
+        'max_connections': 20
+    }
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_postgres_config_with_env_vars():
+    """Test that build_postgres_config uses env vars when set"""
+    with patch.dict(os.environ, {
+        'POSTGRES_HOST': 'db.example.com',
+        'POSTGRES_PORT': '5433',
+        'POSTGRES_USER': 'admin',
+        'POSTGRES_PASSWORD': 'secret',
+        'POSTGRES_DBNAME': 'test_db',
+        'POSTGRES_MIN_CONNECTIONS': '3',
+        'POSTGRES_MAX_CONNECTIONS': '15'
+    }):
+        config = build_postgres_config()
+
+        assert config == {
+            'host': 'db.example.com',
+            'port': 5433,
+            'user': 'admin',
+            'password': 'secret',
+            'dbname': 'test_db',
+            'min_connections': 3,
+            'max_connections': 15
+        }
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_kafka_config_defaults():
+    """Test that build_kafka_config returns default values when no env vars are set"""
+    config = build_kafka_config()
+
+    assert config == {
+        'bootstrap_servers': 'localhost:9092',
+        'polling_time': 1000,
+        'group_id': 'scouter-group'
+    }
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_kafka_config_with_env_vars():
+    """Test that build_kafka_config uses env vars when set"""
+    with patch.dict(os.environ, {
+        'KAFKA_BOOTSTRAP_SERVERS': 'kafka.example.com:9092',
+        'KAFKA_POLLING_TIME': '5000'
+    }):
+        config = build_kafka_config()
+
+        assert config == {
+            'bootstrap_servers': 'kafka.example.com:9092',
+            'polling_time': 5000,
+            'group_id': 'scouter-group'
+        }
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_redis_config_defaults():
+    """Test that build_redis_config returns default values when no env vars are set"""
+    config = build_redis_config()
+
+    assert config == {
+        'host': 'localhost',
+        'port': 6379
+    }
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+def test_build_redis_config_with_env_vars():
+    """Test that build_redis_config uses env vars when set"""
+    with patch.dict(os.environ, {
+        'REDIS_HOST': 'redis.example.com',
+        'REDIS_PORT': '6380'
+    }):
+        config = build_redis_config()
+
+        assert config == {
+            'host': 'redis.example.com',
+            'port': 6380
+        }

tests/utils/test_logger.py

@@ -0,0 +1,37 @@
+import os
+from unittest.mock import patch
+import logging
+import pytest
+from scouter.utils.logger import get_logger
+
+
+@pytest.fixture
+def mock_env_vars():
+    with patch.dict(os.environ, {}, clear=True):
+        yield
+
+
+@pytest.mark.usefixtures("mock_env_vars")
+@patch('scouter.utils.logger.logging.Formatter')
+@patch('scouter.utils.logger.logging.StreamHandler')
+def test_get_logger_defaults(mock_stream_handler, mock_formatter):
+    """Test logger creation with default settings"""
+    # Mock the StreamHandler and Formatter
+
+    logger = get_logger('test_logger')
+
+    # Verify logger settings
+    assert logger.name == 'test_logger'
+    assert logger.level == logging.INFO
+
+    # Verify handler configuration
+    mock_stream_handler.return_value.setLevel.assert_called_once_with('INFO')
+    mock_stream_handler.return_value.setFormatter.assert_called_once()
+
+    # Verify formatter configuration
+    mock_formatter.assert_called_once_with(
+        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+
+    # Verify handler was added to logger
+    assert len(logger.handlers) == 1

tests/workflow/test_scouter.py

@@ -36,3 +36,25 @@ async def test_scouter_workflow(mock_workflow, scouter):
             'data': 'test_data'
         }
     )
+
+
+@mark.asyncio
+@patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock)
+async def test_scouter_workflow_empty(mock_workflow, scouter):
+    mock_workflow.execute_activity_method.return_value = {}
+    await scouter.run(
+        input_data={
+            'topic': 'test_topic'
+        }
+    )
+
+    mock_workflow.execute_activity_method.assert_called_once_with(
+        Activities.load_from_kafka,
+        {
+            'topic': 'test_topic'
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY
+    )
+
+    mock_workflow.execute_child_workflow.assert_not_called()