process_rss_test.py (1840B)
1 """Disappearing procfs entries are not zero samples or permission to run unguarded.""" 2 import subprocess 3 import unittest 4 from pathlib import Path 5 from unittest.mock import Mock, patch 6 from process_rss import ChildRSS 7 8 class ProcfsExitTests(unittest.TestCase): 9 def reader(self, *, exited): 10 sample = ChildRSS.__new__(ChildRSS) 11 sample.pid = 123 12 sample.parent = 1 13 sample.namespace_index = 0 14 sample.proc = Mock(pid=123) 15 sample.proc.poll.side_effect = [None,0 if exited else None] 16 sample.proc.wait.side_effect = subprocess.TimeoutExpired('child',.25) 17 return sample 18 19 def disappearing(self, target, *, exited): 20 sample = self.reader(exited=exited) 21 def read(path, *args, **kwargs): 22 if str(path).endswith(target): 23 raise ProcessLookupError('process exited during procfs read') 24 return 'Pid:\t123\nPPid:\t1\nNSpid:\t123\nState:\tR\n' 25 with patch.object(Path,'read_text',read), patch.object(Path,'glob',return_value=[]): 26 return sample.sample() 27 28 def test_status_esrch_after_exit_is_no_sample(self): 29 self.assertIsNone(self.disappearing('status',exited=True)) 30 31 def test_statm_esrch_after_exit_is_no_sample(self): 32 self.assertIsNone(self.disappearing('statm',exited=True)) 33 34 def test_unreaped_unobservable_child_remains_fatal(self): 35 with self.assertRaisesRegex(RuntimeError,'refusing an unguarded run'): 36 self.disappearing('statm',exited=False) 37 38 def test_permission_failure_is_not_silently_ignored(self): 39 sample = self.reader(exited=True) 40 with patch.object(Path,'read_text',side_effect=PermissionError('denied')): 41 with self.assertRaises(PermissionError): 42 sample.sample() 43 44 if __name__ == '__main__': 45 unittest.main()