#!/usr/bin/env python3 """ Unit tests for DB interface """ import os import unittest from io import StringIO import osaca.db_interface as dbi from osaca.db_interface import sanity_check from osaca.semantics import MachineModel class TestDBInterface(unittest.TestCase): @classmethod def setUpClass(self): sample_entry = { 'name': 'DoItRightAndDoItFast', 'operands': [ {'class': 'memory', 'offset': 'imd', 'base': 'gpr', 'index': 'gpr', 'scale': 8}, {'class': 'register', 'name': 'xmm'}, ], 'throughput': 1.25, 'latency': 125, 'uops': 6, } self.entry_csx = sample_entry.copy() self.entry_tx2 = sample_entry.copy() self.entry_zen1 = sample_entry.copy() # self.entry_csx['port_pressure'] = [1.25, 0, 1.25, 0.5, 0.5, 0.5, 0.5, 0, 1.25, 1.25, 0] self.entry_csx['port_pressure'] = [[5, '0156'], [1, '23'], [1, ['2D', '3D']]] # self.entry_tx2['port_pressure'] = [2.5, 2.5, 0, 0, 0.5, 0.5] self.entry_tx2['port_pressure'] = [[5, '01'], [1, '45']] del self.entry_tx2['operands'][1]['name'] self.entry_tx2['operands'][1]['prefix'] = 'x' # self.entry_zen1['port_pressure'] = [1, 1, 1, 1, 0, 1, 0, 0, 0, 0.5, 1, 0.5, 1] self.entry_zen1['port_pressure'] = [[4, '0123'], [1, '4'], [1, '89'], [2, ['8D', '9D']]] ########### # Tests ########### def test_add_single_entry(self): mm_csx = MachineModel('csx') mm_tx2 = MachineModel('tx2') mm_zen1 = MachineModel('zen1') num_entries_csx = len(mm_csx['instruction_forms']) num_entries_tx2 = len(mm_tx2['instruction_forms']) num_entries_zen1 = len(mm_zen1['instruction_forms']) mm_csx.set_instruction_entry(self.entry_csx) mm_tx2.set_instruction_entry(self.entry_tx2) mm_zen1.set_instruction_entry({'name': 'empty_operation'}) num_entries_csx = len(mm_csx['instruction_forms']) - num_entries_csx num_entries_tx2 = len(mm_tx2['instruction_forms']) - num_entries_tx2 num_entries_zen1 = len(mm_zen1['instruction_forms']) - num_entries_zen1 self.assertEqual(num_entries_csx, 1) self.assertEqual(num_entries_tx2, 1) self.assertEqual(num_entries_zen1, 1) def test_invalid_add(self): entry = {} with self.assertRaises(KeyError): MachineModel('csx').set_instruction_entry(entry) with self.assertRaises(TypeError): MachineModel('csx').set_instruction() def test_sanity_check(self): output = StringIO() # non-verbose sanity_check('csx', verbose=False, internet_check=False, output_file=output) sanity_check('tx2', verbose=False, internet_check=False, output_file=output) sanity_check('zen1', verbose=False, internet_check=False, output_file=output) # verbose sanity_check('csx', verbose=True, internet_check=False, output_file=output) sanity_check('tx2', verbose=True, internet_check=False, output_file=output) sanity_check('zen1', verbose=True, internet_check=False, output_file=output) def test_ibench_import(self): # only check import without dumping the DB file (takes too much time) with open(self._find_file('ibench_import_x86.dat')) as input_file: input_data = input_file.readlines() entries = dbi._get_ibench_output(input_data, 'x86') self.assertEqual(len(entries), 3) for _, e in entries.items(): self.assertIsNotNone(e['throughput']) self.assertIsNotNone(e['latency']) with open(self._find_file('ibench_import_aarch64.dat')) as input_file: input_data = input_file.readlines() entries = dbi._get_ibench_output(input_data, 'aarch64') self.assertEqual(len(entries), 4) for _, e in entries.items(): self.assertIsNotNone(e['throughput']) self.assertIsNotNone(e['latency']) def test_asmbench_import(self): # only check import without dumping the DB file (takes too much time) with open(self._find_file('asmbench_import_x86.dat')) as input_file: input_data = input_file.readlines() entries = dbi._get_asmbench_output(input_data, 'x86') self.assertEqual(len(entries), 3) for _, e in entries.items(): self.assertIsNotNone(e['throughput']) self.assertIsNotNone(e['latency']) with open(self._find_file('asmbench_import_aarch64.dat')) as input_file: input_data = input_file.readlines() entries = dbi._get_asmbench_output(input_data, 'aarch64') self.assertEqual(len(entries), 4) for _, e in entries.items(): self.assertIsNotNone(e['throughput']) self.assertIsNotNone(e['latency']) # remove empty line => no import since broken format del input_data[3] entries = dbi._get_asmbench_output(input_data, 'aarch64') self.assertEqual(len(entries), 0) with self.assertRaises(ValueError): dbi.import_benchmark_output( 'csx', 'invalid_bench_type', self._find_file('asmbench_import_x86.dat') ) with self.assertRaises(AssertionError): dbi.import_benchmark_output('csx', 'ibench', 'invalid_file') def test_online_scraping(self): # addpd -- suspicious instruction, normal URL instr_1 = ['addpd', (True, '(r) (r,w)')] self.assertEqual(dbi._scrape_from_felixcloutier(instr_1[0]), instr_1[1]) # movpd -- not suspicious, instr_2 = ['movapd', (False, '(r) (w)')] self.assertEqual(dbi._scrape_from_felixcloutier(instr_2[0]), instr_2[1]) # vfmadd132pd -- only in combined view with 213/231. # No 2-operand version, therefore, empty string instr_3 = ['vfmadd132pd', (True, '')] self.assertEqual(dbi._scrape_from_felixcloutier(instr_3[0]), instr_3[1]) ################## # Helper functions ################## @staticmethod def _find_file(name): testdir = os.path.dirname(__file__) name = os.path.join(testdir, 'test_files', name) assert os.path.exists(name) return name if __name__ == '__main__': suite = unittest.TestLoader().loadTestsFromTestCase(TestDBInterface) unittest.TextTestRunner(verbosity=2).run(suite)