added tests for analyzer

osaca/analyzer.py

@@ -25,32 +25,35 @@ class Analyzer(object):
         index_start = -1
         index_end = -1
         for i, line in enumerate(lines):
-            if line['instruction'] in mov_instr and lines[i + 1]['directive'] is not None:
-                source = line['operands']['source']
-                destination = line['operands']['destination']
-                # instruction pair matches, check for operands
-                if (
-                    'immediate' in source[0]
-                    and self.parser.normalize_imd(source[0]['immediate']) == mov_vals[0]
-                    and 'register' in destination[0]
-                    and self.parser.get_full_reg_name(destination[0]['register']) == mov_reg
-                ):
-                    # operands of first instruction match start, check for second one
-                    match, line_count = self.match_bytes(lines, i + 1, nop_bytes)
-                    if(match):
-                        # return first line after the marker
-                        index_start = i + 1 + line_count
-                elif (
-                    'immediate' in source[0]
-                    and self.parser.normalize_imd(source[0]['immediate']) == mov_vals[1]
-                    and 'register' in destination[0]
-                    and self.parser.get_full_reg_name(destination[0]['register']) == mov_reg
-                ):
-                    # operand of first instruction match end, check for second one
-                    match, line_count = self.match_bytes(lines, i + 1, nop_bytes)
-                    if(match):
-                        # return line of the marker
-                        index_end = i
+            try:
+                if line['instruction'] in mov_instr and lines[i + 1]['directive'] is not None:
+                    source = line['operands']['source']
+                    destination = line['operands']['destination']
+                    # instruction pair matches, check for operands
+                    if (
+                        'immediate' in source[0]
+                        and self.parser.normalize_imd(source[0]['immediate']) == mov_vals[0]
+                        and 'register' in destination[0]
+                        and self.parser.get_full_reg_name(destination[0]['register']) == mov_reg
+                    ):
+                        # operands of first instruction match start, check for second one
+                        match, line_count = self.match_bytes(lines, i + 1, nop_bytes)
+                        if(match):
+                            # return first line after the marker
+                            index_start = i + 1 + line_count
+                    elif (
+                        'immediate' in source[0]
+                        and self.parser.normalize_imd(source[0]['immediate']) == mov_vals[1]
+                        and 'register' in destination[0]
+                        and self.parser.get_full_reg_name(destination[0]['register']) == mov_reg
+                    ):
+                        # operand of first instruction match end, check for second one
+                        match, line_count = self.match_bytes(lines, i + 1, nop_bytes)
+                        if(match):
+                            # return line of the marker
+                            index_end = i
+            except TypeError:
+                print(i, line)
             if index_start != -1 and index_end != -1:
                 break
         return index_start, index_end

osaca/parser/base_parser.py

@@ -13,15 +13,14 @@ class BaseParser(object):
     def __init__(self):
         self.construct_parser()
 
-    def parse_file(self, file_content):
+    def parse_file(self, file_content, start_line=0):
         '''
-        Parse assembly file. This includes extracting of the marked kernel and
+        Parse assembly file. This includes *not* extracting of the marked kernel and
         the parsing of the instruction forms.
 
         :param str file_content: assembly code
+        :param int start_line: offset, if first line in file_content is meant to be not 1
         :return: list of instruction forms
-        :raises ValueError: if the marker_type attribute is unknown by the
-        function
         '''
         # Create instruction form list
         asm_instructions = []
@@ -29,7 +28,7 @@ class BaseParser(object):
         for i, line in enumerate(lines):
             if line == '':
                 continue
-            asm_instructions.append(self.parse_line(line, i + 1))
+            asm_instructions.append(self.parse_line(line, i + 1 + start_line))
         return asm_instructions
 
     def parse_line(self, line, line_number):

tests/all_tests.py

@@ -7,7 +7,8 @@ sys.path[0:0] = ['.', '..']
 suite = unittest.TestLoader().loadTestsFromNames(
     [
         'test_parser_x86att',
-        'test_parser_AArch64v81'
+        'test_parser_AArch64v81',
+        'test_analyzer'
     ]
 )
 

