source: trunk/src/allmydata/test/test_node.py

from __future__ import annotations

import base64
import os
import stat
import sys
import time
from textwrap import dedent
import configparser

from hypothesis import (
    given,
)
from hypothesis.strategies import (
    integers,
    sets,
)

from unittest import skipIf

from twisted.python.filepath import (
    FilePath,
)
from twisted.python.runtime import platform
from twisted.trial import unittest
from twisted.internet import defer

import foolscap.logging.log

from twisted.application import service
from allmydata.node import (
    PortAssignmentRequired,
    PrivacyError,
    tub_listen_on,
    create_tub_options,
    create_main_tub,
    create_node_dir,
    create_default_connection_handlers,
    create_connection_handlers,
    config_from_string,
    read_config,
    MissingConfigEntry,
    _tub_portlocation,
    formatTimeTahoeStyle,
    UnescapedHashError,
)
from allmydata.introducer.server import create_introducer
from allmydata import client

from allmydata.util import fileutil, iputil
from allmydata.util.namespace import Namespace
from allmydata.util.configutil import (
    ValidConfiguration,
    UnknownConfigError,
)

from allmydata.util.i2p_provider import create as create_i2p_provider
from allmydata.util.tor_provider import create as create_tor_provider
import allmydata.test.common_util as testutil

from .common import (
    ConstantAddresses,
    SameProcessStreamEndpointAssigner,
    UseNode,
    superuser,
)

def port_numbers():
    return integers(min_value=1, max_value=2 ** 16 - 1)

class LoggingMultiService(service.MultiService):
    def log(self, msg, **kw):
        pass


# see https://tahoe-lafs.org/trac/tahoe-lafs/ticket/2946
def testing_tub(reactor, config_data=''):
    """
    Creates a 'main' Tub for testing purposes, from config data
    """
    basedir = 'dummy_basedir'
    config = config_from_string(basedir, 'DEFAULT_PORTNUMFILE_BLANK', config_data)
    fileutil.make_dirs(os.path.join(basedir, 'private'))

    i2p_provider = create_i2p_provider(reactor, config)
    tor_provider = create_tor_provider(reactor, config)
    handlers = create_connection_handlers(config, i2p_provider, tor_provider)
    default_connection_handlers, foolscap_connection_handlers = handlers
    tub_options = create_tub_options(config)

    main_tub = create_main_tub(
        config, tub_options, default_connection_handlers,
        foolscap_connection_handlers, i2p_provider, tor_provider,
        cert_filename='DEFAULT_CERTFILE_BLANK'
    )
    return main_tub


