import os import sys import unittest from datetime import datetime, timezone # Add project root to path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) from o11ydiagnosis.subtasks.slowlogconverter.slowlog import SlowLog from o11ydiagnosis.subtasks.slowlogconverter.slowstats import ( SlowStats, SlowStatsMerger, merge_stats, ) class TestSlowStats(unittest.TestCase): def setUp(self): """Set up test fixtures""" self.stats = SlowStats("test_digest_123") def test_initialization(self): """Test SlowStats initialization""" self.assertEqual(self.stats.digest, "test_digest_123") self.assertEqual(self.stats.count, 0) self.assertEqual(self.stats.query_time_sum, 0) self.assertEqual(self.stats.query_time_max, 0) self.assertEqual(self.stats.mem_max_sum, 0) self.assertEqual(self.stats.mem_max_max, 0) self.assertEqual(self.stats.disk_max_sum, 0) self.assertEqual(self.stats.disk_max_max, 0) def test_merge_with_query_time(self): """Test merging a SlowLog with query_time""" log = SlowLog() log.query_time = 10.5 log.mem_max = 100 log.disk_max = 200 self.stats.merge(log) self.assertEqual(self.stats.count, 1) self.assertEqual(self.stats.query_time_sum, 10.5) self.assertEqual(self.stats.query_time_max, 10.5) self.assertEqual(self.stats.mem_max_sum, 100) self.assertEqual(self.stats.mem_max_max, 100) self.assertEqual(self.stats.disk_max_sum, 200) self.assertEqual(self.stats.disk_max_max, 200) def test_merge_multiple_logs(self): """Test merging multiple SlowLogs""" log1 = SlowLog() log1.query_time = 10.0 log1.mem_max = 100 log1.disk_max = 200 log2 = SlowLog() log2.query_time = 20.0 log2.mem_max = 150 log2.disk_max = 250 log3 = SlowLog() log3.query_time = 5.0 log3.mem_max = 50 log3.disk_max = 150 self.stats.merge(log1) self.stats.merge(log2) self.stats.merge(log3) self.assertEqual(self.stats.count, 3) self.assertEqual(self.stats.query_time_sum, 35.0) self.assertEqual(self.stats.query_time_max, 20.0) # Max value self.assertEqual(self.stats.mem_max_sum, 300) self.assertEqual(self.stats.mem_max_max, 150) # Max value self.assertEqual(self.stats.disk_max_sum, 600) self.assertEqual(self.stats.disk_max_max, 250) # Max value def test_merge_with_none_values(self): """Test merging SlowLog with None values""" log = SlowLog() log.query_time = None log.mem_max = None log.disk_max = None self.stats.merge(log) self.assertEqual(self.stats.count, 1) self.assertEqual(self.stats.query_time_sum, 0) self.assertEqual(self.stats.query_time_max, 0) self.assertEqual(self.stats.mem_max_sum, 0) self.assertEqual(self.stats.mem_max_max, 0) self.assertEqual(self.stats.disk_max_sum, 0) self.assertEqual(self.stats.disk_max_max, 0) class TestSlowStatsMerger(unittest.TestCase): def setUp(self): """Set up test fixtures""" self.begin = 1625097600 # 2021-07-01 00:00:00 UTC self.end = 1625101200 # 2021-07-01 01:00:00 UTC self.merger = SlowStatsMerger( begin=self.begin, end=self.end, meta={}, stats={}, stats_by_db={}, count_per_minute={}, ) def test_initialization(self): """Test SlowStatsMerger initialization""" self.assertEqual(self.merger.begin, self.begin) self.assertEqual(self.merger.end, self.end) self.assertEqual(self.merger.meta, {}) self.assertEqual(self.merger.stats, {}) self.assertEqual(self.merger.stats_by_db, {}) self.assertEqual(self.merger.count_per_minute, {}) def test_merge_meta(self): """Test merging meta information""" log = SlowLog() log.digest = "test_digest" log.query = "SELECT * FROM table" log.is_internal = False self.merger.merge_meta(log) self.assertIn("test_digest", self.merger.meta) self.assertEqual(self.merger.meta["test_digest"], ("SELECT * FROM table", False)) def test_merge_meta_with_none_values(self): """Test merging meta with None values""" log = SlowLog() log.digest = "test_digest" log.query = None log.is_internal = None self.merger.merge_meta(log) self.assertEqual(self.merger.meta["test_digest"], ("", False)) def test_merge_stats(self): """Test merging stats""" log = SlowLog() log.digest = "test_digest" log.query_time = 10.0 log.mem_max = 100 log.disk_max = 200 self.merger.merge_stats(log) self.assertIn("test_digest", self.merger.stats) self.assertEqual(self.merger.stats["test_digest"].count, 1) self.assertEqual(self.merger.stats["test_digest"].query_time_sum, 10.0) def test_merge_stats_by_db(self): """Test merging stats by database""" log = SlowLog() log.digest = "test_digest" log.db = "test_db" log.query_time = 10.0 self.merger.merge_stats_by_db(log) self.assertIn("test_db", self.merger.stats_by_db) self.assertIn("test_digest", self.merger.stats_by_db["test_db"]) self.assertEqual(self.merger.stats_by_db["test_db"]["test_digest"].count, 1) def test_merge_count_per_minute(self): """Test merging count per minute""" log = SlowLog() log.digest = "test_digest" log.is_internal = False log.time = 1625097600 # Exactly on the minute boundary self.merger.merge_count_per_minute(log) # Should have entries for both empty digest and test_digest self.assertIn("", self.merger.count_per_minute) self.assertIn("test_digest", self.merger.count_per_minute) # Time should be rounded down to minute self.assertIn(1625097600, self.merger.count_per_minute[""]) self.assertIn(1625097600, self.merger.count_per_minute["test_digest"]) self.assertEqual(self.merger.count_per_minute[""][1625097600], 1) self.assertEqual(self.merger.count_per_minute["test_digest"][1625097600], 1) def test_merge_count_per_minute_internal(self): """Test that internal queries are not counted""" log = SlowLog() log.digest = "test_digest" log.is_internal = True log.time = 1625097600 self.merger.merge_count_per_minute(log) # Internal queries should not be counted self.assertEqual(len(self.merger.count_per_minute), 0) def test_merge_with_time_filter(self): """Test that logs outside time range are filtered""" log_before = SlowLog() log_before.digest = "digest_before" log_before.time = self.begin - 1 # Before begin log_after = SlowLog() log_after.digest = "digest_after" log_after.time = self.end # At or after end log_in_range = SlowLog() log_in_range.digest = "digest_in_range" log_in_range.time = self.begin + 100 # Within range self.merger.merge(log_before) self.merger.merge(log_after) self.merger.merge(log_in_range) # Only log_in_range should be processed self.assertNotIn("digest_before", self.merger.stats) self.assertNotIn("digest_after", self.merger.stats) self.assertIn("digest_in_range", self.merger.stats) def test_merge_without_digest(self): """Test that logs without digest are skipped""" log = SlowLog() log.digest = None log.time = self.begin + 100 self.merger.merge(log) # Should not create any stats self.assertEqual(len(self.merger.stats), 0) class TestMergeStats(unittest.TestCase): def test_merge_stats(self): """Test merging two SlowStats objects""" dst = SlowStats("test_digest") dst.count = 5 dst.query_time_sum = 50.0 dst.query_time_max = 15.0 dst.mem_max_sum = 500 dst.mem_max_max = 200 dst.disk_max_sum = 1000 dst.disk_max_max = 300 src = SlowStats("test_digest") src.count = 3 src.query_time_sum = 30.0 src.query_time_max = 20.0 # Larger than dst src.mem_max_sum = 300 src.mem_max_max = 250 # Larger than dst src.disk_max_sum = 600 src.disk_max_max = 400 # Larger than dst merge_stats(dst, src) self.assertEqual(dst.count, 8) self.assertEqual(dst.query_time_sum, 80.0) self.assertEqual(dst.query_time_max, 20.0) # Max of both self.assertEqual(dst.mem_max_sum, 800) self.assertEqual(dst.mem_max_max, 250) # Max of both self.assertEqual(dst.disk_max_sum, 1600) self.assertEqual(dst.disk_max_max, 400) # Max of both if __name__ == '__main__': unittest.main()