import os import sys import sqlite3 import tempfile import unittest import shutil 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.db import ( LogsDB, StatsDB, get_stmt_type, ) from o11ydiagnosis.subtasks.slowlogconverter.slowlog import SlowLog from o11ydiagnosis.subtasks.slowlogconverter.slowstats import SlowStats class TestLogsDB(unittest.TestCase): def setUp(self): """Set up test fixtures""" self.temp_dir = tempfile.mkdtemp() self.db_path = os.path.join(self.temp_dir, "test_logs.db") self.db = sqlite3.connect(self.db_path) self.db.isolation_level = None LogsDB.try_init_tables(self.db) def tearDown(self): """Clean up test fixtures""" self.db.close() if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) def test_try_init_tables(self): """Test that tables are created correctly""" # Check that tables exist cursor = self.db.cursor() cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name IN ('keys', 'slowlog') """) tables = [row[0] for row in cursor.fetchall()] self.assertIn('keys', tables) self.assertIn('slowlog', tables) cursor.close() def test_save_slowlogs(self): """Test saving slow logs (currently a no-op, but should not fail)""" # This method is currently a pass, but we test it doesn't fail logs = [] db_instance = LogsDB() db_instance.save_slowlogs(logs) # Should not raise an exception class TestStatsDB(unittest.TestCase): def setUp(self): """Set up test fixtures""" self.temp_dir = tempfile.mkdtemp() self.db_path = os.path.join(self.temp_dir, "test_stats.db") self.db = sqlite3.connect(self.db_path) self.db.isolation_level = None StatsDB.try_init_tables(self.db) def tearDown(self): """Clean up test fixtures""" self.db.close() if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir) def test_try_init_tables(self): """Test that tables are created correctly""" cursor = self.db.cursor() cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name IN ('keys', 'meta', 'stats', 'stats_by_db', 'metric_count_per_minute') """) tables = [row[0] for row in cursor.fetchall()] self.assertIn('keys', tables) self.assertIn('meta', tables) self.assertIn('stats', tables) self.assertIn('stats_by_db', tables) self.assertIn('metric_count_per_minute', tables) cursor.close() def test_save_and_query_keys(self): """Test saving and querying keys""" keys = ["key1", "key2", "key3"] StatsDB.save_keys(self.db, keys) queried_keys = StatsDB.query_keys(self.db) self.assertEqual(set(queried_keys), set(keys)) def test_save_and_query_meta(self): """Test saving and querying meta""" metas = { "digest1": ("SELECT * FROM table1", False), "digest2": ("INSERT INTO table2 VALUES (1)", True), } StatsDB.save_meta(self.db, metas) queried_meta = StatsDB.query_meta(self.db) self.assertEqual(len(queried_meta), 2) self.assertEqual(queried_meta["digest1"], ("SELECT * FROM table1", False)) self.assertEqual(queried_meta["digest2"], ("INSERT INTO table2 VALUES (1)", True)) def test_save_and_query_stats(self): """Test saving and querying stats""" stats = { "digest1": SlowStats("digest1"), "digest2": SlowStats("digest2"), } stats["digest1"].count = 10 stats["digest1"].query_time_sum = 100.0 stats["digest1"].query_time_max = 20.0 stats["digest2"].count = 5 stats["digest2"].query_time_sum = 50.0 stats["digest2"].query_time_max = 15.0 StatsDB.save_stats(self.db, stats) queried_stats = StatsDB.query_stats(self.db) self.assertEqual(len(queried_stats), 2) self.assertEqual(queried_stats["digest1"].count, 10) self.assertEqual(queried_stats["digest1"].query_time_sum, 100.0) self.assertEqual(queried_stats["digest2"].count, 5) self.assertEqual(queried_stats["digest2"].query_time_sum, 50.0) def test_save_and_query_stats_by_db(self): """Test saving and querying stats by database""" stats_by_db = { "db1": { "digest1": SlowStats("digest1"), }, "db2": { "digest2": SlowStats("digest2"), }, } stats_by_db["db1"]["digest1"].count = 10 stats_by_db["db1"]["digest1"].query_time_sum = 100.0 stats_by_db["db2"]["digest2"].count = 5 stats_by_db["db2"]["digest2"].query_time_sum = 50.0 StatsDB.save_stats_by_db(self.db, stats_by_db) queried_stats_by_db = StatsDB.query_stats_by_db(self.db) self.assertEqual(len(queried_stats_by_db), 2) self.assertIn("db1", queried_stats_by_db) self.assertIn("db2", queried_stats_by_db) self.assertEqual(queried_stats_by_db["db1"]["digest1"].count, 10) self.assertEqual(queried_stats_by_db["db2"]["digest2"].count, 5) def test_save_and_query_count_per_minute(self): """Test saving and querying count per minute""" count_per_minute = { "digest1": { 1625097600: 10, # 2021-07-01 00:00:00 1625097660: 5, # 2021-07-01 00:01:00 }, "digest2": { 1625097600: 3, }, } StatsDB.save_count_per_minute(self.db, count_per_minute) queried_count = StatsDB.query_count_per_minute(self.db) self.assertEqual(len(queried_count), 2) self.assertEqual(queried_count["digest1"][1625097600], 10) self.assertEqual(queried_count["digest1"][1625097660], 5) self.assertEqual(queried_count["digest2"][1625097600], 3) class TestGetStmtType(unittest.TestCase): def test_select_statement(self): """Test identifying SELECT statements""" self.assertEqual(get_stmt_type("SELECT * FROM table"), "Select") self.assertEqual(get_stmt_type("select * from table"), "Select") self.assertEqual(get_stmt_type(" SELECT * FROM table"), "Select") def test_insert_statement(self): """Test identifying INSERT statements""" self.assertEqual(get_stmt_type("INSERT INTO table VALUES (1)"), "Insert") self.assertEqual(get_stmt_type("insert into table values (1)"), "Insert") def test_update_statement(self): """Test identifying UPDATE statements""" self.assertEqual(get_stmt_type("UPDATE table SET col=1"), "Update") self.assertEqual(get_stmt_type("update table set col=1"), "Update") def test_delete_statement(self): """Test identifying DELETE statements""" self.assertEqual(get_stmt_type("DELETE FROM table"), "Delete") self.assertEqual(get_stmt_type("delete from table"), "Delete") def test_other_statements(self): """Test identifying other statement types""" self.assertEqual(get_stmt_type("ALTER TABLE table ADD COLUMN col INT"), "AlterTable") self.assertEqual(get_stmt_type("BEGIN"), "Begin") self.assertEqual(get_stmt_type("COMMIT"), "Commit") self.assertEqual(get_stmt_type("SHOW TABLES"), "Show") self.assertEqual(get_stmt_type("SET variable = value"), "Set") def test_empty_or_none(self): """Test handling empty or None SQL""" self.assertEqual(get_stmt_type(""), "Others") self.assertEqual(get_stmt_type(None), "Others") self.assertEqual(get_stmt_type(" "), "Others") self.assertEqual(get_stmt_type("UNKNOWN STATEMENT"), "Others") if __name__ == '__main__': unittest.main()