"""Unit tests for Datadog connector.""" from unittest.mock import patch import pytest from grass.connectors import datadog @patch('grass.connectors.datadog.api') @patch('grass.connectors.datadog.initialize') def test_datadog_connection(mock_datadog_initialize, mock_datadog_api): """Test publish_metric function.""" with datadog.datadog_connection(app_key='1', api_key='2') as conn: mock_datadog_initialize.assert_called_with(app_key='1', api_key='2') conn.Metric.send(metric='page.views', points=100) mock_datadog_api.Metric.send.assert_called_once_with( metric='page.views', points=100 ) def test_key(): """Test key function to generate custom metric name.""" assert datadog.key('my_metric') == 'ows.grass.my_metric' assert datadog.key('foo.bar') == 'ows.grass.foo.bar' @patch('grass.connectors.datadog.api') @patch('grass.connectors.datadog.initialize') @patch('grass.config.SEND_METRICS_TO_DD_ENABLED', False) def test_skip_publish_metric(mock_datadog_initialize, mock_datadog_api): """Test skip publish_metric function.""" datadog.publish_metric(metric='foo', count=1) mock_datadog_api.Metric.send.assert_not_called() @patch('grass.connectors.datadog.api') @patch('grass.connectors.datadog.initialize') @patch('grass.config.SEND_METRICS_TO_DD_ENABLED', True) def test_publish_metric(mock_datadog_initialize, mock_datadog_api): """Test publish_metric function.""" with patch('grass.connectors.datadog.time.time', return_value=12345): datadog.publish_metric(metric='ows.grass.foo', count=1) mock_datadog_api.Metric.send.assert_called_once_with( metric='ows.grass.foo', points=[(12345, 1)], type='count', tags=[ 'environment:test', 'service_name:ows-grass', ], ) @patch('grass.connectors.datadog.api') @patch('grass.connectors.datadog.initialize') @patch('grass.config.SEND_METRICS_TO_DD_ENABLED', True) def test_send_timer(mock_initialize, mock_api): """Test send_timer sends a gauge metric correctly.""" delta_ms = 123 metric_name = 'ows.grass.requests' with patch('grass.connectors.datadog.time.time', return_value=12345): datadog.send_timer(metric_name, delta_ms) mock_api.Metric.send.assert_called_once_with( metric='ows.grass.requests', points=[(12345, delta_ms)], type='gauge', tags=[ 'environment:test', 'service_name:ows-grass', 'unit:ms', ], )