1
from twisted.trial import unittest
2
from twisted.internet import defer
3
from twisted.internet.endpoints import TCP4ServerEndpoint
4
from twisted.internet.interfaces import IProtocolFactory
5
from zope.interface import implements
7
from txtorcon.util import process_from_address, delete_file_or_tree, find_keywords, ip_from_int, find_tor_binary
17
class FakeProtocolFactory:
18
implements(IProtocolFactory)
21
"IProtocolFactory API"
24
"IProtocolFactory API"
26
def buildProtocol(self, addr):
27
"IProtocolFactory API"
31
class TestIPFromInt(unittest.TestCase):
34
self.assertEqual(ip_from_int(0x7f000001), '127.0.0.1')
37
class TestGeoIpDatabaseLoading(unittest.TestCase):
39
def test_bad_geoip_path(self):
40
"fail gracefull if a db is missing"
41
from txtorcon import util
42
self.assertRaises(IOError, util.create_geoip, '_missing_path_')
45
class TestFindKeywords(unittest.TestCase):
47
def test_filter(self):
48
"make sure we filter out keys that look like router IDs"
49
self.assertEqual(find_keywords("foo=bar $1234567890=routername baz=quux".split()),
50
{'foo': 'bar', 'baz': 'quux'})
53
class TestNetLocation(unittest.TestCase):
55
def test_city_fails(self):
56
"make sure we don't fail if the city lookup excepts"
57
from txtorcon import util
60
class Thrower(object):
61
def record_by_addr(*args, **kw):
62
raise RuntimeError("testing failure")
64
nl = util.NetLocation('127.0.0.1')
65
self.assertEqual(None, nl.city)
70
def test_no_city_db(self):
71
"ensure we lookup from country if we have no city"
72
from txtorcon import util
74
origcountry = util.country
79
class CountryCoder(object):
80
def country_code_by_addr(self, ipaddr):
82
util.country = CountryCoder()
83
nl = util.NetLocation('127.0.0.1')
84
self.assertEqual(obj, nl.countrycode)
88
util.country = origcountry
90
def test_no_city_or_country_db(self):
91
"ensure we lookup from asn if we have no city or country"
92
from txtorcon import util
94
origcountry = util.country
101
def org_by_addr(*args, **kw):
102
raise RuntimeError("testing failure")
104
nl = util.NetLocation('127.0.0.1')
105
self.assertEqual('', nl.countrycode)
109
util.country = origcountry
113
class TestProcessFromUtil(unittest.TestCase):
116
self.fakestate = FakeState()
119
"ensure we do something useful on a None address"
120
self.assertEqual(process_from_address(None, 80, self.fakestate), None)
122
def test_internal(self):
123
"look up the (Tor_internal) PID"
124
pfa = process_from_address('(Tor_internal)', 80, self.fakestate)
125
# depends on whether you have psutil installed or not, and on
126
# whether your system always has a PID 0 process...
127
self.assertEqual(pfa, self.fakestate.tor_pid)
129
@defer.inlineCallbacks
130
def test_real_addr(self):
131
## FIXME should choose a port which definitely isn't used.
133
## it's apparently frowned upon to use the "real" reactor in
134
## tests, but I was using "nc" before, and I think this is
136
from twisted.internet import reactor
137
listener = yield TCP4ServerEndpoint(reactor, 9887).listen(FakeProtocolFactory())
140
pid = process_from_address('0.0.0.0', 9887, self.fakestate)
142
listener.stopListening()
144
self.assertEqual(pid, os.getpid())
147
class TestDelete(unittest.TestCase):
149
def test_delete_file(self):
150
(fd, f) = tempfile.mkstemp()
151
os.write(fd, 'some\ndata\n')
153
self.assertTrue(os.path.exists(f))
154
delete_file_or_tree(f)
155
self.assertTrue(not os.path.exists(f))
157
def test_delete_tree(self):
158
d = tempfile.mkdtemp()
159
f = open(os.path.join(d, 'foo'), 'w')
163
self.assertTrue(os.path.exists(d))
164
self.assertTrue(os.path.isdir(d))
165
self.assertTrue(os.path.exists(os.path.join(d, 'foo')))
167
delete_file_or_tree(d)
169
self.assertTrue(not os.path.exists(d))
170
self.assertTrue(not os.path.exists(os.path.join(d, 'foo')))
173
class TestFindTor(unittest.TestCase):
175
def test_simple_find_tor(self):
176
## just test that this doesn't raise an exception
179
def test_find_tor_globs(self):
180
"test searching by globs"
181
find_tor_binary(system_tor=False)
183
def test_find_tor_unfound(self):
184
"test searching by globs"
185
self.assertEqual(None, find_tor_binary(system_tor=False, globs=()))
added
removed
test/test_util.py
