ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-04-08 03:14:21 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #871 from exarkun/2928.remote-allocate_tcp_port-test_node.py

Browse Source 
Remove `allocate_tcp_port` from test_node.py

Fixes: ticket:2928
This commit is contained in:
Jean-Paul Calderone 2020-10-31 19:24:09 -04:00 committed by GitHub
commit 540c70f835
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 59 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
newsfragments/2928.minor Normal file
View File

86
src/allmydata/test/test_node.py
View File

@ -18,13 +18,19 @@ import time
import mock
from textwrap import dedent
from hypothesis import (
given,
)
from hypothesis.strategies import (
integers,
sets,
)
from unittest import skipIf
from twisted.trial import unittest
from twisted.internet import defer
from twisted.python import log
from foolscap.api import flushEventualQueue
import foolscap.logging.log
from twisted.application import service
@ -50,6 +56,9 @@ from allmydata.util.tor_provider import create as create_tor_provider
import allmydata.test.common_util as testutil
def port_numbers():
return integers(min_value=1, max_value=2 ** 16 - 1)
class LoggingMultiService(service.MultiService):
def log(self, msg, **kw):
pass
@ -81,22 +90,39 @@ def testing_tub(config_data=''):
class TestCase(testutil.SignalMixin, unittest.TestCase):
@defer.inlineCallbacks
def setUp(self):
testutil.SignalMixin.setUp(self)
self.parent = LoggingMultiService()
self.parent.startService()
self._available_port = yield iputil.allocate_tcp_port()
# We can use a made-up port number because these tests never actually
# try to bind the port. We'll use a low-numbered one that's likely to
# conflict with another service to prove it.
self._available_port = 22
def tearDown(self):
log.msg("%s.tearDown" % self.__class__.__name__)
testutil.SignalMixin.tearDown(self)
d = defer.succeed(None)
d.addCallback(lambda res: self.parent.stopService())
d.addCallback(flushEventualQueue)
return d
def _test_location(
self,
expected_addresses,
tub_port=None,
tub_location=None,
local_addresses=None,
):
"""
Verify that a Tub configured with the given *tub.port* and *tub.location*
values generates fURLs with the given addresses in its location hints.
def _test_location(self, basedir, expected_addresses, tub_port=None, tub_location=None, local_addresses=None):
:param [str] expected_addresses: The addresses which must appear in
the generated fURL for the test to pass. All addresses must
appear.
:param tub_port: If not ``None`` then a value for the *tub.port*
configuration item.
:param tub_location: If not ``None`` then a value for the *tub.port*
configuration item.
:param local_addresses: If not ``None`` then a list of addresses to
supply to the system under test as local addresses.
"""
basedir = self.mktemp()
create_node_dir(basedir, "testing")
config_data = "[node]\n"
if tub_port:
@ -104,34 +130,30 @@ class TestCase(testutil.SignalMixin, unittest.TestCase):
if tub_location is not None:
config_data += "tub.location = {}\n".format(tub_location)
if local_addresses:
if local_addresses is not None:
self.patch(iputil, 'get_local_addresses_sync',
lambda: local_addresses)
tub = testing_tub(config_data)
tub.setServiceParent(self.parent)
class Foo(object):
pass
furl = tub.registerReference(Foo())
for address in expected_addresses:
self.failUnlessIn(address, furl)
self.assertIn(address, furl)
def test_location1(self):
return self._test_location(basedir="test_node/test_location1",
expected_addresses=["192.0.2.0:1234"],
return self._test_location(expected_addresses=["192.0.2.0:1234"],
tub_location="192.0.2.0:1234")
def test_location2(self):
return self._test_location(basedir="test_node/test_location2",
expected_addresses=["192.0.2.0:1234", "example.org:8091"],
return self._test_location(expected_addresses=["192.0.2.0:1234", "example.org:8091"],
tub_location="192.0.2.0:1234,example.org:8091")
def test_location_not_set(self):
"""Checks the autogenerated furl when tub.location is not set."""
return self._test_location(
basedir="test_node/test_location3",
expected_addresses=[
"127.0.0.1:{}".format(self._available_port),
"192.0.2.0:{}".format(self._available_port),
@ -143,7 +165,6 @@ class TestCase(testutil.SignalMixin, unittest.TestCase):
def test_location_auto_and_explicit(self):
"""Checks the autogenerated furl when tub.location contains 'AUTO'."""
return self._test_location(
basedir="test_node/test_location4",
expected_addresses=[
"127.0.0.1:{}".format(self._available_port),
"192.0.2.0:{}".format(self._available_port),
@ -596,15 +617,26 @@ class Listeners(unittest.TestCase):
str(ctx.exception),
)
def test_multiple_ports(self):
# Randomly allocate a couple distinct port numbers to try out. The test
# never actually binds these port numbers so we don't care if they're "in
# use" on the system or not. We just want a couple distinct values we can
# check expected results against.
@given(ports=sets(elements=port_numbers(), min_size=2, max_size=2))
def test_multiple_ports(self, ports):
"""
When there are multiple listen addresses suggested by the ``tub.port`` and
``tub.location`` configuration, the node's *main* port listens on all
of them.
"""
basedir = self.mktemp()
create_node_dir(basedir, "testing")
port1 = iputil.allocate_tcp_port()
port2 = iputil.allocate_tcp_port()
config_fname = os.path.join(basedir, "tahoe.cfg")
os.mkdir(basedir)
os.mkdir(os.path.join(basedir, "private"))
port1, port2 = iter(ports)
port = ("tcp:%d:interface=127.0.0.1,tcp:%d:interface=127.0.0.1" %
(port1, port2))
location = "tcp:localhost:%d,tcp:localhost:%d" % (port1, port2)
with open(os.path.join(basedir, "tahoe.cfg"), "w") as f:
with open(config_fname, "w") as f:
f.write(BASE_CONFIG)
f.write("[node]\n")
f.write("tub.port = %s\n" % port)