class TestCase(testutil.SignalMixin, unittest.TestCase):

    def setUp(self):
        testutil.SignalMixin.setUp(self)
        self.parent = LoggingMultiService()
        # 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
        self.port_assigner = SameProcessStreamEndpointAssigner()
        self.port_assigner.setUp()
        self.addCleanup(self.port_assigner.tearDown)

    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.

        :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.
        """
        from twisted.internet import reactor

        basedir = self.mktemp()
        create_node_dir(basedir, "testing")
        if tub_port is None:
            # Always configure a usable tub.port address instead of relying on
            # the automatic port assignment.  The automatic port assignment is
            # prone to collisions and spurious test failures.
            _, tub_port = self.port_assigner.assign(reactor)

        config_data = "[node]\n"
        config_data += "tub.port = {}\n".format(tub_port)

        # If they wanted a certain location, go for it.  This probably won't
        # agree with the tub.port value we set but that only matters if
        # anything tries to use this to establish a connection ... which
        # nothing in this test suite will.
        if tub_location is not None:
            config_data += "tub.location = {}\n".format(tub_location)

        if local_addresses is not None:
            self.patch(iputil, 'get_local_addresses_sync',
                       lambda: local_addresses)

        tub = testing_tub(reactor, config_data)

        class Foo:
            pass

        furl = tub.registerReference(Foo())
        for address in expected_addresses:
            self.assertIn(address, furl)

    def test_location1(self):
        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(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(
            expected_addresses=[
                "127.0.0.1:{}".format(self._available_port),
                "192.0.2.0:{}".format(self._available_port),
            ],
            tub_port=self._available_port,
            local_addresses=["127.0.0.1", "192.0.2.0"],
        )

    def test_location_auto_and_explicit(self):
        """Checks the autogenerated furl when tub.location contains 'AUTO'."""
        return self._test_location(
            expected_addresses=[
                "127.0.0.1:{}".format(self._available_port),
                "192.0.2.0:{}".format(self._available_port),
                "example.com:4321",
            ],
            tub_port=self._available_port,
            tub_location="AUTO,example.com:{}".format(self._available_port),
            local_addresses=["127.0.0.1", "192.0.2.0", "example.com:4321"],
        )

    def test_tahoe_cfg_utf8(self):
        basedir = "test_node/test_tahoe_cfg_utf8"
        fileutil.make_dirs(basedir)
        f = open(os.path.join(basedir, 'tahoe.cfg'), 'wb')
        f.write(u"\uFEFF[node]\n".encode('utf-8'))
        f.write(u"nickname = \u2621\n".encode('utf-8'))
        f.close()

        config = read_config(basedir, "")
        self.failUnlessEqual(config.get_config("node", "nickname"),
                             u"\u2621")

    def test_tahoe_cfg_hash_in_name(self):
        basedir = "test_node/test_cfg_hash_in_name"
        nickname = "Hash#Bang!" # a clever nickname containing a hash
        fileutil.make_dirs(basedir)
        f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
        f.write("[node]\n")
        f.write("nickname = %s\n" % (nickname,))
        f.close()

        config = read_config(basedir, "")
        self.failUnless(config.nickname == nickname)

    def test_hash_in_furl(self):
        """
        Hashes in furl options are not allowed, resulting in exception.
        """
        basedir = self.mktemp()
        fileutil.make_dirs(basedir)
        with open(os.path.join(basedir, 'tahoe.cfg'), 'wt') as f:
            f.write("[node]\n")
            f.write("log_gatherer.furl = lalal#onohash\n")

        config = read_config(basedir, "")
        with self.assertRaises(UnescapedHashError):
            config.get_config("node", "log_gatherer.furl")

    def test_missing_config_item(self):
        """
        If a config item is missing:

        1. Given a default, return default.
        2. Otherwise, raise MissingConfigEntry.
        """
        basedir = self.mktemp()
        fileutil.make_dirs(basedir)
        with open(os.path.join(basedir, 'tahoe.cfg'), 'wt') as f:
            f.write("[node]\n")
        config = read_config(basedir, "")

        self.assertEqual(config.get_config("node", "log_gatherer.furl", "def"), "def")
        with self.assertRaises(MissingConfigEntry):
            config.get_config("node", "log_gatherer.furl")

    def test_missing_config_section(self):
        """
        Enumerating a missing section returns empty dict
        """
        basedir = self.mktemp()
        fileutil.make_dirs(basedir)
        with open(os.path.join(basedir, 'tahoe.cfg'), 'w'):
            pass
        config = read_config(basedir, "")
        self.assertEqual(
            config.enumerate_section("not-a-section"),
            {}
        )

    def test_config_required(self):
        """
        Asking for missing (but required) configuration is an error
        """
        basedir = u"test_node/test_config_required"
        config = read_config(basedir, "portnum")

        with self.assertRaises(Exception):
            config.get_config_from_file("it_does_not_exist", required=True)

    def test_config_items(self):
        """
        All items in a config section can be retrieved.
        """
        basedir = u"test_node/test_config_items"
        create_node_dir(basedir, "testing")

        with open(os.path.join(basedir, 'tahoe.cfg'), 'wt') as f:
            f.write(dedent(
                """
                [node]
                nickname = foo
                timeout.disconnect = 12
                """
            ))
        config = read_config(basedir, "portnum")
        self.assertEqual(
            config.items("node"),
            [("nickname", "foo"),
             ("timeout.disconnect", "12"),
            ],
        )
        self.assertEqual(
            config.items("node", [("unnecessary", "default")]),
            [("nickname", "foo"),
             ("timeout.disconnect", "12"),
            ],
        )


    def test_config_items_missing_section(self):
        """
        If a default is given for a missing section, the default is used.

        Lacking both default and section, an error is raised.
        """
        basedir = self.mktemp()
        create_node_dir(basedir, "testing")

        with open(os.path.join(basedir, 'tahoe.cfg'), 'wt') as f:
            f.write("")

        config = read_config(basedir, "portnum")
        with self.assertRaises(configparser.NoSectionError):
            config.items("nosuch")
        default = [("hello", "world")]
        self.assertEqual(config.items("nosuch", default), default)

    @skipIf(platform.isWindows(), "We don't know how to set permissions on Windows.")
    @skipIf(superuser, "cannot test as superuser with all permissions")
    def test_private_config_unreadable(self):
        """
        Asking for inaccessible private config is an error
        """
        basedir = u"test_node/test_private_config_unreadable"
        create_node_dir(basedir, "testing")
        config = read_config(basedir, "portnum")
        config.get_or_create_private_config("foo", "contents")
        fname = os.path.join(basedir, "private", "foo")
        os.chmod(fname, 0)

        with self.assertRaises(Exception):
            config.get_or_create_private_config("foo")

    @skipIf(platform.isWindows(), "We don't know how to set permissions on Windows.")
    @skipIf(superuser, "cannot test as superuser with all permissions")
    def test_private_config_unreadable_preexisting(self):
        """
        error if reading private config data fails
        """
        basedir = u"test_node/test_private_config_unreadable_preexisting"
        create_node_dir(basedir, "testing")
        config = read_config(basedir, "portnum")
        fname = os.path.join(basedir, "private", "foo")
        with open(fname, "w") as f:
            f.write("stuff")
        os.chmod(fname, 0)

        with self.assertRaises(Exception):
            config.get_private_config("foo")

    def test_private_config_missing(self):
        """
        a missing config with no default is an error
        """
        basedir = u"test_node/test_private_config_missing"
        create_node_dir(basedir, "testing")
        config = read_config(basedir, "portnum")

        with self.assertRaises(MissingConfigEntry):
            config.get_or_create_private_config("foo")

    def test_private_config(self):
        basedir = u"test_node/test_private_config"
        privdir = os.path.join(basedir, "private")
        fileutil.make_dirs(privdir)
        f = open(os.path.join(privdir, 'already'), 'wt')
        f.write("secret")
        f.close()

        basedir = fileutil.abspath_expanduser_unicode(basedir)
        config = config_from_string(basedir, "", "")

        self.assertEqual(config.get_private_config("already"), "secret")
        self.assertEqual(config.get_private_config("not", "default"), "default")
        self.assertRaises(MissingConfigEntry, config.get_private_config, "not")
        value = config.get_or_create_private_config("new", "start")
        self.assertEqual(value, "start")
        self.assertEqual(config.get_private_config("new"), "start")
        counter = []
        def make_newer():
            counter.append("called")
            return "newer"
        value = config.get_or_create_private_config("newer", make_newer)
        self.assertEqual(len(counter), 1)
        self.assertEqual(value, "newer")
        self.assertEqual(config.get_private_config("newer"), "newer")

        value = config.get_or_create_private_config("newer", make_newer)
        self.assertEqual(len(counter), 1) # don't call unless necessary
        self.assertEqual(value, "newer")

    @skipIf(superuser, "cannot test as superuser with all permissions")
    def test_write_config_unwritable_file(self):
        """
        Existing behavior merely logs any errors upon writing
        configuration files; this bad behavior should probably be
        fixed to do something better (like fail entirely). See #2905
        """
        basedir = "test_node/configdir"
        fileutil.make_dirs(basedir)
        config = config_from_string(basedir, "", "")
        with open(os.path.join(basedir, "bad"), "w") as f:
            f.write("bad")
        os.chmod(os.path.join(basedir, "bad"), 0o000)

        config.write_config_file("bad", "some value")

        errs = self.flushLoggedErrors(IOError)
        self.assertEqual(1, len(errs))

    def test_timestamp(self):
        # this modified logger doesn't seem to get used during the tests,
        # probably because we don't modify the LogObserver that trial
        # installs (only the one that twistd installs). So manually exercise
        # it a little bit.
        t = formatTimeTahoeStyle("ignored", time.time())
        self.failUnless("Z" in t)
        t2 = formatTimeTahoeStyle("ignored", int(time.time()))
        self.failUnless("Z" in t2)

    def test_secrets_dir(self):
        basedir = "test_node/test_secrets_dir"
        create_node_dir(basedir, "testing")
        self.failUnless(os.path.exists(os.path.join(basedir, "private")))

    def test_secrets_dir_protected(self):
        if "win32" in sys.platform.lower() or "cygwin" in sys.platform.lower():
            # We don't know how to test that unprivileged users can't read this
            # thing.  (Also we don't know exactly how to set the permissions so
            # that unprivileged users can't read this thing.)
            raise unittest.SkipTest("We don't know how to set permissions on Windows.")
        basedir = "test_node/test_secrets_dir_protected"
        create_node_dir(basedir, "nothing to see here")

        # make sure private dir was created with correct modes
        privdir = os.path.join(basedir, "private")
        st = os.stat(privdir)
        bits = stat.S_IMODE(st[stat.ST_MODE])
        self.failUnless(bits & 0o001 == 0, bits)

    @defer.inlineCallbacks
    def test_logdir_is_str(self):
        from twisted.internet import reactor

        basedir = FilePath(self.mktemp())
        fixture = UseNode(None, None, basedir, "pb://introducer/furl", {}, reactor=reactor)
        fixture.setUp()
        self.addCleanup(fixture.cleanUp)

        ns = Namespace()
        ns.called = False
        def call_setLogDir(logdir):
            ns.called = True
            self.failUnless(isinstance(logdir, str), logdir)
        self.patch(foolscap.logging.log, 'setLogDir', call_setLogDir)

        yield fixture.create_node()
        self.failUnless(ns.called)

    def test_set_config_unescaped_furl_hash(self):
        """
        ``_Config.set_config`` raises ``UnescapedHashError`` if the item being set
        is a furl and the value includes ``"#"`` and does not set the value.
        """
        basedir = self.mktemp()
        new_config = config_from_string(basedir, "", "")
        with self.assertRaises(UnescapedHashError):
            new_config.set_config("foo", "bar.furl", "value#1")
        with self.assertRaises(MissingConfigEntry):
            new_config.get_config("foo", "bar.furl")

    def test_set_config_new_section(self):
        """
        ``_Config.set_config`` can be called with the name of a section that does
        not already exist to create that section and set an item in it.
        """
        basedir = self.mktemp()
        new_config = config_from_string(basedir, "", "", ValidConfiguration.everything())
        new_config.set_config("foo", "bar", "value1")
        self.assertEqual(
            new_config.get_config("foo", "bar"),
            "value1"
        )

    def test_set_config_replace(self):
        """
        ``_Config.set_config`` can be called with a section and item that already
        exists to change an existing value to a new one.
        """
        basedir = self.mktemp()
        new_config = config_from_string(basedir, "", "", ValidConfiguration.everything())
        new_config.set_config("foo", "bar", "value1")
        new_config.set_config("foo", "bar", "value2")
        self.assertEqual(
            new_config.get_config("foo", "bar"),
            "value2"
        )

    def test_set_config_write(self):
        """
        ``_Config.set_config`` persists the configuration change so it can be
        re-loaded later.
        """
        # Let our nonsense config through
        valid_config = ValidConfiguration.everything()
        basedir = FilePath(self.mktemp())
        basedir.makedirs()
        cfg = basedir.child(b"tahoe.cfg")
        cfg.setContent(b"")
        new_config = read_config(basedir.path, "", [], valid_config)
        new_config.set_config("foo", "bar", "value1")
        loaded_config = read_config(basedir.path, "", [], valid_config)
        self.assertEqual(
            loaded_config.get_config("foo", "bar"),
            "value1",
        )

    def test_set_config_rejects_invalid_config(self):
        """
        ``_Config.set_config`` raises ``UnknownConfigError`` if the section or
        item is not recognized by the validation object and does not set the
        value.
        """
        # Make everything invalid.
        valid_config = ValidConfiguration.nothing()
        new_config = config_from_string(self.mktemp(), "", "", valid_config)
        with self.assertRaises(UnknownConfigError):
            new_config.set_config("foo", "bar", "baz")
        with self.assertRaises(MissingConfigEntry):
            new_config.get_config("foo", "bar")


def _stub_get_local_addresses_sync():
    """
    A function like ``allmydata.util.iputil.get_local_addresses_sync``.
    """
    return ["LOCAL"]


def _stub_allocate_tcp_port():
    """
    A function like ``allmydata.util.iputil.allocate_tcp_port``.
    """
    return 999

def _stub_none():
    """
    A function like ``_stub_allocate_tcp`` or ``_stub_get_local_addresses_sync``
    but that return an empty list since ``allmydata.node._tub_portlocation`` requires a
    callable for paramter 1 and 2 counting from 0.
    """
    return []


class TestMissingPorts(unittest.TestCase):
    """
    Test certain ``_tub_portlocation`` error cases for ports setup.
    """
    def setUp(self):
        self.basedir = self.mktemp()
        create_node_dir(self.basedir, "testing")

    def test_listen_on_zero(self):
        """
        ``_tub_portlocation`` raises ``PortAssignmentRequired`` called with a
        listen address including port 0 and no interface.
        """
        config_data = (
            "[node]\n"
            "tub.port = tcp:0\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)
        with self.assertRaises(PortAssignmentRequired):
            _tub_portlocation(config, _stub_none, _stub_none)

    def test_listen_on_zero_with_host(self):
        """
        ``_tub_portlocation`` raises ``PortAssignmentRequired`` called with a
        listen address including port 0 and an interface.
        """
        config_data = (
            "[node]\n"
            "tub.port = tcp:0:interface=127.0.0.1\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)
        with self.assertRaises(PortAssignmentRequired):
            _tub_portlocation(config, _stub_none, _stub_none)

    def test_parsing_tcp(self):
        """
        When ``tub.port`` is given and ``tub.location`` is **AUTO** the port
        number from ``tub.port`` is used as the port number for the value
        constructed for ``tub.location``.
        """
        config_data = (
            "[node]\n"
            "tub.port = tcp:777\n"
            "tub.location = AUTO\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        tubport, tublocation = _tub_portlocation(
            config,
            _stub_get_local_addresses_sync,
            _stub_allocate_tcp_port,
        )
        self.assertEqual(tubport, "tcp:777")
        self.assertEqual(tublocation, b"tcp:LOCAL:777")

    def test_parsing_defaults(self):
        """
        parse empty config, check defaults
        """
        config_data = (
            "[node]\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        tubport, tublocation = _tub_portlocation(
            config,
            _stub_get_local_addresses_sync,
            _stub_allocate_tcp_port,
        )
        self.assertEqual(tubport, "tcp:999")
        self.assertEqual(tublocation, b"tcp:LOCAL:999")

    def test_parsing_location_complex(self):
        """
        location with two options (including defaults)
        """
        config_data = (
            "[node]\n"
            "tub.location = tcp:HOST:888,AUTO\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        tubport, tublocation = _tub_portlocation(
            config,
            _stub_get_local_addresses_sync,
            _stub_allocate_tcp_port,
        )
        self.assertEqual(tubport, "tcp:999")
        self.assertEqual(tublocation, b"tcp:HOST:888,tcp:LOCAL:999")

    def test_parsing_all_disabled(self):
        """
        parse config with both port + location disabled
        """
        config_data = (
            "[node]\n"
            "tub.port = disabled\n"
            "tub.location = disabled\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        res = _tub_portlocation(
            config,
            _stub_get_local_addresses_sync,
            _stub_allocate_tcp_port,
        )
        self.assertTrue(res is None)

    def test_empty_tub_port(self):
        """
        port povided, but empty is an error
        """
        config_data = (
            "[node]\n"
            "tub.port = \n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        with self.assertRaises(ValueError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )
        self.assertIn(
            "tub.port must not be empty",
            str(ctx.exception)
        )

    def test_empty_tub_location(self):
        """
        location povided, but empty is an error
        """
        config_data = (
            "[node]\n"
            "tub.location = \n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        with self.assertRaises(ValueError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )
        self.assertIn(
            "tub.location must not be empty",
            str(ctx.exception)
        )

    def test_disabled_port_not_tub(self):
        """
        error to disable port but not location
        """
        config_data = (
            "[node]\n"
            "tub.port = disabled\n"
            "tub.location = not_disabled\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        with self.assertRaises(ValueError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )
        self.assertIn(
            "tub.port is disabled, but not tub.location",
            str(ctx.exception)
        )

    def test_disabled_tub_not_port(self):
        """
        error to disable location but not port
        """
        config_data = (
            "[node]\n"
            "tub.port = not_disabled\n"
            "tub.location = disabled\n"
        )
        config = config_from_string(self.basedir, "portnum", config_data)

        with self.assertRaises(ValueError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )
        self.assertIn(
            "tub.location is disabled, but not tub.port",
            str(ctx.exception)
        )

    def test_tub_location_tcp(self):
        """
        If ``reveal-IP-address`` is set to false and ``tub.location`` includes a
        **tcp** hint then ``_tub_portlocation`` raises `PrivacyError`` because
        TCP leaks IP addresses.
        """
        config = config_from_string(
            "fake.port",
            "no-basedir",
            "[node]\nreveal-IP-address = false\ntub.location=tcp:hostname:1234\n",
        )
        with self.assertRaises(PrivacyError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )
        self.assertEqual(
            str(ctx.exception),
            "tub.location includes tcp: hint",
        )

    def test_tub_location_legacy_tcp(self):
        """
        If ``reveal-IP-address`` is set to false and ``tub.location`` includes a
        "legacy" hint with no explicit type (which means it is a **tcp** hint)
        then the behavior is the same as for an explicit **tcp** hint.
        """
        config = config_from_string(
            "fake.port",
            "no-basedir",
            "[node]\nreveal-IP-address = false\ntub.location=hostname:1234\n",
        )

        with self.assertRaises(PrivacyError) as ctx:
            _tub_portlocation(
                config,
                _stub_get_local_addresses_sync,
                _stub_allocate_tcp_port,
            )

        self.assertEqual(
            str(ctx.exception),
            "tub.location includes tcp: hint",
        )


BASE_CONFIG = """
[tor]
enabled = false
[i2p]
enabled = false
"""

NOLISTEN = """
[node]
tub.port = disabled
tub.location = disabled
"""

DISABLE_STORAGE = """
[storage]
enabled = false
"""

ENABLE_STORAGE = """
[storage]
enabled = true
"""

ENABLE_HELPER = """
[helper]
enabled = true
"""

class FakeTub:
    def __init__(self):
        self.tubID = base64.b32encode(b"foo")
        self.listening_ports = []
    def setOption(self, name, value): pass
    def removeAllConnectionHintHandlers(self): pass
    def addConnectionHintHandler(self, hint_type, handler): pass
    def listenOn(self, what):
        self.listening_ports.append(what)
    def setLocation(self, location): pass
    def setServiceParent(self, parent): pass

class Listeners(unittest.TestCase):

    # 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.
        """
        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)
        t = FakeTub()
        tub_listen_on(None, None, t, port, location)
        self.assertEqual(t.listening_ports,
                         ["tcp:%d:interface=127.0.0.1" % port1,
                          "tcp:%d:interface=127.0.0.1" % port2])

    def test_tor_i2p_listeners(self):
        """
        When configured to listen on an "i2p" or "tor" address, ``tub_listen_on``
        tells the Tub to listen on endpoints supplied by the given Tor and I2P
        providers.
        """
        t = FakeTub()

        i2p_listener = object()
        i2p_provider = ConstantAddresses(i2p_listener)
        tor_listener = object()
        tor_provider = ConstantAddresses(tor_listener)

        tub_listen_on(
            i2p_provider,
            tor_provider,
            t,
            "listen:i2p,listen:tor",
            "tcp:example.org:1234",
        )
        self.assertEqual(
            t.listening_ports,
            [i2p_listener, tor_listener],
        )


class ClientNotListening(unittest.TestCase):

    @defer.inlineCallbacks
    def test_disabled(self):
        basedir = "test_node/test_disabled"
        create_node_dir(basedir, "testing")
        f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
        f.write(BASE_CONFIG)
        f.write(NOLISTEN)
        f.write(DISABLE_STORAGE)
        f.close()
        n = yield client.create_client(basedir)
        self.assertEqual(n.tub.getListeners(), [])

    @defer.inlineCallbacks
    def test_disabled_but_storage(self):
        basedir = "test_node/test_disabled_but_storage"
        create_node_dir(basedir, "testing")
        f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
        f.write(BASE_CONFIG)
        f.write(NOLISTEN)
        f.write(ENABLE_STORAGE)
        f.close()
        with self.assertRaises(ValueError) as ctx:
            yield client.create_client(basedir)
        self.assertIn(
            "storage is enabled, but tub is not listening",
            str(ctx.exception),
        )

    @defer.inlineCallbacks
    def test_disabled_but_helper(self):
        basedir = "test_node/test_disabled_but_helper"
        create_node_dir(basedir, "testing")
        f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
        f.write(BASE_CONFIG)
        f.write(NOLISTEN)
        f.write(DISABLE_STORAGE)
        f.write(ENABLE_HELPER)
        f.close()
        with self.assertRaises(ValueError) as ctx:
            yield client.create_client(basedir)
        self.assertIn(
            "helper is enabled, but tub is not listening",
            str(ctx.exception),
        )

class IntroducerNotListening(unittest.TestCase):

    @defer.inlineCallbacks
    def test_port_none_introducer(self):
        basedir = "test_node/test_port_none_introducer"
        create_node_dir(basedir, "testing")
        with open(os.path.join(basedir, 'tahoe.cfg'), 'wt') as f:
            f.write("[node]\n")
            f.write("tub.port = disabled\n")
            f.write("tub.location = disabled\n")
        with self.assertRaises(ValueError) as ctx:
            yield create_introducer(basedir)
        self.assertIn(
            "we are Introducer, but tub is not listening",
            str(ctx.exception),
        )

class Configuration(unittest.TestCase):

    def setUp(self):
        self.basedir = self.mktemp()
        fileutil.make_dirs(self.basedir)

    def test_read_invalid_config(self):
        with open(os.path.join(self.basedir, 'tahoe.cfg'), 'w') as f:
            f.write(
                '[invalid section]\n'
                'foo = bar\n'
            )
        with self.assertRaises(UnknownConfigError) as ctx:
            read_config(
                self.basedir,
                "client.port",
            )

        self.assertIn(
            "invalid section",
            str(ctx.exception),
        )

    @defer.inlineCallbacks
    def test_create_client_invalid_config(self):
        with open(os.path.join(self.basedir, 'tahoe.cfg'), 'w') as f:
            f.write(
                '[invalid section]\n'
                'foo = bar\n'
            )
        with self.assertRaises(UnknownConfigError) as ctx:
            yield client.create_client(self.basedir)

        self.assertIn(
            "invalid section",
            str(ctx.exception),
        )



class CreateDefaultConnectionHandlersTests(unittest.TestCase):
    """
    Tests for create_default_connection_handlers().
    """

    def test_tcp_disabled(self):
        """
        If tcp is set to disabled, no TCP handler is set.
        """
        config = config_from_string("", "", dedent("""
        [connections]
        tcp = disabled
        """))
        default_handlers = create_default_connection_handlers(
            config,
            {},
        )
        self.assertIs(default_handlers["tcp"], None)