diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..8ff8196 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,4 @@ +# .coveragerc +[run] +omit = + scouter/worker/* \ No newline at end of file diff --git a/.gitignore b/.gitignore index a34260d..a446ab3 100644 --- a/.gitignore +++ b/.gitignore @@ -32,4 +32,8 @@ __pycache__/ *.tmp *.bak *.old -.secret \ No newline at end of file +.secret + +# Ignorar coverage +htmlcov/ +.coverage \ No newline at end of file diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index 58833ea..c9f1395 100644 --- a/scouter/activities/gates.py +++ b/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': diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index c0780a1..17cfe4a 100644 --- a/scouter/activities/redis.py +++ b/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']] = np.nan - - else: - data_hold[row['name']] = value + 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") diff --git a/scouter/utils/connectors_config.py b/scouter/utils/connectors_config.py new file mode 100644 index 0000000..f36da92 --- /dev/null +++ b/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')), + } diff --git a/scouter/utils/logger.py b/scouter/utils/logger.py new file mode 100644 index 0000000..42a9cfd --- /dev/null +++ b/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 diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index 67428e7..c0be8f2 100644 --- a/scouter/worker/worker.py +++ b/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 - - -def build_postgres_config(): - return { - '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')), - } + from scouter.utils.logger import get_logger + from scouter.utils.connectors_config import ( + build_postgres_config, + build_kafka_config, + build_redis_config + ) 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( - logging.Formatter( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - ) - ) - - logger.addHandler(stream_handler) + logger = get_logger(__name__) logger.info('Starting Worker...') diff --git a/tests/activities/test_activities.py b/tests/activities/test_activities.py new file mode 100644 index 0000000..2a42392 --- /dev/null +++ b/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'] diff --git a/tests/activities/test_faker.py b/tests/activities/test_faker.py index 791ed72..992cfb3 100644 --- a/tests/activities/test_faker.py +++ b/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( - 'test_topic', value=expected_data) + }, { + 'tag': 'ns=1;i=1002', + '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 diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py index d954d88..9569d57 100644 --- a/tests/activities/test_gates.py +++ b/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 diff --git a/tests/activities/test_kafka.py b/tests/activities/test_kafka.py index 1e8cac3..d38abfd 100644 --- a/tests/activities/test_kafka.py +++ b/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) diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py new file mode 100644 index 0000000..48e6ed8 --- /dev/null +++ b/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 + } diff --git a/tests/utils/test_logger.py b/tests/utils/test_logger.py new file mode 100644 index 0000000..b931714 --- /dev/null +++ b/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 diff --git a/tests/workflow/test_scouter.py b/tests/workflow/test_scouter.py index 181aec2..8cc0165 100644 --- a/tests/workflow/test_scouter.py +++ b/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()