Skip to content

pytester: slightly clean up LsofFdLeakChecker #7436

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jul 4, 2020
Merged

pytester: slightly clean up LsofFdLeakChecker #7436

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 23 additions & 27 deletions src/_pytest/pytester.py
View file
Open in desktop
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]]:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we not handle any errors? Just curious since the original method doesn't seem to have done it either. But, it looks like it is possible for subprocess.run to throw at least CalledProcessError and TimeoutExpired.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point! I think it would be good to handle errors from the lsof call here.

If the lsof command doesn't exist, the LsofFdLeakChecker plugin isn't registered at all, so the issue can't be that lsof isn't found at all, but it can still return an error itself.

I don't think TimeoutExpired can happen because we don't pass timeout to subprocess.run, but CalledProcessError can happen.

Then there's a question what to do if an error occurs. I guess in this case the most reasonable thing is to just skip the check for the particular test, and issue a warning.

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"))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just out of curiosity, why do you prefer run to check_output here?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's no good reason, only that the subprocess docs start with describing only the run function and say

The run() function was added in Python 3.5; if you need to retain compatibility with older versions, see the Older high-level API section.

And that "Older high-level API" includes check_output. So I just use run for everything.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks for explaining @bluetech

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