From 492b6f4f0208e45865e0808d1c44f7c6a811096e Mon Sep 17 00:00:00 2001 From: Clement Rouault Date: Fri, 8 Sep 2017 17:53:59 +0200 Subject: [PATCH] pytest added --leaks for leak check + test Handle.description --- pytesting/conftest.py | 75 +++++++++++++++++++++++++++++++++++-- pytesting/test_debugger.py | 6 +-- windows/__init__.py | 2 +- windows/winobject/handle.py | 21 +++++++++++ 4 files changed, 96 insertions(+), 8 deletions(-) diff --git a/pytesting/conftest.py b/pytesting/conftest.py index ea1f71b..e3f1a30 100644 --- a/pytesting/conftest.py +++ b/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") \ No newline at end of file + + +## 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("") \ No newline at end of file diff --git a/pytesting/test_debugger.py b/pytesting/test_debugger.py index 1bc5551..f31eacf 100644 --- a/pytesting/test_debugger.py +++ b/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)) diff --git a/windows/__init__.py b/windows/__init__.py index 4974385..daa7bd0 100644 --- a/windows/__init__.py +++ b/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() diff --git a/windows/winobject/handle.py b/windows/winobject/handle.py index ca9af9b..0de08c0 100644 --- a/windows/winobject/handle.py +++ b/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)