tests/test_analyzer.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""
+Unit tests for Analyzer object
+"""
+import os
+import unittest
+
+from osaca.analyzer import Analyzer
+from osaca.parser import ParserAArch64v81, ParserX86ATT
+
+
+class TestAnalyzer(unittest.TestCase):
+    def setUp(self):
+        self.parser_AArch = ParserAArch64v81()
+        self.parser_x86 = ParserX86ATT()
+        with open(self._find_file('triad-arm-iaca.s')) as f:
+            triad_code_arm = f.read()
+        with open(self._find_file('triad-x86-iaca.s')) as f:
+            triad_code_x86 = f.read()
+        self.parsed_AArch = self.parser_AArch.parse_file(triad_code_arm)
+        self.parsed_x86 = self.parser_x86.parse_file(triad_code_x86)
+
+    #################
+    # Test
+    #################
+
+    def test_marker_detection_AArch64(self):
+        analyzer = Analyzer(self.parsed_AArch, 'AArch64')
+        self.assertEquals(len(analyzer.kernel), 138)
+        self.assertEquals(analyzer.kernel[0]['line_number'], 307)
+        self.assertEquals(analyzer.kernel[-1]['line_number'], 444)
+
+    def test_marker_detection_x86(self):
+        analyzer = Analyzer(self.parsed_x86, 'x86')
+        self.assertEquals(len(analyzer.kernel), 9)
+        self.assertEquals(analyzer.kernel[0]['line_number'], 146)
+        self.assertEquals(analyzer.kernel[-1]['line_number'], 154)
+
+    def test_marker_matching_x86(self):
+        # preparation
+        bytes_1_line = '.byte     100,103,144\n'
+        bytes_2_lines = '.byte     100,103\n' + '.byte 144\n'
+        bytes_3_lines = (
+            '.byte     100 # IACA MARKER UTILITY\n'
+            + '.byte     103 # IACA MARKER UTILITY\n'
+            + '.byte     144 # IACA MARKER UTILITY\n'
+        )
+        mov_start_1 = 'movl      $111, %ebx # IACA START\n'
+        mov_start_2 = 'mov      $111, %ebx # IACA START\n'
+        mov_end_1 = 'movl      $222, %ebx # IACA END\n'
+        mov_end_2 = 'mov      $222, %ebx # IACA END\n'
+        prologue = 'movl    -92(%rbp), %r11d\n' + 'movl      $111, %ebx\n'
+        kernel = (
+            'vfmadd132sd (%r15,%rcx,8), %xmm5, %xmm0\n'
+            + 'vmovsd  %xmm0, (%r14,%rcx,8)\n'
+            + 'cmpl    %ebx, %ecx\n'
+            + 'jge .L8\n'
+        )
+        epilogue = '.LE9:\t\t#12.2\n' 'call    dummy\n'
+        kernel_length = len(list(filter(None, kernel.split('\n'))))
+
+        bytes_variations = [bytes_1_line, bytes_2_lines, bytes_3_lines]
+        mov_start_variations = [mov_start_1, mov_start_2]
+        mov_end_variations = [mov_end_1, mov_end_2]
+        # actual tests
+        for mov_start_var in mov_start_variations:
+            for bytes_var_1 in bytes_variations:
+                for mov_end_var in mov_end_variations:
+                    for bytes_var_2 in bytes_variations:
+                        sample_code = (
+                            prologue
+                            + mov_start_var
+                            + bytes_var_1
+                            + kernel
+                            + mov_end_var
+                            + bytes_var_2
+                            + epilogue
+                        )
+                        with self.subTest(
+                            mov_start=mov_start_var,
+                            bytes_start=bytes_var_1,
+                            mov_end=mov_end_var,
+                            bytes_end=bytes_var_2,
+                        ):
+                            sample_parsed = self.parser_x86.parse_file(sample_code)
+                            analyzer = Analyzer(sample_parsed, 'x86')
+                            self.assertEquals(len(analyzer.kernel), kernel_length)
+                            kernel_start = len(
+                                list(
+                                    filter(
+                                        None, (prologue + mov_start_var + bytes_var_1).split('\n')
+                                    )
+                                )
+                            )
+                            parsed_kernel = self.parser_x86.parse_file(
+                                kernel, start_line=kernel_start
+                            )
+                            self.assertEquals(analyzer.kernel, parsed_kernel)
+
+    @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(TestAnalyzer)
+    unittest.TextTestRunner(verbosity=2).run(suite)