Skip to content

Commit

Permalink
use lambda + refact tests
Browse files Browse the repository at this point in the history
  • Loading branch information
giampaolo committed Apr 10, 2024
1 parent a02cab7 commit d51706d
Show file tree
Hide file tree
Showing 2 changed files with 15 additions and 6 deletions.
2 changes: 1 addition & 1 deletion psutil/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1507,7 +1507,7 @@ def remove(pid):
_pmap = pmap


process_iter.cache_clear = functools.partial(_pmap.clear)
process_iter.cache_clear = lambda: _pmap.clear() # noqa
process_iter.cache_clear.__doc__ = "Clear process_iter() internal cache."


Expand Down
19 changes: 14 additions & 5 deletions psutil/tests/test_system.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,8 @@
# ===================================================================


class TestProcessAPIs(PsutilTestCase):
def test_process_iter(self):
class TestProcessIter(PsutilTestCase):
def test_pid_presence(self):
self.assertIn(os.getpid(), [x.pid for x in psutil.process_iter()])
sproc = self.spawn_testproc()
self.assertIn(sproc.pid, [x.pid for x in psutil.process_iter()])
Expand All @@ -71,14 +71,14 @@ def test_process_iter(self):
p.wait()
self.assertNotIn(sproc.pid, [x.pid for x in psutil.process_iter()])

# assert there are no duplicates
def test_no_duplicates(self):
ls = [x for x in psutil.process_iter()]
self.assertEqual(
sorted(ls, key=lambda x: x.pid),
sorted(set(ls), key=lambda x: x.pid),
)

# emulate NSP for all PIDs
def test_emulate_nsp(self):
list(psutil.process_iter()) # populate cache
for x in range(2):
with mock.patch(
Expand All @@ -90,6 +90,7 @@ def test_process_iter(self):
)
psutil.process_iter.cache_clear() # repeat test without cache

def test_emulate_access_denied(self):
list(psutil.process_iter()) # populate cache
for x in range(2):
with mock.patch(
Expand All @@ -100,7 +101,7 @@ def test_process_iter(self):
list(psutil.process_iter(attrs=["cpu_times"]))
psutil.process_iter.cache_clear() # repeat test without cache

def test_process_iter_w_attrs(self):
def test_attrs(self):
for p in psutil.process_iter(attrs=['pid']):
self.assertEqual(list(p.info.keys()), ['pid'])
# yield again
Expand Down Expand Up @@ -128,6 +129,14 @@ def test_process_iter_w_attrs(self):
self.assertGreaterEqual(p.info['pid'], 0)
assert m.called

def test_cache_clear(self):
list(psutil.process_iter()) # populate cache
assert psutil._pmap
psutil.process_iter.cache_clear()
assert not psutil._pmap


class TestProcessAPIs(PsutilTestCase):
@unittest.skipIf(
PYPY and WINDOWS, "spawn_testproc() unreliable on PYPY + WINDOWS"
)
Expand Down

0 comments on commit d51706d

Please sign in to comment.