-
Notifications
You must be signed in to change notification settings - Fork 71
/
test_util.py
106 lines (72 loc) · 2.94 KB
/
test_util.py
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
103
104
105
106
from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet.endpoints import TCP4ServerEndpoint
from twisted.internet.interfaces import IProtocolFactory
from zope.interface import implements
from txtorcon.util import process_from_address, delete_file_or_tree, find_keywords, ip_from_int, find_tor_binary
import os
import tempfile
class FakeState:
tor_pid = 0
class FakeProtocolFactory:
implements(IProtocolFactory)
def doStart(self):
"IProtocolFactory API"
def doStop(self):
"IProtocolFactory API"
def buildProtocol(self, addr):
"IProtocolFactory API"
return None
class TestIPFromInt(unittest.TestCase):
def test_cast(self):
self.assertEqual(ip_from_int(0x7f000001), '127.0.0.1')
class TestFindKeywords(unittest.TestCase):
def test_filter(self):
self.assertEqual(find_keywords("foo=bar $1234567890=routername baz=quux".split()),
{'foo': 'bar', 'baz': 'quux'})
class TestProcessFromUtil(unittest.TestCase):
def setUp(self):
self.fakestate = FakeState()
def test_none(self):
self.assertEqual(process_from_address(None, 80, self.fakestate), None)
def test_internal(self):
pfa = process_from_address('(Tor_internal)', 80, self.fakestate)
# depends on whether you have psutil installed or not, and on
# whether your system always has a PID 0 process...
self.assertEqual(pfa, self.fakestate.tor_pid)
@defer.inlineCallbacks
def test_real_addr(self):
## FIXME should choose a port which definitely isn't used.
## it's apparently frowned upon to use the "real" reactor in
## tests, but I was using "nc" before, and I think this is
## preferable.
from twisted.internet import reactor
listener = yield TCP4ServerEndpoint(reactor, 9887).listen(FakeProtocolFactory())
try:
pid = process_from_address('0.0.0.0', 9887, self.fakestate)
finally:
listener.stopListening()
self.assertEqual(pid, os.getpid())
class TestDelete(unittest.TestCase):
def test_delete_file(self):
(fd, f) = tempfile.mkstemp()
os.write(fd, 'some\ndata\n')
os.close(fd)
self.assertTrue(os.path.exists(f))
delete_file_or_tree(f)
self.assertTrue(not os.path.exists(f))
def test_delete_tree(self):
d = tempfile.mkdtemp()
f = open(os.path.join(d, 'foo'), 'w')
f.write('foo\n')
f.close()
self.assertTrue(os.path.exists(d))
self.assertTrue(os.path.isdir(d))
self.assertTrue(os.path.exists(os.path.join(d, 'foo')))
delete_file_or_tree(d)
self.assertTrue(not os.path.exists(d))
self.assertTrue(not os.path.exists(os.path.join(d, 'foo')))
class TestFindTor(unittest.TestCase):
def test_simple_find_tor(self):
## just test that this doesn't raise an exception
find_tor_binary()