From 7958afbadf858a5b16da4b85d9b0e0c660f8a7e7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 8 Jan 2026 16:13:59 -0300 Subject: [PATCH] SIENTIAPDE-1478 Update pytest_asyncio fixture scopes in conftest.py for improved test isolation and add asyncio_default_fixture_loop_scope in pyproject.toml. Remove outdated scenarios from scenarios.md and delete unused test files for cleaner codebase. --- e2e/conftest.py | 27 +- e2e/scenarios.md | 19 - e2e/test_core_scouter_early_exit.py | 97 --- e2e/test_core_scouter_errors.py | 99 --- e2e/test_pi_web_api_scouter.py | 106 ---- e2e/test_pi_web_api_scouter_errors.py | 2 +- pyproject.toml | 1 + tests/utils/clients/test_pi_web_api_client.py | 588 ------------------ tests/utils/test_connectors_config.py | 160 ----- 9 files changed, 14 insertions(+), 1085 deletions(-) delete mode 100644 e2e/test_pi_web_api_scouter.py delete mode 100644 tests/utils/clients/test_pi_web_api_client.py diff --git a/e2e/conftest.py b/e2e/conftest.py index 9fe947b..34042fa 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -30,7 +30,7 @@ TEST_MONGODB_CONNECTION_STRING = 'mongodb://localhost:27017' TEST_DATABASE_NAME = 'test_db' -@pytest.fixture(scope='session') +@pytest_asyncio.fixture(scope='session') def postgres_container(): """ Create a PostgreSQL container using testcontainers. @@ -44,7 +44,7 @@ def postgres_container(): postgres.stop() -@pytest.fixture +@pytest_asyncio.fixture def postgres_engine(postgres_container): """ Create SQLAlchemy engine for PostgreSQL test database. @@ -81,7 +81,6 @@ def _create_schema_and_table(engine): # Create table WITHOUT partitioning (simpler for tests) # Same structure as production, but without PARTITION BY RANGE - # Use UNIQUE constraint directly since table is not partitioned create_table_sql = f""" CREATE TABLE IF NOT EXISTS {schema_name}.{table_name} ( id SERIAL NOT NULL, @@ -90,8 +89,7 @@ def _create_schema_and_table(engine): value numeric NULL, "timestamp" timestamptz NOT NULL, created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL, - PRIMARY KEY (id, created_at), - UNIQUE (model_id, timestamp, variable) + PRIMARY KEY (id, created_at) ); """ @@ -99,7 +97,7 @@ def _create_schema_and_table(engine): # Transaction is automatically committed when exiting the 'with' block -@pytest.fixture(autouse=True) +@pytest_asyncio.fixture(autouse=True) def setup_postgres_schema_and_table(postgres_engine): """ Automatically create necessary schema and table before each test. @@ -108,15 +106,14 @@ def setup_postgres_schema_and_table(postgres_engine): that the sientia_data schema and laborious_data table exist with the correct structure before tests execute. - Note: For tests, we use a non-partitioned table with a UNIQUE constraint - directly in the table definition, which is simpler and avoids issues + Note: For tests, we use a non-partitioned table which is simpler and avoids issues with pandas to_sql recognizing partitioned tables. """ _create_schema_and_table(postgres_engine) yield -@pytest.fixture +@pytest_asyncio.fixture def mock_logger(): """Mock logger for testing.""" logger = MagicMock(spec=Logger) @@ -128,7 +125,7 @@ def mock_logger(): return logger -@pytest.fixture +@pytest_asyncio.fixture def mock_mongo_client(): """ Mock MongoDB client to avoid real connections. @@ -153,7 +150,7 @@ def mock_mongo_client(): return mock_client -@pytest.fixture +@pytest_asyncio.fixture def notification_handler(mock_logger, mock_mongo_client): """ Create a real NotificationHandler instance with mocked MongoDB client. @@ -173,7 +170,7 @@ def notification_handler(mock_logger, mock_mongo_client): handler.shutdown() -@pytest.fixture +@pytest_asyncio.fixture def metrics_controller(mock_logger): """ Create a real MetricsController instance. @@ -186,7 +183,7 @@ def metrics_controller(mock_logger): # MetricsController might have cleanup, but it's optional -@pytest.fixture +@pytest_asyncio.fixture def mock_pi_web_api_client(): """Mock PI Web API client.""" mock_client = MagicMock() @@ -212,8 +209,8 @@ def mock_pi_web_api_client(): return mock_client -@pytest_asyncio.fixture -async def test_activities( +@pytest_asyncio.fixture(scope='function') +def test_activities( postgres_engine, postgres_container, mock_logger, diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 253c19e..edf97b7 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -299,25 +299,6 @@ The `pi_web_api_scouter` workflow: --- -#### Scenario 2.2.2: Zero Affected Rows After Export -**Description**: PostgreSQL export returns zero affected rows - -**Input**: -- Data that results in `affected_rows: 0` from export - -**Expected Behavior**: -- `export_data_to_postgres` returns `{'affected_rows': 0}` -- Workflow checks `if data_exported.get('affected_rows', 0) <= 0:` and returns early -- `write_metrics` NOT called -- `store_data_package` NOT called - -**Assertions**: -- Early return after export -- No metrics written -- Workflow completes without error - ---- - ### 2.3 Error Scenarios #### Scenario 2.3.1: Redis Connection Error diff --git a/e2e/test_core_scouter_early_exit.py b/e2e/test_core_scouter_early_exit.py index ca2e47f..2de7704 100644 --- a/e2e/test_core_scouter_early_exit.py +++ b/e2e/test_core_scouter_early_exit.py @@ -94,100 +94,3 @@ async def test_scenario_2_2_1_empty_data_after_grouping( # Should have 0 rows since export_data_to_postgres was not called assert row_count == 0, f"Expected no data in PostgreSQL, got {row_count} rows" - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_scenario_2_2_2_zero_affected_rows_after_export( - temporal_test_env: WorkflowEnvironment, - temporal_worker: Worker, - test_activities: Activities, - postgres_engine, -): - """ - Scenario 2.2.2: Zero Affected Rows After Export - - PostgreSQL export returns zero affected rows, workflow exits early. - - Note: This scenario is hard to test directly in e2e because we'd need to - simulate a conflict or other condition that results in 0 affected rows. - We'll test by inserting duplicate data first, then running the workflow again. - """ - client = temporal_test_env.client - - # First, insert some data directly to create a conflict scenario - test_data = [ - { - 'timestamp': '2024-01-01 12:00:00+0000', - 'name': 'tag1', - 'value': 10.5, - 'tag': 'webid1', - }, - ] - - # Insert data directly into PostgreSQL to create duplicates - schema_name = 'sientia_data' - table_name = 'laborious_data' - full_table_name = f"{schema_name}.{table_name}" - - with postgres_engine.connect() as conn: - conn.execute( - text(f""" - INSERT INTO {full_table_name} (model_id, variable, value, timestamp) - VALUES (1, 'tag1', 10.5, '2024-01-01 12:00:00+00:00') - ON CONFLICT (model_id, timestamp, variable) DO NOTHING - """) - ) - conn.commit() - - # Prepare input data with the same data (will result in conflict) - input_data = { - 'metadata': { - 'metadata': { - 'model_id': '1', - 'model_name': 'Test Model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'pi_web_api_scouter', - } - }, - 'workflow_name': 'pi_web_api_scouter', - 'schedule_name': 'test-schedule', - 'model_name': 'Test Model', - 'model_id': '1', - 'data': test_data, - 'trigger_laborious': False, - 'filters': {}, - 'schema': 'sientia_data', - 'table_name': 'laborious_data', - 'retention_time': 3600, - 'fill_missing_tags': False, - 'model_tags': { - 'tag1': { - 'webid': 'webid1', - 'aggr_function': 'avg', - 'data_range': [0, 100], - 'frequency': 60000, - }, - }, - } - - # Start workflow - handle = await client.start_workflow( - CoreScouter.run, - input_data, - id=f'test-core-scouter-zero-{datetime.now().timestamp()}', - task_queue='test-queue', - ) - - # Wait for workflow completion (should complete without error) - await handle.result() - - # Verify the count didn't increase (conflict handled, 0 affected rows) - with postgres_engine.connect() as conn: - result = conn.execute( - text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = 1") - ) - row_count = result.scalar() - - # Should still have 1 row (the original one, duplicate was ignored) - assert row_count == 1, f"Expected 1 row (duplicate ignored), got {row_count}" - diff --git a/e2e/test_core_scouter_errors.py b/e2e/test_core_scouter_errors.py index 6f723ef..e7d60f5 100644 --- a/e2e/test_core_scouter_errors.py +++ b/e2e/test_core_scouter_errors.py @@ -88,102 +88,3 @@ async def test_scenario_2_3_1_redis_connection_error( # Restore original method test_activities.redis_repository.get = original_get - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_scenario_2_3_2_postgresql_unique_constraint_violation( - temporal_test_env: WorkflowEnvironment, - temporal_worker: Worker, - test_activities: Activities, - postgres_engine, -): - """ - Scenario 2.3.2: PostgreSQL Unique Constraint Violation - - Duplicate data violates unique constraint, handled gracefully with ON CONFLICT DO NOTHING. - """ - client = temporal_test_env.client - - # Generate unique model_id to avoid conflicts with other tests - unique_id = int(datetime.now().timestamp() * 1000) % 1000000 - - # First, insert data directly to create a duplicate - schema_name = 'sientia_data' - table_name = 'laborious_data' - full_table_name = f"{schema_name}.{table_name}" - - with postgres_engine.connect() as conn: - conn.execute( - text(f""" - INSERT INTO {full_table_name} (model_id, variable, value, timestamp) - VALUES (:model_id, 'tag1', 10.5, '2024-01-01 12:00:00+00:00') - ON CONFLICT (model_id, timestamp, variable) DO NOTHING - """), - {'model_id': unique_id} - ) - conn.commit() - - # Prepare the same data to trigger conflict - test_data = [ - { - 'timestamp': '2024-01-01 12:00:00+0000', - 'name': 'tag1', - 'value': 10.5, - 'tag': 'webid1', - }, - ] - - input_data = { - 'metadata': { - 'metadata': { - 'model_id': str(unique_id), - 'model_name': 'Test Model', - 'schedule_name': 'test-schedule', - 'workflow_name': 'pi_web_api_scouter', - } - }, - 'workflow_name': 'pi_web_api_scouter', - 'schedule_name': 'test-schedule', - 'model_name': 'Test Model', - 'model_id': str(unique_id), - 'data': test_data, - 'trigger_laborious': False, - 'filters': {}, - 'schema': 'sientia_data', - 'table_name': 'laborious_data', - 'retention_time': 3600, - 'fill_missing_tags': False, - 'model_tags': { - 'tag1': { - 'webid': 'webid1', - 'aggr_function': 'avg', - 'data_range': [0, 100], - 'frequency': 60000, - }, - }, - } - - # Start workflow - handle = await client.start_workflow( - CoreScouter.run, - input_data, - id=f'test-core-scouter-conflict-{datetime.now().timestamp()}', - task_queue='test-queue', - ) - - # Wait for workflow completion - should complete without error - # (conflict is handled gracefully with ON CONFLICT DO NOTHING) - await handle.result() - - # Verify no exception was raised and workflow completed - # The duplicate should be ignored (0 affected rows), but workflow should complete - with postgres_engine.connect() as conn: - result = conn.execute( - text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = :model_id"), - {'model_id': unique_id} - ) - row_count = result.scalar() - - # Should still have 1 row (duplicate was ignored) - assert row_count == 1, f"Expected 1 row (duplicate ignored), got {row_count}" - diff --git a/e2e/test_pi_web_api_scouter.py b/e2e/test_pi_web_api_scouter.py deleted file mode 100644 index 71f002f..0000000 --- a/e2e/test_pi_web_api_scouter.py +++ /dev/null @@ -1,106 +0,0 @@ -""" -End-to-end tests for PI Web API Scouter workflow. -""" - -from datetime import datetime - -import pytest -from sqlalchemy import inspect, text -from temporalio.testing import WorkflowEnvironment -from temporalio.worker import Worker - -from scouter.activities.activities import Activities -from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter - - -@pytest.mark.asyncio -@pytest.mark.integration -async def test_pi_web_api_scouter_e2e( - temporal_test_env: WorkflowEnvironment, - temporal_worker: Worker, - test_activities: Activities, - mock_pi_web_api_client, - postgres_engine, -): - """ - End-to-end test for PI Web API Scouter workflow. - - This test: - 1. Starts the workflow with test data - 2. Verifies PI Web API is called - 3. Verifies data flows through CoreScouter - 4. Verifies data is stored in PostgreSQL (schema: sientia_data, table: laborious_data) - 5. Verifies data is cached in Redis - """ - client = temporal_test_env.client - - # Prepare test input - input_data = { - 'model_name': 'PI Web API Scouter Test Model', - 'model_id': '1', - 'schedule_name': 'pi-web-api-scouter-test', - 'model_tags': { - 'tag1': { - 'webid': 'webid1', - 'aggr_function': 'avg', - 'data_range': [0, 100], - 'frequency': 60000, - }, - 'tag2': { - 'webid': 'webid2', - 'aggr_function': 'avg', - 'data_range': [0, 100], - 'frequency': 60000, - }, - }, - 'trigger_laborious': False, - 'filters': {}, - 'schema': 'sientia_data', - 'table_name': 'laborious_data', - 'retention_time': 3600, - 'fill_missing_tags': False, - 'pi_web_api_query': { - 'endpoint': '/streamsets/recorded', - 'period': '*-1d', - 'max_count': 10, - 'api_timeout': 30, - }, - } - - # Start workflow - handle = await client.start_workflow( - PIWebAPIScouter.run, - input_data, - id=f'test-workflow-{datetime.now().timestamp()}', - task_queue='test-queue', - ) - - # Wait for workflow completion - await handle.result() - - # Verify PI Web API was called - mock_pi_web_api_client.get_latest_values_df.assert_called_once() - - # Verify data was stored in PostgreSQL - inspector = inspect(postgres_engine) - - # Schema and table are created by the setup_postgres_schema_and_table fixture - schema_name = 'sientia_data' - table_name = 'laborious_data' - full_table_name = f"{schema_name}.{table_name}" - - # Check if table exists in the schema - table_exists = inspector.has_table(table_name, schema=schema_name) - - assert table_exists, f"Expected table {full_table_name} to exist in PostgreSQL" - - # Verify data was inserted - with postgres_engine.connect() as conn: - result = conn.execute(text(f"SELECT COUNT(*) FROM {full_table_name}")) - row_count = result.scalar() - - assert row_count > 0, f"Expected data in PostgreSQL table {full_table_name}, got {row_count} rows" - - # Verify data was cached in Redis - keys = await test_activities.redis_repository.keys('*') - assert len(keys) > 0, "Expected data in Redis" diff --git a/e2e/test_pi_web_api_scouter_errors.py b/e2e/test_pi_web_api_scouter_errors.py index cb4e512..f1088be 100644 --- a/e2e/test_pi_web_api_scouter_errors.py +++ b/e2e/test_pi_web_api_scouter_errors.py @@ -11,7 +11,7 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from scouter.activities.activities import Activities -from scouter.utils.clients.pi_web_api_client import PIMSRequestError +from sientia_do.repository.pi_web_api_client import PIMSRequestError from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter diff --git a/pyproject.toml b/pyproject.toml index 25e2eab..f77f9f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -116,6 +116,7 @@ addopts = [ "--strict-markers", ] asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" markers = [ "asyncio: marks tests as async", "integration: marks tests as integration tests", diff --git a/tests/utils/clients/test_pi_web_api_client.py b/tests/utils/clients/test_pi_web_api_client.py deleted file mode 100644 index f4f8870..0000000 --- a/tests/utils/clients/test_pi_web_api_client.py +++ /dev/null @@ -1,588 +0,0 @@ -import json -from unittest.mock import AsyncMock, MagicMock, patch - -import pandas as pd -import pycurl -import pytest - -from scouter.utils.clients.pi_web_api_client import PIMSRequestError, PIWebAPIClient - - -@pytest.fixture -def mock_logger(): - return MagicMock() - - -@pytest.fixture -def mock_notification_handler(): - return AsyncMock() - - -@pytest.fixture -def mock_metrics_controller(): - return AsyncMock() - - -@pytest.fixture -def auth_config_basic(): - return {'type': 'basic', 'token': 'test_token_123'} - - -@pytest.fixture -def auth_config_bearer(): - return {'type': 'bearer', 'token': 'bearer_token_456'} - - -@pytest.fixture -def pi_client(mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic): - return PIWebAPIClient( - base_url='https://pi.example.com', - auth_config=auth_config_basic, - logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, - ) - - -def test_init_with_basic_auth( - mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic -): - """Test initialization with basic authentication""" - client = PIWebAPIClient( - base_url='https://pi.example.com/', - auth_config=auth_config_basic, - logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, - ) - - assert client.base_url == 'https://pi.example.com' - assert client.auth_config['type'] == 'basic' - assert client.headers['Authorization'] == 'Basic test_token_123' - assert client.headers['Content-Type'] == 'application/json' - assert client.headers['Accept'] == 'application/json' - mock_logger.info.assert_called_with('Authenticating with basic authentication') - - -def test_init_with_bearer_auth( - mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_bearer -): - """Test initialization with bearer authentication""" - client = PIWebAPIClient( - base_url='https://pi.example.com', - auth_config=auth_config_bearer, - logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, - ) - - assert client.base_url == 'https://pi.example.com' - assert client.auth_config['type'] == 'bearer' - assert client.headers['Authorization'] == 'Bearer bearer_token_456' - mock_logger.info.assert_called_with('Authenticating with bearer authentication') - - -def test_init_with_custom_headers( - mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic -): - """Test initialization with custom headers""" - custom_headers = { - 'Content-Type': 'application/xml', - 'Custom-Header': 'custom_value', - } - - client = PIWebAPIClient( - base_url='https://pi.example.com', - auth_config=auth_config_basic, - logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, - headers_config=custom_headers, - ) - - assert client.headers['Content-Type'] == 'application/xml' - assert client.headers['Custom-Header'] == 'custom_value' - assert client.headers['Authorization'] == 'Basic test_token_123' - - -def test_authenticate_invalid_type(mock_logger, mock_notification_handler, mock_metrics_controller): - """Test that invalid authentication type raises ValueError""" - invalid_auth_config = {'type': 'invalid', 'token': 'test_token'} - - with pytest.raises(ValueError) as exc_info: - PIWebAPIClient( - base_url='https://pi.example.com', - auth_config=invalid_auth_config, - logger=mock_logger, - notification_handler=mock_notification_handler, - metrics_controller=mock_metrics_controller, - ) - - assert 'Invalid authentication type: invalid' in str(exc_info.value) - - -@patch('scouter.utils.clients.pi_web_api_client.SientiaMonitoring.shutdown') -def test_close(mock_shutdown, pi_client): - """Test close method calls shutdown""" - pi_client.close() - - mock_shutdown.assert_called_once() - - -def test_to_clean_timestamp(pi_client): - """Test timestamp cleaning and normalization""" - timestamps = pd.Series( - [ - '2025-01-15T10:30:45.123456Z', - '2025-01-15T10:30:46.789012Z', - '2025-01-15T10:30:47.999999Z', - ] - ) - - result = pi_client._to_clean_timestamp(timestamps) - - assert isinstance(result, pd.Series) - assert result.dtype == 'datetime64[ns, UTC]' - # Verify microseconds are floored to seconds - assert result[0] == pd.Timestamp('2025-01-15T10:30:45Z') - assert result[1] == pd.Timestamp('2025-01-15T10:30:46Z') - assert result[2] == pd.Timestamp('2025-01-15T10:30:47Z') - - -def test_to_clean_timestamp_with_invalid_values(pi_client): - """Test timestamp cleaning with invalid values returns NaT""" - timestamps = pd.Series(['invalid', 'not_a_date', '2025-01-15T10:30:45Z']) - - result = pi_client._to_clean_timestamp(timestamps) - - assert pd.isna(result[0]) - assert pd.isna(result[1]) - assert result[2] == pd.Timestamp('2025-01-15T10:30:45Z') - - -def test_extract_numeric_with_float(pi_client): - """Test extracting numeric value from float""" - result = pi_client._extract_numeric(42.5) - - assert result == pytest.approx(42.5) - - -def test_extract_numeric_with_int(pi_client): - """Test extracting numeric value from int""" - result = pi_client._extract_numeric(42) - - assert result == pytest.approx(42.0) - - -def test_extract_numeric_with_string(pi_client): - """Test extracting numeric value from string""" - result = pi_client._extract_numeric('123.45') - - assert result == pytest.approx(123.45) - - -def test_extract_numeric_with_dict(pi_client): - """Test extracting numeric value from dictionary""" - result = pi_client._extract_numeric({'Value': 99.9}) - - assert result == pytest.approx(99.9) - - -def test_extract_numeric_with_invalid_value(pi_client): - """Test extracting numeric value from invalid value returns None/NaN""" - result = pi_client._extract_numeric('invalid_number') - - assert pd.isna(result) - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_success(mock_curl_class, pi_client): - """Test successful GET request with JSON response""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - response_data = {'status': 'success', 'data': [1, 2, 3]} - response_json = json.dumps(response_data).encode('utf-8') - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(response_json) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 200 - - result = await pi_client._curl_get_json('https://pi.example.com/api/test') - - assert result == response_data - mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 30) - mock_curl.perform.assert_called_once() - mock_curl.close.assert_called_once() - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_with_params(mock_curl_class, pi_client): - """Test GET request with query parameters""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - response_data = {'result': 'ok'} - response_json = json.dumps(response_data).encode('utf-8') - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(response_json) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 200 - - params = [('key1', 'value1'), ('key2', 'value2')] - result = await pi_client._curl_get_json('https://pi.example.com/api', params=params) - - assert result == response_data - # Verify URL includes query parameters - set_url_call = [call for call in mock_curl.setopt.call_args_list if call[0][0] == pycurl.URL][0] - assert b'key1=value1' in set_url_call[0][1] - assert b'key2=value2' in set_url_call[0][1] - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_http_error(mock_curl_class, pi_client): - """Test GET request with HTTP error response""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - error_response = b'{"error": "Not found"}' - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(error_response) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 404 - - with pytest.raises(PIMSRequestError) as exc_info: - await pi_client._curl_get_json('https://pi.example.com/api/notfound') - - assert 'HTTP 404' in str(exc_info.value) - mock_curl.close.assert_called_once() - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_connection_error(mock_curl_class, pi_client): - """Test GET request with connection error""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - mock_curl.perform.side_effect = pycurl.error('Connection failed') - - with pytest.raises(PIMSRequestError) as exc_info: - await pi_client._curl_get_json('https://pi.example.com/api/test') - - assert 'Connection error' in str(exc_info.value) - mock_curl.close.assert_called_once() - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_invalid_json(mock_curl_class, pi_client): - """Test GET request with invalid JSON response""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - invalid_json = b'This is not valid JSON' - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(invalid_json) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 200 - - with pytest.raises(PIMSRequestError) as exc_info: - await pi_client._curl_get_json('https://pi.example.com/api/test') - - assert 'Error decoding JSON response' in str(exc_info.value) - mock_curl.close.assert_called_once() - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_with_custom_timeout(mock_curl_class, pi_client): - """Test GET request with custom timeout""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - response_data = {'status': 'ok'} - response_json = json.dumps(response_data).encode('utf-8') - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(response_json) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 200 - - await pi_client._curl_get_json('https://pi.example.com/api/test', timeout=60) - - mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 60) - - -@pytest.mark.asyncio -@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') -async def test_curl_get_json_without_ssl_verify(mock_curl_class, pi_client): - """Test GET request with SSL verification disabled""" - mock_curl = MagicMock() - mock_curl_class.return_value = mock_curl - - response_data = {'status': 'ok'} - response_json = json.dumps(response_data).encode('utf-8') - - def mock_perform(): - buffer = mock_curl.setopt.call_args_list[1][0][1] - buffer.write(response_json) - - mock_curl.perform.side_effect = mock_perform - mock_curl.getinfo.return_value = 200 - - await pi_client._curl_get_json('https://pi.example.com/api/test', verify=False) - - mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYPEER, 0) - mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYHOST, 0) - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_success(mock_curl_get_json, pi_client): - """Test successful retrieval of latest values""" - mock_curl_get_json.return_value = { - 'Items': [ - { - 'Name': 'tag1', - 'Items': [ - {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, - {'Timestamp': '2025-01-15T10:31:00Z', 'Value': 43.0}, - ], - }, - { - 'Name': 'tag2', - 'Items': [ - {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 100.0}, - ], - }, - ] - } - - web_ids = { - 'tag1': {'webid': 'webid1'}, - 'tag2': {'webid': 'webid2'}, - } - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - start_time='*-1d', - end_time='*', - max_count=10, - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 3 - assert list(result.columns) == ['timestamp', 'name', 'value', 'tag'] - assert result['name'].tolist() == ['tag1', 'tag1', 'tag2'] - assert result['value'].tolist() == [42.5, 43.0, 100.0] - - mock_curl_get_json.assert_called_once() - call_args = mock_curl_get_json.call_args - assert call_args[1]['url'] == 'https://pi.example.com/streamsets/recorded' - assert call_args[1]['timeout'] == 30 - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_with_custom_params(mock_curl_get_json, pi_client): - """Test get_latest_values_df with custom parameters""" - mock_curl_get_json.return_value = { - 'Items': [ - { - 'Name': 'tag1', - 'Items': [ - {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, - ], - } - ] - } - - web_ids = {'tag1': {'webid': 'webid1'}} - metadata = {'model_id': 'test_model'} - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - start_time='*-7d', - end_time='*-1d', - max_count=100, - timeout=60, - metadata=metadata, - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 1 - - mock_curl_get_json.assert_called_once() - call_args = mock_curl_get_json.call_args - params = call_args[1]['params'] - - # Verify parameters (inverted: startTime uses end_time, endTime uses start_time) - assert ('startTime', '*-1d') in params - assert ('endtime', '*-7d') in params - assert ('maxCount', '100') in params - assert call_args[1]['timeout'] == 60 - assert call_args[1]['metadata'] == metadata - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_empty_response(mock_curl_get_json, pi_client): - """Test get_latest_values_df with empty response""" - mock_curl_get_json.return_value = {'Items': []} - - web_ids = {'tag1': {'webid': 'webid1'}} - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 0 - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_no_items_in_tag(mock_curl_get_json, pi_client): - """Test get_latest_values_df when tag has no items""" - mock_curl_get_json.return_value = { - 'Items': [ - { - 'Name': 'tag1', - 'Items': [], - } - ] - } - - web_ids = {'tag1': {'webid': 'webid1'}} - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 0 - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_with_missing_timestamp(mock_curl_get_json, pi_client): - """Test get_latest_values_df filters out items with missing timestamp""" - mock_curl_get_json.return_value = { - 'Items': [ - { - 'Name': 'tag1', - 'Items': [ - {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, - {'Value': 43.0}, # Missing Timestamp - {'Timestamp': None, 'Value': 44.0}, # None Timestamp - ], - } - ] - } - - web_ids = {'tag1': {'webid': 'webid1'}} - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 1 # Only the first item should be included - assert result['value'].tolist() == [42.5] - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_with_nested_value(mock_curl_get_json, pi_client): - """Test get_latest_values_df with nested value extraction""" - mock_curl_get_json.return_value = { - 'Items': [ - { - 'Name': 'tag1', - 'Items': [ - {'Timestamp': '2025-01-15T10:30:00Z', 'Value': {'Value': 42.5}}, - ], - } - ] - } - - web_ids = {'tag1': {'webid': 'webid1'}} - - result = await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - ) - - assert isinstance(result, pd.DataFrame) - assert len(result) == 1 - assert result['value'].tolist() == [42.5] - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_default_max_count(mock_curl_get_json, pi_client): - """Test get_latest_values_df uses default max_count of 1""" - mock_curl_get_json.return_value = {'Items': []} - - web_ids = {'tag1': {'webid': 'webid1'}} - - await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - ) - - call_args = mock_curl_get_json.call_args - params = call_args[1]['params'] - - assert ('maxCount', '1') in params - # Verify default time parameters are inverted (startTime uses end_time default, endTime uses start_time default) - assert ('startTime', '*') in params # Default end_time - assert ('endtime', '*-1d') in params # Default start_time - - -@pytest.mark.asyncio -@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) -async def test_get_latest_values_df_with_none_max_count(mock_curl_get_json, pi_client): - """Test get_latest_values_df does not send maxCount parameter when max_count is None""" - mock_curl_get_json.return_value = {'Items': []} - - web_ids = {'tag1': {'webid': 'webid1'}} - - await pi_client.get_latest_values_df( - web_ids=web_ids, - endpoint='/streamsets/recorded', - max_count=None, - ) - - call_args = mock_curl_get_json.call_args - params = call_args[1]['params'] - - # Verify maxCount parameter is not present when max_count is None - assert ('maxCount', '1') not in params - assert ('maxCount', None) not in params - # Verify time parameters are still present - assert ('startTime', '*') in params - assert ('endtime', '*-1d') in params diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index be7ab70..fe24347 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -4,12 +4,7 @@ from unittest.mock import patch import pytest from scouter.utils.connectors_config import ( - build_api_config, - build_druid_config, build_kafka_config, - build_mongodb_config, - build_postgres_config, - build_redis_config, ) @@ -19,50 +14,6 @@ def mock_env_vars(): 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""" @@ -89,114 +40,3 @@ def test_build_kafka_config_with_env_vars(): '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, 'username': None, 'password': None} - - -@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', - 'REDIS_USERNAME': 'test', - 'REDIS_PASSWORD': 'test', - }, - ): - config = build_redis_config() - - assert config == { - 'host': 'redis.example.com', - 'port': 6380, - 'username': 'test', - 'password': 'test', - } - - -def test_build_mongodb_config_defaults(): - """Test that build_mongodb_config returns default values when no env vars are set""" - os.environ['MONGODB_URL'] = 'localhost:27017' - os.environ['MONGODB_DATABASE_NAME'] = 'sientia' - os.environ['MONGODB_USERNAME'] = 'sientia' - os.environ['MONGODB_PASSWORD'] = 'sientia' - - config = build_mongodb_config() - - assert config == { - 'connection_string': 'mongodb://sientia:sientia@localhost:27017', # NOSONAR - 'database_name': 'sientia', - } - - -def test_build_mongodb_config_with_env_vars(): - """Test that build_mongodb_config uses env vars when set""" - with patch.dict( - os.environ, - { - 'MONGODB_URL': 'mongodb.example.com:27017', - 'MONGODB_DATABASE_NAME': 'test_db', - 'MONGODB_USERNAME': 'test', - 'MONGODB_PASSWORD': 'test', - }, - ): - config = build_mongodb_config() - - assert config == { - 'connection_string': 'mongodb://test:test@mongodb.example.com:27017', - 'database_name': 'test_db', - } - - -@pytest.mark.usefixtures('mock_env_vars') -def test_build_api_config_defaults(): - """Test that build_api_config returns default values when no env vars are set""" - config = build_api_config() - - assert config == { - 'base_url': 'https://pi.example.com', - 'auth_type': 'basic', - 'auth_token': None, - } - - -@pytest.mark.usefixtures('mock_env_vars') -def test_build_api_config_with_env_vars(): - """Test that build_api_config uses env vars when set""" - with patch.dict( - os.environ, - { - 'PI_WEB_API_BASE_URL': 'https://api.production.com', - 'PI_WEB_API_AUTH_TYPE': 'bearer', - 'PI_WEB_API_AUTH_TOKEN': 'secret_token_123', - }, - ): - config = build_api_config() - - assert config == { - 'base_url': 'https://api.production.com', - 'auth_type': 'bearer', - 'auth_token': 'secret_token_123', - } - - -def test_build_druid_config_defaults(): - """Test that build_druid_config returns default values when no env vars are set""" - config = build_druid_config() - - assert config == {'host': 'localhost', 'port': 8082} - - -def test_build_druid_config_with_env_vars(): - """Test that build_druid_config uses env vars when set""" - with patch.dict(os.environ, {'DRUID_HOST': 'druid.example.com', 'DRUID_PORT': '8083'}): - config = build_druid_config() - - assert config == {'host': 'druid.example.com', 'port': 8083}