1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
"""
Integration tests for stem.util.proc functions against the tor process that
we're running.
"""
import os
import unittest
import test.runner
from stem.util import proc
class TestProc(unittest.TestCase):
def test_get_cwd(self):
"""
Checks that stem.util.proc.get_cwd matches our tor instance's cwd.
"""
if not proc.is_available():
test.runner.skip(self, '(proc unavailable)')
return
elif not test.runner.get_runner().is_ptraceable():
test.runner.skip(self, '(DisableDebuggerAttachment is set)')
return
runner = test.runner.get_runner()
runner_pid, tor_cwd = runner.get_pid(), runner.get_tor_cwd()
self.assertEquals(tor_cwd, proc.get_cwd(runner_pid))
def test_get_uid(self):
"""
Checks that stem.util.proc.get_uid matches our tor instance's uid.
"""
if not proc.is_available():
test.runner.skip(self, '(proc unavailable)')
return
tor_pid = test.runner.get_runner().get_pid()
self.assertEquals(os.geteuid(), proc.get_uid(tor_pid))
def test_get_memory_usage(self):
"""
Checks that stem.util.proc.get_memory_usage looks somewhat reasonable.
"""
if not proc.is_available():
test.runner.skip(self, '(proc unavailable)')
return
tor_pid = test.runner.get_runner().get_pid()
res_size, vir_size = proc.get_memory_usage(tor_pid)
# checks that they're larger than a kilobyte
self.assertTrue(res_size > 1024)
self.assertTrue(vir_size > 1024)
def test_get_stats(self):
"""
Checks that stem.util.proc.get_stats looks somewhat reasonable.
"""
if not proc.is_available():
test.runner.skip(self, '(proc unavailable)')
return
tor_pid = test.runner.get_runner().get_pid()
command, utime, stime, start_time = proc.get_stats(tor_pid, 'command', 'utime', 'stime', 'start time')
self.assertEquals('tor', command)
self.assertTrue(float(utime) > 0)
self.assertTrue(float(stime) >= 0)
self.assertTrue(float(start_time) > proc.get_system_start_time())
def test_get_connections(self):
"""
Checks for our control port in the stem.util.proc.get_connections output if
we have one.
"""
runner = test.runner.get_runner()
if not proc.is_available():
test.runner.skip(self, '(proc unavailable)')
return
elif not test.runner.Torrc.PORT in runner.get_options():
test.runner.skip(self, '(no control port)')
return
elif not test.runner.get_runner().is_ptraceable():
test.runner.skip(self, '(DisableDebuggerAttachment is set)')
return
# making a controller connection so that we have something to query for
with runner.get_tor_socket():
tor_pid = test.runner.get_runner().get_pid()
for conn in proc.get_connections(tor_pid):
if ('127.0.0.1', test.runner.CONTROL_PORT) == conn[:2]:
return
self.fail()
|
