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
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 TestFindKeywords(unittest.TestCase):
39
def test_filter(self):
40
self.assertEqual(find_keywords("foo=bar $1234567890=routername baz=quux".split()),
41
{'foo': 'bar', 'baz': 'quux'})
44
class TestProcessFromUtil(unittest.TestCase):
47
self.fakestate = FakeState()
50
self.assertEqual(process_from_address(None, 80, self.fakestate), None)
52
def test_internal(self):
53
pfa = process_from_address('(Tor_internal)', 80, self.fakestate)
54
# depends on whether you have psutil installed or not, and on
55
# whether your system always has a PID 0 process...
56
self.assertEqual(pfa, self.fakestate.tor_pid)
58
@defer.inlineCallbacks
59
def test_real_addr(self):
60
## FIXME should choose a port which definitely isn't used.
62
## it's apparently frowned upon to use the "real" reactor in
63
## tests, but I was using "nc" before, and I think this is
65
from twisted.internet import reactor
66
listener = yield TCP4ServerEndpoint(reactor, 9887).listen(FakeProtocolFactory())
69
pid = process_from_address('0.0.0.0', 9887, self.fakestate)
71
listener.stopListening()
73
self.assertEqual(pid, os.getpid())
76
class TestDelete(unittest.TestCase):
78
def test_delete_file(self):
79
(fd, f) = tempfile.mkstemp()
80
os.write(fd, 'some\ndata\n')
82
self.assertTrue(os.path.exists(f))
83
delete_file_or_tree(f)
84
self.assertTrue(not os.path.exists(f))
86
def test_delete_tree(self):
87
d = tempfile.mkdtemp()
88
f = open(os.path.join(d, 'foo'), 'w')
92
self.assertTrue(os.path.exists(d))
93
self.assertTrue(os.path.isdir(d))
94
self.assertTrue(os.path.exists(os.path.join(d, 'foo')))
96
delete_file_or_tree(d)
98
self.assertTrue(not os.path.exists(d))
99
self.assertTrue(not os.path.exists(os.path.join(d, 'foo')))