Skip to content

Commit

Permalink
Merge pull request #7436 from bluetech/cleanup-lsof
Browse files Browse the repository at this point in the history
pytester: slightly clean up LsofFdLeakChecker
  • Loading branch information
bluetech authored Jul 4, 2020
2 parents f76ac4c + 7b1ba7c commit 64dd700
Showing 1 changed file with 23 additions and 27 deletions.
50 changes: 23 additions & 27 deletions src/_pytest/pytester.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,21 +94,16 @@ def pytest_configure(config: Config) -> None:


class LsofFdLeakChecker:
def get_open_files(self):
out = self._exec_lsof()
open_files = self._parse_lsof_output(out)
return open_files

def _exec_lsof(self):
pid = os.getpid()
# py3: use subprocess.DEVNULL directly.
with open(os.devnull, "wb") as devnull:
return subprocess.check_output(
("lsof", "-Ffn0", "-p", str(pid)), stderr=devnull
).decode()

def _parse_lsof_output(self, out):
def isopen(line):
def get_open_files(self) -> List[Tuple[str, str]]:
out = subprocess.run(
("lsof", "-Ffn0", "-p", str(os.getpid())),
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
check=True,
universal_newlines=True,
).stdout

def isopen(line: str) -> bool:
return line.startswith("f") and (
"deleted" not in line
and "mem" not in line
Expand All @@ -130,9 +125,9 @@ def isopen(line):

return open_files

def matching_platform(self):
def matching_platform(self) -> bool:
try:
subprocess.check_output(("lsof", "-v"))
subprocess.run(("lsof", "-v"), check=True)
except (OSError, subprocess.CalledProcessError):
return False
else:
Expand All @@ -149,16 +144,17 @@ def pytest_runtest_protocol(self, item: Item) -> Generator[None, None, None]:
new_fds = {t[0] for t in lines2} - {t[0] for t in lines1}
leaked_files = [t for t in lines2 if t[0] in new_fds]
if leaked_files:
error = []
error.append("***** %s FD leakage detected" % len(leaked_files))
error.extend([str(f) for f in leaked_files])
error.append("*** Before:")
error.extend([str(f) for f in lines1])
error.append("*** After:")
error.extend([str(f) for f in lines2])
error.append(error[0])
error.append("*** function %s:%s: %s " % item.location)
error.append("See issue #2366")
error = [
"***** %s FD leakage detected" % len(leaked_files),
*(str(f) for f in leaked_files),
"*** Before:",
*(str(f) for f in lines1),
"*** After:",
*(str(f) for f in lines2),
"***** %s FD leakage detected" % len(leaked_files),
"*** function %s:%s: %s " % item.location,
"See issue #2366",
]
item.warn(pytest.PytestWarning("\n".join(error)))


Expand Down

0 comments on commit 64dd700

Please sign in to comment.