File size: 3,526 Bytes
3740bd7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """
Unit tests for utils module
Tests the PipelineTimer class
"""
import time
import unittest
from src.utils import PipelineTimer
class TestPipelineTimer(unittest.TestCase):
"""Test cases for PipelineTimer class"""
def setUp(self):
"""Set up test fixtures"""
self.timer = PipelineTimer()
def test_initialization(self):
"""Test timer initialization"""
self.assertIsNotNone(self.timer.start_time)
self.assertEqual(self.timer.step_times, {})
self.assertIsNone(self.timer.step_start)
self.assertIsNone(self.timer.current_step)
def test_reset(self):
"""Test timer reset functionality"""
# Add some data
self.timer.step_times = {"test": 100}
self.timer.current_step = "test"
# Reset
self.timer.reset()
# Verify reset
self.assertEqual(self.timer.step_times, {})
self.assertIsNone(self.timer.current_step)
def test_time_step_context_manager(self):
"""Test timing a step using context manager"""
with self.timer.time_step("test_step"):
time.sleep(0.1) # Sleep for 100ms
# Check that step was timed
self.assertIn("test_step", self.timer.step_times)
# Should be approximately 100ms (allowing some variance)
self.assertGreater(self.timer.step_times["test_step"], 90)
self.assertLess(self.timer.step_times["test_step"], 150)
def test_multiple_steps(self):
"""Test timing multiple steps"""
with self.timer.time_step("step1"):
time.sleep(0.05)
with self.timer.time_step("step2"):
time.sleep(0.05)
# Both steps should be recorded
self.assertIn("step1", self.timer.step_times)
self.assertIn("step2", self.timer.step_times)
self.assertEqual(len(self.timer.step_times), 2)
def test_get_total_time(self):
"""Test getting total elapsed time"""
time.sleep(0.1)
total_time = self.timer.get_total_time()
# Should be at least 100ms
self.assertGreater(total_time, 90)
def test_get_timing_summary(self):
"""Test getting timing summary"""
with self.timer.time_step("step1"):
time.sleep(0.05)
summary = self.timer.get_timing_summary()
# Check summary structure
self.assertIn("total_time_ms", summary)
self.assertIn("step_times", summary)
self.assertIn("timestamp", summary)
self.assertIn("step1", summary["step_times"])
def test_current_step_tracking(self):
"""Test that current_step is tracked correctly"""
self.assertIsNone(self.timer.current_step)
with self.timer.time_step("test_step"):
# During execution, current_step should be set
self.assertEqual(self.timer.current_step, "test_step")
# After execution, current_step should be None
self.assertIsNone(self.timer.current_step)
def test_exception_handling_in_timer(self):
"""Test that timer handles exceptions properly"""
try:
with self.timer.time_step("error_step"):
raise ValueError("Test error")
except ValueError:
pass
# Step should still be recorded even if exception occurred
self.assertIn("error_step", self.timer.step_times)
# current_step should be None after context manager exits
self.assertIsNone(self.timer.current_step)
if __name__ == "__main__":
unittest.main()
|