test_util_misc.py

# This file is part of Buildbot.  Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members

from parameterized import parameterized

from twisted.trial import unittest

from buildbot_worker.util import misc


class TestParsePortString(unittest.TestCase):

    @parameterized.expand([
        ('', ([''], {})),
        ('a', (['a'], {})),
        ('a:b', (['a', 'b'], {})),
        ('a:b=c', (['a'], {'b': 'c'})),
        ('a:b:c', (['a', 'b', 'c'], {})),
        ('a=', ([], {'a': ''})),
        ('a=b', ([], {'a': 'b'})),
        ('a=:b:d', (['b', 'd'], {'a': ''})),
        ('a=b:c:d:e', (['c', 'd', 'e'], {'a': 'b'})),
        ('a:b=c:d:e', (['a', 'd', 'e'], {'b': 'c'})),
        ('a:b:c:d=e', (['a', 'b', 'c'], {'d': 'e'})),
        ('a=b:c:d=e', (['c'], {'a': 'b', 'd': 'e'})),
        ('a:b=c:d=e', (['a'], {'b': 'c', 'd': 'e'})),
        ('a=b::c:d=e', (['', 'c'], {'a': 'b', 'd': 'e'})),
    ])
    def test_parse_port_string(self, input, expected):
        self.assertEqual(misc.parse_port_string(input), (expected))

    @parameterized.expand([
        ('='),
        ('=b'),
        ('a:=b'),
        ('a::=b'),
        ('a:=b=c'),
        ('a::==c:'),
        ('a==b'),
        ('a=b=c'),
        ('a:b==c'),
        ('a=b==c=d'),
        ('a===b'),
        ('a:=:b'),
        ('a:b:c===b'),
        ('a=b=c=d'),
    ])
    def test_parse_port_string_exception(self, input):
        with self.assertRaises(Exception):
            misc.parse_port_string(input)


class TestGetPort(unittest.TestCase):

    @parameterized.expand([
        (12345, (None, 12345)),
        ('12345', (None, 12345)),
        ('unix', (None, None)),
        ('port=12345', (None, 12345)),
        ('tcp:0', (None, 0)),
        ('tcp:12345', (None, 12345)),
        ('tls:12345', (None, 12345)),
        ('tls:tcp:12345', (None, 12345)),
        ('tls:tcp:my_host:12345', ('my_host', 12345)),
        ('0:interface=127.0.0.1', (None, 0)),
        ('my_host:12345', ('my_host', 12345)),
        ('0:interface=127.0.0.1', (None, 0)),
        ('12345:host=my_host', ('my_host', 12345)),
        ('host=my_host:port=12345', ('my_host', 12345)),
        ('tcp:0:interface=127.0.0.1', (None, 0)),
        ('tcp:interface=127.0.0.1:port=12345', (None, 12345)),
        ('tcp:12345:host=my_host', ('my_host', 12345)),
        ('tcp:12345:interface=127.0.0.1', (None, 12345)),
        ('tcp:host=my_host:port=12345', ('my_host', 12345)),
        ('tcp:my_host:12345', ('my_host', 12345))
        ])
    def test_get_host_port(self, input, expected):
        self.assertEqual(misc.get_host_port(input), (expected))

    @parameterized.expand([
        (''),
        ('my_host:unix'),
        ('my_host=test'),
        ('port=test'),
        ('port='),
        ('tls'),
        ('tcp'),
        ('tcp:my_host:port:test'),
        ('tls:tcp'),
        ('tls:tcp:abc:abc'),
        ('tls:tcp:a:b:c'),
        ('host:host'),
        ('host:host:host')
    ])
    def test_get_host_port_exception(self, input):
        with self.assertRaises(ValueError):
            misc.get_host_port(input)