refactor: coverage

executing/_index_node_finder.py

@@ -78,6 +78,7 @@ class InstFingerprint:
 
     @property
     def opname(self) -> str:
+        # only used for debugging
         return dis.opname[self.opcode] # pragma: no cover
 
 
@@ -102,9 +103,9 @@ def normalize(value: Any) -> Any:
 
 
 class OpCodes:
-    def __getattr__(self, _: str) -> int:
+    def __getattr__(self, name: str) -> int:
         # getattr makes mypy happy
-        raise AttributeError() # pragma: no cover
+        assert False, "%s is not a valid opcode"%name
 
 
 for key, value in dis.opmap.items():
@@ -216,8 +217,9 @@ def add_offset(offset: Union[int, None]) -> None:
                 next=tuple(next_offsets),
             )
 
-    def skip(self, offset: int) -> Union[int, None]:
-        #return next_opcode(offset).offset
+
+
+    def next_opcode(self, offset: int) -> Optional[dis.Instruction]:
         done = set()
         while offset // 2 < len(self.instructions) and offset not in done:
             done.add(offset)
@@ -227,22 +229,26 @@ def skip(self, offset: int) -> Union[int, None]:
             elif inst.opcode in (opcodes.JUMP_FORWARD, opcodes.JUMP_ABSOLUTE):
                 offset = inst.argval
             else:
-                return offset
+                return inst
+        return None
 
+    def skip(self, offset: int) -> Optional[int]:
+        inst= self.next_opcode(offset)
+        if inst is not None:
+            return inst.offset
         return None
 
-    def next_opcode(self, offset: int) -> dis.Instruction:
-        done = set()
-        while offset // 2 < len(self.instructions) and offset not in done:
-            done.add(offset)
-            inst = self.instructions[offset // 2]
-            if inst.opcode in skip_opcodes:
-                offset += 2
-            elif inst.opcode in (opcodes.JUMP_FORWARD, opcodes.JUMP_ABSOLUTE):
-                offset = inst.argval
-            else:
-                return inst
-        assert False
+    def follow_jump(self, offset: int) -> dis.Instruction:
+        inst= self.next_opcode(offset)
+
+        # follow jump is used in optimize_jump, which is only used for instructions like JUMP_IF_X_OR_POP
+        # JUMP_IF_X_OR_POP is only generated for code like:
+        # a = b or c
+        # there can never be a loop and therefor next_opcode can never return None
+        assert inst is not None
+
+        return inst
+
 
     def optimize_jump(self, inst: dis.Instruction) -> Tuple[int, int]:
         pop = False
@@ -264,16 +270,15 @@ def optimize_jump(self, inst: dis.Instruction) -> Tuple[int, int]:
             assert False, inst
 
         while inst.opcode == jump_or_pop:
-            inst = self.next_opcode(inst.argval)
+            inst = self.follow_jump(inst.argval)
 
         if inst.opcode == pop_and_jump:
-            assert False, "TODO remove deadcode"
             pop = True
-            inst = self.next_opcode(inst.argval)
+            inst = self.follow_jump(inst.argval)
 
         if inst.opcode in pop_and_no_jump:
             pop = True
-            inst = self.next_opcode(inst.offset + 2)
+            inst = self.follow_jump(inst.offset + 2)
 
         return (pop_and_jump if pop else jump_or_pop, inst.offset)
 
@@ -289,7 +294,6 @@ def optimize_tuple(self, inst: dis.Instruction) -> Optional[Tuple[int, Any, int]
                     values.append(inst.argval)
 
                 if inst.offset // 2 == len(self.instructions):
-                    assert False, "TODO remove deadcode"
                     break
                 inst = self.instructions[inst.offset // 2 + 1]
 
@@ -468,7 +472,6 @@ def __getitem__(self, code: CodeType) -> Tuple[List[Match], List[dis.Instruction
                 if normalize(key) == code_key:
                     return (value, self.instructions[key])
 
-        assert False, "TODO remove deadcode"
         return ([], [])
 
 
@@ -538,7 +541,6 @@ def __init__(
                 raise MultipleMatches(nodes)
 
         if instructions[lasti // 2].opcode == opcodes.NOP:
-            assert False, "TODO remove deadcode"
             raise KnownIssue("can not map NOP")
 
         self.result, self.decorator = self.get_node(match, instructions, lasti, tree)
@@ -630,8 +632,6 @@ def known_issues(
                     ][-len(pattern) :]
                     if exit_instructions == pattern:
                         raise KnownIssue("call to __exit__ can not be mapped")
-                else:
-                    assert False, "TODO coverage"
 
         if (
             # todo find correct version
@@ -672,11 +672,10 @@ def get_node(
         instruction = instructions[index]
 
         if result is None:
-            assert False, "TODO remove deadcode"
             if instruction.opcode in (opcodes.NOP, opcodes.JUMP_ABSOLUTE):
                 raise KnownIssue("%s can not be mapped" % instruction.opname)
             else:
-                raise NoMatch()
+                raise NotOneValueFound("could not match %s"%instruction.opname)
 
         # workaround for method calls
         # thing.method()

tests/analyse.py

@@ -58,7 +58,7 @@ def inspect_opcode(self, bytecode, index, lineno):
             return result
         except RuntimeError:
             raise
-        except (KnownIssue,VerifierFailure,NoMatch,NotOneValueFound) as e:
+        except (KnownIssue,VerifierFailure,NoMatch,NotOneValueFound,AssertionError) as e:
             color= "[yellow]" if isinstance(e,KnownIssue) else "[red]"
             return color + type(e).__name__ + ": " + str(e).split("\n")[0]
         finally:
@@ -250,7 +250,7 @@ def main(self):
         self.dump_deadcode(filename)
 
 
-        self.profile.print_stats(sort="cumtime")
+        #self.profile.print_stats(sort="cumtime")
 
 
 if __name__ == "__main__":

tests/generate_small_sample.py

@@ -106,28 +106,28 @@ def main():
     console.print(f"Check files in tests/last_samples and {folder}:")
     console.print()
 
-    checked_files=[]
+    checked_files=set()
 
     if checked_files_json.exists():
-        checked_files=[Path(file) for file in  json.loads(checked_files_json.read_text())]
+        checked_files={Path(file) for file in  json.loads(checked_files_json.read_text())}
 
     with Progress() as progress:
 
         task_collect = progress.add_task(description="collect files ...", total=None)
 
         with get_context("spawn").Pool(maxtasksperchild=100) as p:
-            files = list(
+            files = set(
                 progress.track(
                     big_samples(folder),
                     task_id=task_collect,
                     description="collect files...",
                 )
             )
-            checked_files=[file for file in checked_files if file in files]
-            files=[file for file in files if file not in checked_files]
+            checked_files=checked_files & files
+            files=files - checked_files
             # check the already checked files last
             # this improves the propability of finding new issues sooner
-            files=files+checked_files
+            files=[*files,*checked_files]
 
             progress.reset(task_collect, description="check files...", total=len(files))
 
@@ -149,7 +149,7 @@ def main():
                     failing_code = filename.read_text()
                     break
                 else:
-                    checked_files.append(filename)
+                    checked_files.add(filename)
             else:
                 progress.stop()
                 console.print()