"""Unit tests for o11ydiagnosis.subtasks.slowlogconverter package __init__ helpers and Config.""" import os import unittest from unittest.mock import MagicMock, patch from o11ydiagnosis.subtasks.slowlogconverter import ( Config, drhouse_s3_bucket, drhouse_s3_prefix, o11y_s3_bucket, o11y_s3_prefix, run_backfill_from_payload, ) def _task_info( provider: str, *, env: str = "prod", region: str = "us-west-2", cluster_type: str = "dedicated", cluster: str = "cluster-1", tenant: str = "tenant-1", project: str = "project-1", o11y_cluster: str = "o11y-1", ): return { "meta": { "env": env, "provider": provider, "region": region, "tenantID": tenant, "projectID": project, "clusterID": cluster, "o11yClusterID": o11y_cluster, "clusterType": cluster_type, } } class TestBucketAndPrefixHelpers(unittest.TestCase): def test_o11y_s3_bucket_respects_bucket_name_env(self): with patch.dict(os.environ, {"BUCKET_NAME": "env-bucket"}, clear=False): self.assertEqual(o11y_s3_bucket("prod", "us-west-2", False, False), "env-bucket") def test_o11y_s3_bucket_branches_without_bucket_name(self): with patch.dict(os.environ, {}, clear=False): os.environ.pop("BUCKET_NAME", None) self.assertEqual(o11y_s3_bucket("prod", "r1", True, False), "o11y-prod-shared-r1") self.assertEqual(o11y_s3_bucket("prod", "r2", False, True), "o11y-prod-shared-r2") self.assertEqual(o11y_s3_bucket("staging", "r3", False, False), "o11y-prod-shared-r3") self.assertEqual(o11y_s3_bucket("dev", "r4", False, False), "o11y-dev-shared-r4") self.assertEqual(o11y_s3_bucket("prod", "r5", False, False), "o11y-prod-shared-r5") self.assertEqual(o11y_s3_bucket("other", "r6", False, False), "o11y-test-shared-r6") def test_o11y_s3_prefix(self): self.assertEqual( o11y_s3_prefix("t", "p", "c", True, False), "0/prod-devtier/serverless/t/p/c", ) self.assertEqual(o11y_s3_prefix("t", "p", "c", False, True), "0/next-gen/c") self.assertEqual(o11y_s3_prefix("t", "p", "c", False, False), "0/t/p/c") def test_drhouse_s3_bucket(self): with patch.dict(os.environ, {"BUCKET_NAME": "drhouse-bucket"}, clear=False): self.assertEqual(drhouse_s3_bucket("us-west-2", "staging"), "drhouse-bucket") with patch.dict(os.environ, {}, clear=False): os.environ.pop("BUCKET_NAME", None) self.assertEqual(drhouse_s3_bucket("us-west-2", "staging"), "o11y-prod-shared-us-west-2") self.assertEqual(drhouse_s3_bucket("eu-1", "dev"), "o11y-dev-shared-eu-1") self.assertEqual(drhouse_s3_bucket("ap-1", "prod"), "o11y-prod-shared-ap-1") self.assertEqual(drhouse_s3_bucket("xx", "sandbox"), "o11y-test-shared-xx") def test_drhouse_s3_prefix(self): self.assertEqual(drhouse_s3_prefix("cid"), "diagnosis/data/cid") class TestConfig(unittest.TestCase): def _make_engine_task(self, local_path: str): t = MagicMock() t.cache_prefix = local_path return t @patch("o11ydiagnosis.subtasks.slowlogconverter.metrics") @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob") @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS") @patch("o11ydiagnosis.subtasks.slowlogconverter.S3") def test_config_providers( self, mock_s3_cls, mock_gcs_cls, mock_az_cls, mock_task_cls, mock_metrics ): engine = self._make_engine_task("/tmp/slowlog-cache") subtask = {} mock_task_inst = MagicMock() mock_task_cls.return_value = mock_task_inst for provider, expected in ( ("aws", mock_s3_cls), ("alicloud", mock_s3_cls), ("gcp", mock_gcs_cls), ("azure", mock_az_cls), ): mock_s3_cls.reset_mock() mock_gcs_cls.reset_mock() mock_az_cls.reset_mock() mock_task_cls.reset_mock() cfg = Config(engine, _task_info(provider), subtask) self.assertEqual(cfg.cluster, "cluster-1") self.assertIs(cfg.task, mock_task_inst) expected.assert_called() with self.assertRaises(ValueError): Config(engine, _task_info("unknown"), subtask) @patch("o11ydiagnosis.subtasks.slowlogconverter.metrics") @patch("o11ydiagnosis.subtasks.slowlogconverter.Task", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.S3", return_value=MagicMock()) def test_config_is_timing_ready_close(self, _mock_s3, _mock_gcs, _mock_az, _mock_task, _mock_metrics): engine = self._make_engine_task("/tmp/x") cfg = Config(engine, _task_info("aws"), {}) self.assertTrue(cfg.is_timing(0)) self.assertFalse(cfg.is_timing(1)) self.assertTrue(cfg.ready(0)) cfg.close(0) @patch("o11ydiagnosis.subtasks.slowlogconverter.time.time", return_value=10_000_000) @patch("o11ydiagnosis.subtasks.slowlogconverter.metrics") @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.S3", return_value=MagicMock()) def test_config_start_supplement_mode(self, _mock_s3, _mock_gcs, _mock_az, mock_task_cls, _mock_metrics, _mock_time): engine = self._make_engine_task("/tmp/x") mock_inner = MagicMock() mock_task_cls.return_value = mock_inner cfg = Config(engine, _task_info("aws"), {}) # Aligned hour and strictly in supplement window (ts + 3600 < now) ts = 9_993_600 # 2776 * 3600 self.assertLess(ts + 3600, 10_000_000) self.assertTrue(ts % 3600 == 0 or ts % 3600 == 3570) cfg.start(ts) mock_inner.run.assert_called_once_with(ts) @patch("o11ydiagnosis.subtasks.slowlogconverter.time.time", return_value=10_000_000) @patch("o11ydiagnosis.subtasks.slowlogconverter.metrics") @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.S3", return_value=MagicMock()) def test_config_start_realtime_mode(self, _mock_s3, _mock_gcs, _mock_az, mock_task_cls, _mock_metrics, _mock_time): engine = self._make_engine_task("/tmp/x") mock_inner = MagicMock() mock_task_cls.return_value = mock_inner cfg = Config(engine, _task_info("aws"), {}) ts = 9_999_500 self.assertGreaterEqual(ts + 3600, 10_000_000) cfg.start(ts) self.assertGreaterEqual(mock_inner.run.call_count, 1) @patch("o11ydiagnosis.subtasks.slowlogconverter.time.time", return_value=10_000_000) @patch("o11ydiagnosis.subtasks.slowlogconverter.metrics") @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.S3", return_value=MagicMock()) def test_config_start_realtime_triggers_previous_hour( self, _mock_s3, _mock_gcs, _mock_az, mock_task_cls, _mock_metrics, _mock_time ): """When ts % 3600 <= 300, also compacts the previous hour window.""" engine = self._make_engine_task("/tmp/x") mock_inner = MagicMock() mock_task_cls.return_value = mock_inner cfg = Config(engine, _task_info("aws"), {}) ts = 9_997_400 self.assertGreaterEqual(ts + 3600, 10_000_000) self.assertLessEqual(ts % 3600, 300) cfg.start(ts) self.assertEqual(mock_inner.run.call_count, 2) class TestRunBackfillFromPayload(unittest.TestCase): def test_missing_required_fields(self): with self.assertRaises(ValueError) as ctx: run_backfill_from_payload({"provider": "aws"}) self.assertIn("missing required fields", str(ctx.exception)) def test_start_ts_not_before_end_ts(self): base = { "provider": "aws", "region": "us-west-2", "tenant": "t", "project": "p", "cluster": "c", "o11y_cluster": "o", "env": "prod", "cluster_type": "dedicated", "start_ts": 200, "end_ts": 100, } with self.assertRaises(ValueError) as ctx: run_backfill_from_payload(base) self.assertIn("start_ts must be less than end_ts", str(ctx.exception)) @patch("o11ydiagnosis.subtasks.slowlogconverter.S3", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") def test_success_calls_run_range(self, mock_task_cls, _mock_s3): inst = MagicMock() mock_task_cls.return_value = inst payload = { "provider": "aws", "region": "us-west-2", "tenant": "t", "project": "p", "cluster": "c", "o11y_cluster": "o", "env": "prod", "cluster_type": "serverless", "start_ts": 1000, "end_ts": 7200, "o11y_bucket": "custom-o11y", "o11y_prefix": "custom/prefix", "drhouse_bucket": "custom-dh", "drhouse_prefix": "custom/dh/p", "drhouse_local_path": "/tmp/backfill-test", } run_backfill_from_payload(payload) inst.run_range.assert_called_once_with(1000, 7200) call_kw = mock_task_cls.call_args.kwargs self.assertEqual(call_kw["o11y_bucket"], "custom-o11y") self.assertEqual(call_kw["o11y_prefix"], "custom/prefix") self.assertEqual(call_kw["drhouse_bucket"], "custom-dh") self.assertEqual(call_kw["drhouse_prefix"], "custom/dh/p") self.assertEqual(call_kw["drhouse_local_path"], "/tmp/backfill-test") @patch("o11ydiagnosis.subtasks.slowlogconverter.AzureBlob", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.GCS", return_value=MagicMock()) @patch("o11ydiagnosis.subtasks.slowlogconverter.Task") def test_build_objstore_gcp_and_azure(self, mock_task_cls, _mock_gcs, _mock_az): inst = MagicMock() mock_task_cls.return_value = inst base = { "region": "r", "tenant": "t", "project": "p", "cluster": "c", "o11y_cluster": "o", "env": "prod", "cluster_type": "dedicated", "start_ts": 0, "end_ts": 3600, } for prov in ("gcp", "azure"): mock_task_cls.reset_mock() run_backfill_from_payload({**base, "provider": prov}) inst.run_range.assert_called_once() with self.assertRaises(ValueError): run_backfill_from_payload({**base, "provider": "invalid"}) if __name__ == "__main__": unittest.main()