""" Test cases for o11ydiagnosis/subtasks/logreporter/__init__.py """ import os import sys import tempfile import shutil import unittest from unittest.mock import patch, MagicMock, Mock # Add project root to path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../'))) # Import the module to patch it import o11ydiagnosis.subtasks.logreporter as logreporter_module from o11ydiagnosis.subtasks.logreporter import Config, drhouse_s3_bucket, drhouse_s3_prefix from o11ydiagnosis.subtasks.slowlogconverter.objstore import S3, GCS, AzureBlob class TestConfig(unittest.TestCase): """Test Config class""" def setUp(self): """Set up test fixtures""" self.temp_dir = tempfile.mkdtemp() self.cache_prefix = os.path.join(self.temp_dir, 'cache') os.makedirs(self.cache_prefix, exist_ok=True) self.mock_task = MagicMock() self.mock_task.cache_prefix = self.cache_prefix self.task_info_aws = { "meta": { "provider": "aws", "region": "us-east-1", "clusterID": "test-cluster-aws" } } self.task_info_gcp = { "meta": { "provider": "gcp", "region": "us-central1", "clusterID": "test-cluster-gcp" } } self.task_info_azure = { "meta": { "provider": "azure", "region": "eastus", "clusterID": "test-cluster-azure" } } def tearDown(self): """Clean up test fixtures""" if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_init_aws(self, mock_s3_class, mock_task_class): """Test Config initialization with AWS provider""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_aws) self.assertEqual(config.cluster, "test-cluster-aws") self.assertEqual(config.interval_secs, 30) self.assertEqual(config.local_path, self.cache_prefix) mock_s3_class.assert_called_once() mock_task_class.assert_called_once() @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'GCS') def test_config_init_gcp(self, mock_gcs_class, mock_task_class): """Test Config initialization with GCP provider""" mock_gcs = MagicMock() mock_gcs_class.return_value = mock_gcs mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_gcp) self.assertEqual(config.cluster, "test-cluster-gcp") self.assertEqual(config.interval_secs, 30) mock_gcs_class.assert_called_once() mock_task_class.assert_called_once() @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'AzureBlob') def test_config_init_azure(self, mock_azure_class, mock_task_class): """Test Config initialization with Azure provider""" mock_azure = MagicMock() mock_azure_class.return_value = mock_azure mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_azure) self.assertEqual(config.cluster, "test-cluster-azure") self.assertEqual(config.interval_secs, 30) mock_azure_class.assert_called_once() mock_task_class.assert_called_once() def test_config_init_invalid_provider(self): """Test Config initialization with invalid provider""" task_info_invalid = { "meta": { "provider": "invalid", "region": "us-east-1", "clusterID": "test-cluster" } } with self.assertRaises(ValueError) as context: Config(self.mock_task, task_info_invalid) self.assertIn("invalid provider", str(context.exception)) @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_is_timing(self, mock_s3_class, mock_task_class): """Test is_timing method""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_aws) # Test timing at UTC 01:00 (3600 seconds into the day) # 1609459200 is 2021-01-01 00:00:00 UTC, so 01:00:00 is 1609459200 + 3600 ts_at_0100 = 1609459200 + 3600 # 2021-01-01 01:00:00 UTC self.assertTrue(config.is_timing(ts_at_0100)) # Test timing at other times ts_at_0200 = ts_at_0100 + 3600 # 02:00:00 self.assertFalse(config.is_timing(ts_at_0200)) ts_at_0000 = 1609459200 # 00:00:00 self.assertFalse(config.is_timing(ts_at_0000)) @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_ready(self, mock_s3_class, mock_task_class): """Test ready method""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_aws) # ready should always return True self.assertTrue(config.ready(1234567890)) self.assertTrue(config.ready(0)) self.assertTrue(config.ready(9999999999)) @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_start(self, mock_s3_class, mock_task_class): """Test start method""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_aws) # Test start with a timestamp ts = 1609459200 # 2021-01-01 00:00:00 UTC config.start(ts) # Should call prepare and run with adjusted timestamp # config.task is the mock Task instance created in __init__ mock_task_instance.prepare.assert_called_once() mock_task_instance.run.assert_called_once() # Check that timestamp was adjusted (should be previous day at 00:00:00) prepare_call_args = mock_task_instance.prepare.call_args run_call_args = mock_task_instance.run.call_args # Both should be called with 7 and adjusted timestamp self.assertEqual(prepare_call_args[0][0], 7) self.assertEqual(run_call_args[0][0], 7) # Timestamp should be adjusted to previous day expected_ts = ts - (24 * 3600) self.assertEqual(prepare_call_args[0][1], expected_ts) self.assertEqual(run_call_args[0][1], expected_ts) @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_close(self, mock_s3_class, mock_task_class): """Test close method""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance config = Config(self.mock_task, self.task_info_aws) ts = 1609459200 config.close(ts) # config.task is the mock Task instance created in __init__ mock_task_instance.upload_modified_objects.assert_called_once() @patch.object(logreporter_module, 'Task') @patch.object(logreporter_module, 'S3') def test_config_local_path_with_trailing_slash(self, mock_s3_class, mock_task_class): """Test Config handles local_path with trailing slash""" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_task_instance = MagicMock() mock_task_class.return_value = mock_task_instance cache_prefix_with_slash = self.cache_prefix + "/" self.mock_task.cache_prefix = cache_prefix_with_slash config = Config(self.mock_task, self.task_info_aws) # config.local_path is directly assigned from task.cache_prefix (no rstrip) # So it should equal cache_prefix_with_slash self.assertEqual(config.local_path, cache_prefix_with_slash) # However, the drhouse_local_path passed to Task should have trailing slash stripped # Check that Task was called with rstrip'd path call_args = mock_task_class.call_args drhouse_local_path_arg = call_args[0][4] # 5th positional argument self.assertEqual(drhouse_local_path_arg, self.cache_prefix) class TestHelperFunctions(unittest.TestCase): """Test helper functions""" def test_drhouse_s3_bucket(self): """Test drhouse_s3_bucket function""" self.assertEqual(drhouse_s3_bucket("us-east-1"), "o11y-prod-shared-us-east-1") self.assertEqual(drhouse_s3_bucket("us-west-2"), "o11y-prod-shared-us-west-2") self.assertEqual(drhouse_s3_bucket("eu-west-1"), "o11y-prod-shared-eu-west-1") def test_drhouse_s3_prefix(self): """Test drhouse_s3_prefix function""" self.assertEqual(drhouse_s3_prefix("cluster-1"), "diagnosis/data/cluster-1") self.assertEqual(drhouse_s3_prefix("test-cluster"), "diagnosis/data/test-cluster") self.assertEqual(drhouse_s3_prefix("my-cluster-id"), "diagnosis/data/my-cluster-id") if __name__ == '__main__': unittest.main()