""" Test cases for datasdk/sources/slowlog.py """ import os import sys import unittest import tempfile import shutil import gzip from unittest.mock import patch, MagicMock, mock_open from datetime import datetime # Add project root to path pa = os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')) sys.path.insert(0, pa) from datasdk.sources import slowlog from datasdk.config import config class TestSlowLog(unittest.TestCase): """Test SlowLog class""" def setUp(self): """Set up test fixtures""" self.temp_dir = tempfile.mkdtemp() self.cfg = config.Config(config.Env.dev, "test-client-id", "test-client-secret") def tearDown(self): """Clean up test fixtures""" if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_slowlog_initialization(self, mock_o11y_api_class): """Test SlowLog initialization""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api slow_log = slowlog.SlowLog(self.cfg) self.assertEqual(slow_log.cfg, self.cfg) self.assertEqual(slow_log.api, mock_o11y_api) mock_o11y_api_class.assert_called_once_with(self.cfg) @patch('datasdk.sources.slowlog.SlowLog.getSlowLogObjStorage') @patch('datasdk.sources.slowlog.SlowLog.getDedicatedSlowLog') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getSlowLog_serverless(self, mock_o11y_api_class, mock_get_dedicated, mock_get_obj_storage): """Test getSlowLog for serverless cluster""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_o11y_api.getO11yClusterInfo.return_value = mock_cluster mock_get_obj_storage.return_value = "serverless log data" slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getSlowLog("cluster1", True, 1000, 2000) self.assertEqual(result, "serverless log data") mock_o11y_api.getO11yClusterInfo.assert_called_once_with("cluster1", True) mock_get_obj_storage.assert_called_once_with(mock_cluster, 1000, 2000) mock_get_dedicated.assert_not_called() @patch('datasdk.sources.slowlog.SlowLog.getSlowLogObjStorage') @patch('datasdk.sources.slowlog.SlowLog.getDedicatedSlowLog') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getSlowLog_dedicated(self, mock_o11y_api_class, mock_get_dedicated, mock_get_obj_storage): """Test getSlowLog for dedicated cluster""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_o11y_api.getO11yClusterInfo.return_value = mock_cluster mock_get_dedicated.return_value = "dedicated log data" slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getSlowLog("cluster1", False, 1000, 2000) self.assertEqual(result, "dedicated log data") mock_o11y_api.getO11yClusterInfo.assert_called_once_with("cluster1", False) mock_get_dedicated.assert_called_once_with(mock_cluster, 1000, 2000) mock_get_obj_storage.assert_not_called() @patch('datasdk.sources.slowlog.get_files') @patch('datasdk.sources.slowlog.S3') @patch('datasdk.sources.slowlog.utils.getO11yBucketName') @patch('datasdk.sources.slowlog.utils.getO11yObjPrefix') @patch('datasdk.sources.slowlog.SlowLog.getCommomPrefix') @patch('datasdk.sources.slowlog.SlowLog.getTimestemp') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getDedicatedSlowLog_aws(self, mock_o11y_api_class, mock_get_timestemp, mock_get_common_prefix, mock_get_prefix, mock_get_bucket, mock_s3_class, mock_get_files): """Test getDedicatedSlowLog for AWS""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.vendor = "aws" mock_cluster.region = "us-west-2" mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_get_bucket.return_value = "test-bucket" mock_get_common_prefix.return_value = ["1676"] mock_get_prefix.return_value = "prefix/test" mock_s3.get_file_list.return_value = [ "prefix/test/1676466135-abc.log.gz", "prefix/test/1676466136-def.log.gz" ] mock_get_timestemp.side_effect = [1676466135, 1676466136] compressed_content = gzip.compress(b"log content 1") mock_get_files.return_value = [compressed_content] slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getDedicatedSlowLog(mock_cluster, 1676466135, 1676466136) self.assertIn("log content 1", result) mock_s3_class.assert_called_once_with("test-bucket") mock_get_files.assert_called_once() @patch('datasdk.sources.slowlog.gcs_get_files') @patch('datasdk.sources.slowlog.GCS') @patch('datasdk.sources.slowlog.utils.getO11yBucketName') @patch('datasdk.sources.slowlog.utils.getO11yObjPrefix') @patch('datasdk.sources.slowlog.SlowLog.getCommomPrefix') @patch('datasdk.sources.slowlog.SlowLog.getTimestemp') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getDedicatedSlowLog_gcp(self, mock_o11y_api_class, mock_get_timestemp, mock_get_common_prefix, mock_get_prefix, mock_get_bucket, mock_gcs_class, mock_gcs_get_files): """Test getDedicatedSlowLog for GCP""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.vendor = "gcp" mock_cluster.region = "us-central1" mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_gcs = MagicMock() mock_gcs_class.return_value = mock_gcs mock_get_bucket.return_value = "test-bucket" mock_get_common_prefix.return_value = ["1676"] mock_get_prefix.return_value = "prefix/test" mock_gcs.get_file_list.return_value = [ "prefix/test/1676466135-abc.log.gz" ] mock_get_timestemp.return_value = 1676466135 # GCS returns string content (not bytes like AWS) mock_gcs_get_files.return_value = ["gcp log content"] slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getDedicatedSlowLog(mock_cluster, 1676466135, 1676466136) self.assertIn("gcp log content", result) mock_gcs_class.assert_called_once_with("test-bucket", "gcp-autonomous-prod") mock_gcs_get_files.assert_called_once() @patch('datasdk.sources.slowlog.get_files') @patch('datasdk.sources.slowlog.S3') @patch('datasdk.sources.slowlog.utils.getO11yBucketName') @patch('datasdk.sources.slowlog.utils.getO11yServerlessObjPerfix') @patch('datasdk.sources.slowlog.SlowLog.getCommomPrefix') @patch('datasdk.sources.slowlog.SlowLog.getTimestemp') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getSlowLogObjStorage_aws(self, mock_o11y_api_class, mock_get_timestemp, mock_get_common_prefix, mock_get_prefix, mock_get_bucket, mock_s3_class, mock_get_files): """Test getSlowLogObjStorage for AWS""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.vendor = "aws" mock_cluster.region = "us-west-2" mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_get_bucket.return_value = "test-bucket" mock_get_common_prefix.return_value = ["1676"] mock_get_prefix.return_value = "serverless/prefix/test" mock_s3.get_file_list.return_value = [ "serverless/prefix/test/1676466135-abc.log.gz" ] mock_get_timestemp.return_value = 1676466135 compressed_content = gzip.compress(b"serverless log content") mock_get_files.return_value = [compressed_content] slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getSlowLogObjStorage(mock_cluster, 1676466135, 1676466136) self.assertIn("serverless log content", result) mock_s3_class.assert_called_once_with("test-bucket") mock_get_files.assert_called_once() def test_getCommomPrefix_simple(self): """Test getCommomPrefix with simple case""" slow_log = slowlog.SlowLog(self.cfg) # Same first digit result = slow_log.getCommomPrefix(1676466135, 1676466136) self.assertEqual(result, ["167646613"]) # Different first digit result = slow_log.getCommomPrefix(1000000000, 2000000000) self.assertEqual(result, [""]) def test_getCommomPrefix_recursive(self): """Test getCommomPrefix with recursive case (prefix length < 2)""" slow_log = slowlog.SlowLog(self.cfg) # This will trigger recursion because common prefix length is 1 result = slow_log.getCommomPrefix(1000000000, 1999999999) # Should return a list with two elements (split ranges) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) def test_getTimestemp(self): """Test getTimestemp extraction from object name""" slow_log = slowlog.SlowLog(self.cfg) # Test with standard format obj_name = "0/prod-devtier/serverless/1372813089187041280/1372813089206301327/1379661944642684098/logs/slowlog/1676466135-a054c164-5d13-4604-a3cb-ad612a102d33.log.gz" result = slow_log.getTimestemp(obj_name) self.assertEqual(result, 1676466135) # Test with different format obj_name = "prefix/1676466136-abc.log.gz" result = slow_log.getTimestemp(obj_name) self.assertEqual(result, 1676466136) def test_parseContent(self): """Test parseContent parsing slow log content""" slow_log = slowlog.SlowLog(self.cfg) content = """# Time: 2023-01-01T12:00:00.123456789Z SELECT * FROM test; # Time: 2023-01-01T12:01:00.987654321Z UPDATE test SET value = 1; """ result = slow_log.parseContent(content) self.assertEqual(len(result), 2) self.assertIn("timestamp", result[0]) self.assertIn("value", result[0]) self.assertIn("SELECT * FROM test;", result[0]["value"]) self.assertIn("UPDATE test SET value = 1;", result[1]["value"]) def test_parseContent_empty(self): """Test parseContent with empty content""" slow_log = slowlog.SlowLog(self.cfg) # Empty string splits to [''], which is one empty line # The code will try to append to current_record['value'] which doesn't exist # So we expect KeyError with self.assertRaises(KeyError): slow_log.parseContent("") def test_parseContent_no_time_header(self): """Test parseContent with content without time header""" slow_log = slowlog.SlowLog(self.cfg) # Content without time header will cause KeyError when trying to append to value # This is a bug in the code, but we test the current behavior content = "SELECT * FROM test;\nUPDATE test SET value = 1;" # The code will fail because current_record['value'] doesn't exist # So we expect an exception or empty result try: result = slow_log.parseContent(content) # If it doesn't fail, check the result if len(result) > 0: self.assertIn("value", result[0]) except (KeyError, AttributeError): # Expected behavior due to bug in code pass @patch('datasdk.sources.slowlog.time.sleep') @patch('datasdk.sources.slowlog.utils.getO11yLokiPath') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getSlowLogLoki(self, mock_o11y_api_class, mock_get_loki_path, mock_sleep): """Test getSlowLogLoki""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_get_loki_path.return_value = "http://loki.test.com" # Mock Loki response mock_loki_response = [ { "stream": {"instance": "instance1"}, "values": [ ["1676466135000000000", "log line 1"], ["1676466140000000000", "log line 2"] ] } ] mock_o11y_api.getO11yLoki.return_value = mock_loki_response mock_cluster = MagicMock() mock_cluster.vendor = "aws" mock_cluster.region = "us-west-2" mock_cluster.applicationID = "app1" slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getSlowLogLoki(mock_cluster, 1676466135, 1676466145) self.assertIn("instance1", result) self.assertEqual(len(result["instance1"]), 2) self.assertEqual(result["instance1"][0]["timestemp"], 1676466135) self.assertEqual(result["instance1"][0]["value"], "log line 1") mock_get_loki_path.assert_called_once_with("aws", "us-west-2") mock_o11y_api.getO11yLoki.assert_called() def test_writeToCSV(self): """Test writeToCSV""" slow_log = slowlog.SlowLog(self.cfg) log_data = { "instance1": [ {"timestemp": 1676466135, "value": "log line 1"}, {"timestemp": 1676466140, "value": "log line 2"} ], "instance2": [ {"timestemp": 1676466135, "value": "log line 3"} ] } csv_path = os.path.join(self.temp_dir, "test.csv") slow_log.writeToCSV(log_data, csv_path) self.assertTrue(os.path.exists(csv_path)) # Verify CSV content with open(csv_path, 'r') as f: lines = f.readlines() self.assertEqual(len(lines), 4) # 1 header + 3 data rows self.assertIn("instance", lines[0]) self.assertIn("timestemp", lines[0]) self.assertIn("value", lines[0]) def test_writeToLog(self): """Test writeToLog""" slow_log = slowlog.SlowLog(self.cfg) log_data = "This is test log data\nLine 2\nLine 3" log_path = os.path.join(self.temp_dir, "test.log") slow_log.writeToLog(log_data, log_path) self.assertTrue(os.path.exists(log_path)) # Verify log content with open(log_path, 'r') as f: content = f.read() self.assertEqual(content, log_data) def test_writeToLog_empty(self): """Test writeToLog with empty data""" slow_log = slowlog.SlowLog(self.cfg) log_path = os.path.join(self.temp_dir, "test_empty.log") slow_log.writeToLog("", log_path) self.assertTrue(os.path.exists(log_path)) with open(log_path, 'r') as f: content = f.read() self.assertEqual(content, "") @patch('datasdk.sources.slowlog.get_files') @patch('datasdk.sources.slowlog.S3') @patch('datasdk.sources.slowlog.utils.getO11yBucketName') @patch('datasdk.sources.slowlog.utils.getO11yObjPrefix') @patch('datasdk.sources.slowlog.SlowLog.getCommomPrefix') @patch('datasdk.sources.slowlog.SlowLog.getTimestemp') @patch('datasdk.sources.slowlog.o11y.O11yAPI') def test_getDedicatedSlowLog_no_matching_files(self, mock_o11y_api_class, mock_get_timestemp, mock_get_common_prefix, mock_get_prefix, mock_get_bucket, mock_s3_class, mock_get_files): """Test getDedicatedSlowLog when no files match time range""" mock_o11y_api = MagicMock() mock_o11y_api_class.return_value = mock_o11y_api mock_cluster = MagicMock() mock_cluster.vendor = "aws" mock_cluster.region = "us-west-2" mock_cluster.tenantID = "tenant1" mock_cluster.projectID = "project1" mock_cluster.applicationID = "app1" mock_s3 = MagicMock() mock_s3_class.return_value = mock_s3 mock_get_bucket.return_value = "test-bucket" mock_get_common_prefix.return_value = ["1676"] mock_get_prefix.return_value = "prefix/test" mock_s3.get_file_list.return_value = [ "prefix/test/1676466130-abc.log.gz", # Before range "prefix/test/1676466140-def.log.gz" # After range ] mock_get_timestemp.side_effect = [1676466130, 1676466140] # Even if need_download is empty, get_files will still be called # because the code doesn't check if need_download is empty before calling mock_get_files.return_value = [] slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getDedicatedSlowLog(mock_cluster, 1676466135, 1676466136) # Should return empty string (no newline if no files downloaded) self.assertEqual(result, "") # get_files will be called even with empty list # Check that need_download list is empty self.assertTrue(mock_get_files.called) call_args = mock_get_files.call_args self.assertEqual(len(call_args[0][2]), 0) # need_download is 3rd positional arg def test_getCommomPrefix_exact_match(self): """Test getCommomPrefix with exact same timestamps""" slow_log = slowlog.SlowLog(self.cfg) result = slow_log.getCommomPrefix(1676466135, 1676466135) self.assertEqual(result, ["1676466135"]) def test_getCommomPrefix_different_lengths(self): """Test getCommomPrefix with different length timestamps""" slow_log = slowlog.SlowLog(self.cfg) # Different lengths should still work result = slow_log.getCommomPrefix(1000000000, 2000000000) self.assertEqual(result, [""]) def test_getCommomPrefix_zero_length(self): """Test getCommomPrefix with zero common prefix""" slow_log = slowlog.SlowLog(self.cfg) # Completely different timestamps result = slow_log.getCommomPrefix(1234567890, 9876543210) self.assertEqual(result, [""]) def test_getTimestemp_edge_cases(self): """Test getTimestemp with edge cases""" slow_log = slowlog.SlowLog(self.cfg) # Test with no dashes - split('-')[0] returns entire filename # This will cause ValueError when trying to convert to int obj_name = "prefix/1676466135.log.gz" with self.assertRaises(ValueError): slow_log.getTimestemp(obj_name) # Test with multiple dashes - should work, takes first part obj_name = "prefix/1676466135-abc-def-ghi.log.gz" result = slow_log.getTimestemp(obj_name) self.assertEqual(result, 1676466135) def test_parseContent_multiple_records(self): """Test parseContent with multiple records""" slow_log = slowlog.SlowLog(self.cfg) content = """# Time: 2023-01-01T12:00:00.123456789Z SELECT * FROM test1; # Time: 2023-01-01T12:01:00.987654321Z UPDATE test2 SET value = 1; # Time: 2023-01-01T12:02:00.111222333Z DELETE FROM test3; """ result = slow_log.parseContent(content) self.assertEqual(len(result), 3) self.assertIn("SELECT * FROM test1;", result[0]["value"]) self.assertIn("UPDATE test2 SET value = 1;", result[1]["value"]) self.assertIn("DELETE FROM test3;", result[2]["value"]) def test_parseContent_single_line(self): """Test parseContent with single line""" slow_log = slowlog.SlowLog(self.cfg) content = "# Time: 2023-01-01T12:00:00.123456789Z\nSELECT * FROM test;" result = slow_log.parseContent(content) self.assertEqual(len(result), 1) self.assertIn("timestamp", result[0]) self.assertIn("SELECT * FROM test;", result[0]["value"]) def test_writeToCSV_empty_data(self): """Test writeToCSV with empty data""" slow_log = slowlog.SlowLog(self.cfg) csv_path = os.path.join(self.temp_dir, "test_empty.csv") slow_log.writeToCSV({}, csv_path) self.assertTrue(os.path.exists(csv_path)) with open(csv_path, 'r') as f: lines = f.readlines() self.assertEqual(len(lines), 1) # Only header self.assertIn("instance", lines[0]) if __name__ == '__main__': unittest.main()