from aergia.aergia import Aergia import yappi import unittest class AergiaUnitTestCase(unittest.TestCase): interval = 0.01 Aergia = Aergia(interval) # yappi is a singleton def setUp(self): self.Aergia.clear() yappi.stop() yappi.clear_stats() yappi.set_clock_type('wall') yappi.set_context_id_callback(None) yappi.set_context_name_callback(None) yappi.set_tag_callback(None) def assert_reasonable_delay(self, func_name, time_expected, samples): '''Compares the results reported by Aergia for FUNC_NAME with time_expected.''' samples_actual = self.aergia_extract_values_by_func(samples, func_name) time_actual = self.aergia_expected_time(samples_actual) self.assert_roughly_equal(time_actual, time_expected) def assert_similar_delay(self, func_name, yappi_stats, aergia_samples): '''Compares the results reported by Aergia for FUNC_NAME with yappi.''' time_yappi = self.yappi_extract_values_by_func(yappi_stats, func_name) samples_aergia = self.aergia_extract_values_by_func( aergia_samples, func_name) time_aergia = self.aergia_expected_time(samples_aergia) self.assert_roughly_equal(time_aergia, time_yappi) def assert_roughly_equal(self, v1, v2): '''Throws an exception if values V1 and V2 are not within 0.035 seconds of each other. This number is influenced by instrumentation overhead, sampling inconsistency, etc.''' a = abs(v1 - v2) self.assertTrue(a <= .035, f'{v1} (actual) not roughly {v2} (expected)') def aergia_expected_time(self, total_samples): '''Given TOTAL_SAMPLES, returns the total time, using the sampling interval.''' return total_samples * self.interval def aergia_extract_values_by_func(self, samples, func_name): '''Returns the total number of samples for function name, given an Aergia SAMPLES object.''' return sum( value for key, value in samples.items() if key.func == func_name ) def yappi_extract_values_by_func(self, stats, func_name): '''Returns the total number of samples for function name, given a yappi stats object.''' for s in stats: if s.name == func_name: return s.ttot