pytest added --leaks for leak check + test Handle.description

pytesting/conftest.py

@@ -1,5 +1,6 @@
 import gc
 import pytest
+import collections
 
 import windows
 import windows.generated_def as gdef
@@ -40,6 +41,7 @@ def generate_pop_and_exit_fixtures(proc_popers, ids=[], dwCreationFlags=DEFAULT_
         except WindowsError as e:
             if not proc.is_exit:
                 raise
+        # print("DEL PROC")
         del proc
     return pop_and_exit_process
 
@@ -86,6 +88,9 @@ class HandleDebugger(object):
 current_process_hdebugger = HandleDebugger(windows.current_process.pid)
 current_process_hdebugger.refresh_handles()
 
+class NoLeakAssert(AssertionError):
+    pass
+
 
 
 @pytest.fixture()
@@ -98,9 +103,18 @@ def check_for_handle_leak(request):
     except Exception as e:
         leaked_handles = current_process_hdebugger.get_new_handle(x)
         leaked_handles_types = set(h.type for h in leaked_handles)
-        # print("ERROR FOR TYPE, newleaked = {0}".format(current_process_hdebugger.get_new_handle(x)))
-    leaked_handles_types -= set(['EtwRegistration', 'Key', 'DebugObject', 'Event'])
-    assert not leaked_handles_types, "Test Leaked <{0}> handles of types ({1})".format(len(leaked_handles), leaked_handles_types)
+
+    res = collections.defaultdict(list)
+    for lh in leaked_handles:
+        res[lh.type].append(lh)
+    # import pdb;pdb.set_trace()
+    for rmt in ['EtwRegistration', 'Key', 'DebugObject', 'Event']:
+        if rmt in res:
+            del res[rmt]
+    # leaked_handles_types -= set(['EtwRegistration', 'Key', 'DebugObject', 'Event'])
+    if res:
+        raise NoLeakAssert(res)
+    # assert not leaked_handles_types, "Test Leaked <{0}> handles of types ({1})".format(len(leaked_handles), leaked_handles_types)
 
 
 @pytest.fixture()
@@ -111,4 +125,57 @@ def check_for_gc_garbage(request):
         gc.collect()
         new_garbage = set(gc.garbage) - garbage_before
         assert not new_garbage, "Test generated uncollectable object ({0})".format(new_garbage)
-        # print("GC CHECK END")
+
+
+## Handle leak 'plugin'
+
+def pytest_addoption(parser):
+    parser.addoption("--leaks", action="store_true",
+                     default=False, help="Check windows handle leaks")
+
+
+def pytest_configure(config):
+    if not config.getoption("--leaks"):
+        return # no leaks check
+    config.addinivalue_line("usefixtures", "check_for_handle_leak")
+
+
+@pytest.hookimpl(hookwrapper=True, trylast=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    # print("Make report {0} | {1}".format(item, call))
+    if call.when == "teardown" and call.excinfo and type(call.excinfo.value) == NoLeakAssert:
+        x = outcome.get_result()
+        x.outcome = "failed"
+        # import pdb;pdb.set_trace()
+        x.LEAK = call.excinfo.value.args[0]
+
+
+@pytest.hookimpl(hookwrapper=True, trylast=True)
+def pytest_report_teststatus(report):
+    outcome = yield
+    if getattr(report, "LEAK", None):
+        report.outcome = "failed"
+        outcome.force_result(('leaked', '0', 'LEAKED'))
+
+
+@pytest.hookimpl(hookwrapper=True, trylast=True)
+def pytest_terminal_summary(terminalreporter, exitstatus):
+    outcome = yield
+    if terminalreporter.config.option.tbstyle != "no":
+        # import pdb;pdb.set_trace()
+        reports = terminalreporter.getreports('leaked')
+        if not reports:
+            return
+        terminalreporter.write_sep("=", "Handle leaks")
+        for leak_report in reports:
+            file, _, test = leak_report.location
+            terminalreporter.write_sep("_", "{0}::{1}".format(file, test))
+            for type, items in leak_report.LEAK.items():
+                terminalreporter.write_line("Leaked handles of type <{0}>".format(type) , Purple=True, bold=True)
+                for item in items:
+                    descr = item.description()
+                    if descr is None:
+                        descr = item.name
+                    terminalreporter.write_line(" * <{0}>".format(descr) , Purple=True, bold=True)
+            terminalreporter.write_line("")

pytesting/test_debugger.py

@@ -43,7 +43,7 @@ def get_debug_process_ndll(proc):
     ntdll_addr = proc.query_memory(proc_pc).AllocationBase
     return windows.pe_parse.GetPEFile(ntdll_addr, target=proc)
 
-@check_for_handle_leak
+# @check_for_handle_leak
 def test_simple_standard_breakpoint(proc32_64_debug):
     """Check that a standard Breakpoint method `trigger` is called with the correct informations"""
     class TSTBP(windows.debug.Breakpoint):
@@ -58,7 +58,7 @@ def test_simple_standard_breakpoint(proc32_64_debug):
     d.add_bp(TSTBP(LdrLoadDll))
     d.loop()
 
-@check_for_handle_leak
+# @check_for_handle_leak
 def test_simple_hwx_breakpoint(proc32_64_debug):
     """Test that simple HXBP are trigger"""
 
@@ -107,7 +107,7 @@ def test_multiple_hwx_breakpoint(proc32_64_debug):
     assert TSTBP.COUNTER == 4
 
 # @check_for_gc_garbage
-@check_for_handle_leak
+# @check_for_handle_leak
 def test_four_hwx_breakpoint_fail(proc32_64_debug):
     """Check that setting 4HXBP in the same thread fails"""
     # print("test_four_hwx_breakpoint_fail {0}".format(proc32_64_debug))

windows/__init__.py

@@ -16,7 +16,7 @@ from windows import winproxy
 from windows import winobject
 
 from winobject.system import System
-from winobject.process import CurrentProcess, CurrentThread
+from winobject.process import CurrentProcess, CurrentThread, WinProcess, WinThread
 
 
 system = System()

windows/winobject/handle.py

@@ -84,6 +84,27 @@ class Handle(SYSTEM_HANDLE):
         winproxy.DuplicateHandle(self.process.handle, self.wValue, windows.current_process.handle, ctypes.byref(res), dwOptions=DUPLICATE_SAME_ACCESS)
         return res.value
 
+    def description(self):
+        stype = self.type
+        descr_func = getattr(self, "description_" + stype, None)
+        if descr_func is None:
+            return None
+        return descr_func()
+
+    def description_Process(self):
+        proc = windows.WinProcess(handle=self.wValue)
+        res = str(proc)
+        del proc._handle
+        return res
+
+    def description_Thread(self):
+        thread = windows.WinThread(handle=self.wValue)
+        res = str(thread)
+        del thread._handle
+        return res
+
+
+
     def __repr__(self):
         return "<{0} value=<0x{1:x}> in process pid={2}>".format(type(self).__name__, self.wValue, self.dwProcessId)