WIP: Add support for tor using PySocks and optionally stem #2
7
minode/tests/common.py
Normal file
7
minode/tests/common.py
Normal file
|
@ -0,0 +1,7 @@
|
|||
import socket
|
||||
|
||||
try:
|
||||
socket.socket().bind(('127.0.0.1', 9050))
|
||||
tor_port_free = True
|
||||
except (OSError, socket.error):
|
||||
tor_port_free = False
|
|
@ -13,16 +13,13 @@ import psutil
|
|||
from minode.i2p import util
|
||||
from minode.structure import NetAddrNoPrefix
|
||||
|
||||
from .common import tor_port_free
|
||||
|
||||
try:
|
||||
socket.socket().bind(('127.0.0.1', 7656))
|
||||
i2p_port_free = True
|
||||
except (OSError, socket.error):
|
||||
i2p_port_free = False
|
||||
try:
|
||||
socket.socket().bind(('127.0.0.1', 9050))
|
||||
tor_port_free = True
|
||||
except (OSError, socket.error):
|
||||
tor_port_free = False
|
||||
|
||||
|
||||
class TestProcessProto(unittest.TestCase):
|
||||
|
|
64
minode/tests/test_tor.py
Normal file
64
minode/tests/test_tor.py
Normal file
|
@ -0,0 +1,64 @@
|
|||
"""Tests for tor module"""
|
||||
import collections
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from minode import shared
|
||||
|
||||
from .common import tor_port_free
|
||||
|
||||
try:
|
||||
from minode import tor
|
||||
except ImportError:
|
||||
tor = None
|
||||
|
||||
|
||||
Proxy = collections.namedtuple('Proxy', ['hostname', 'port'])
|
||||
|
||||
|
||||
@unittest.skipIf(
|
||||
tor_port_free or tor is None, 'Inapropriate environment for tor service')
|
||||
class TestTor(unittest.TestCase):
|
||||
"""A test case running the tor service"""
|
||||
tor = None
|
||||
_files = ['onion_dest_priv.key', 'onion_dest.pub']
|
||||
|
||||
@classmethod
|
||||
def cleanup(cls):
|
||||
"""Remove used files"""
|
||||
for f in cls._files:
|
||||
try:
|
||||
os.remove(os.path.join(shared.data_directory, f))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
shared.data_directory = tempfile.gettempdir()
|
||||
shared.socks_proxy = Proxy('127.0.0.1', 9050)
|
||||
cls.cleanup()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
if cls.tor:
|
||||
cls.tor.close()
|
||||
cls.cleanup()
|
||||
|
||||
def test_tor(self):
|
||||
"""Start the tor service as in main and check the environment"""
|
||||
self.tor = tor.start_tor_service()
|
||||
if not self.tor:
|
||||
self.fail('The tor service has hot started.')
|
||||
|
||||
with open(
|
||||
os.path.join(shared.data_directory, 'onion_dest.pub'),
|
||||
'r', encoding='ascii'
|
||||
) as key_file:
|
||||
onion_key = key_file.read()
|
||||
self.assertEqual(onion_key + '.onion', shared.onion_hostname)
|
||||
|
||||
# with open(
|
||||
# os.path.join(shared.data_directory, 'onion_dest_priv.key'), 'rb'
|
||||
# ) as key_file:
|
||||
# private_key = key_file.read()
|
|
@ -120,7 +120,7 @@ def start_tor_service():
|
|||
logging.info('Started hidden service %s', shared.onion_hostname)
|
||||
|
||||
if onionkey:
|
||||
return True
|
||||
return controller
|
||||
|
||||
try:
|
||||
with open(
|
||||
|
@ -144,4 +144,4 @@ def start_tor_service():
|
|||
logging.warning(
|
||||
'Error while saving onion service public key.', exc_info=True)
|
||||
|
||||
return True
|
||||
return controller
|
||||
|
|
Loading…
Reference in New Issue
Block a user