|
@@ -10,7 +10,7 @@ from supervisor import datatypes
|
|
|
|
|
|
class DatatypesTest(unittest.TestCase):
|
|
|
def test_boolean_returns_true_for_truthy_values(self):
|
|
|
- for s in datatypes.TRUTHY_STRINGS:
|
|
|
+ for s in datatypes.TRUTHY_STRINGS:
|
|
|
actual = datatypes.boolean(s)
|
|
|
self.assertEqual(actual, True)
|
|
|
|
|
@@ -30,7 +30,7 @@ class DatatypesTest(unittest.TestCase):
|
|
|
self.assertEqual(actual, False)
|
|
|
|
|
|
def test_boolean_raises_value_error_for_bad_value(self):
|
|
|
- self.assertRaises(ValueError,
|
|
|
+ self.assertRaises(ValueError,
|
|
|
datatypes.boolean, 'not-a-value')
|
|
|
|
|
|
def test_list_of_strings_returns_empty_list_for_empty_string(self):
|
|
@@ -52,15 +52,15 @@ class DatatypesTest(unittest.TestCase):
|
|
|
def test_list_of_ints_returns_empty_list_for_empty_string(self):
|
|
|
actual = datatypes.list_of_ints('')
|
|
|
self.assertEqual(actual, [])
|
|
|
-
|
|
|
+
|
|
|
def test_list_of_ints_returns_list_of_ints_by_comma_split(self):
|
|
|
actual = datatypes.list_of_ints('1,42')
|
|
|
self.assertEqual(actual, [1,42])
|
|
|
-
|
|
|
+
|
|
|
def test_list_of_ints_returns_ints_even_if_whitespace_in_string(self):
|
|
|
actual = datatypes.list_of_ints(' 1 , 42 ')
|
|
|
self.assertEqual(actual, [1,42])
|
|
|
-
|
|
|
+
|
|
|
def test_list_of_ints_raises_value_error_when_comma_split_fails(self):
|
|
|
self.assertRaises(ValueError,
|
|
|
datatypes.list_of_ints, 42)
|
|
@@ -84,7 +84,7 @@ class DatatypesTest(unittest.TestCase):
|
|
|
def test_dict_of_key_value_pairs_returns_empty_dict_for_empty_str(self):
|
|
|
actual = datatypes.dict_of_key_value_pairs('')
|
|
|
self.assertEqual({}, actual)
|
|
|
-
|
|
|
+
|
|
|
def test_dict_of_key_value_pairs_returns_dict_from_single_pair_str(self):
|
|
|
actual = datatypes.dict_of_key_value_pairs('foo=bar')
|
|
|
expected = {'foo': 'bar'}
|
|
@@ -129,20 +129,20 @@ class DatatypesTest(unittest.TestCase):
|
|
|
self.assertEqual(actual, expected)
|
|
|
|
|
|
def test_dict_of_key_value_pairs_raises_value_error_on_too_short(self):
|
|
|
- self.assertRaises(ValueError,
|
|
|
+ self.assertRaises(ValueError,
|
|
|
datatypes.dict_of_key_value_pairs, 'foo')
|
|
|
- self.assertRaises(ValueError,
|
|
|
+ self.assertRaises(ValueError,
|
|
|
datatypes.dict_of_key_value_pairs, 'foo=')
|
|
|
- self.assertRaises(ValueError,
|
|
|
+ self.assertRaises(ValueError,
|
|
|
datatypes.dict_of_key_value_pairs, 'foo=bar,baz')
|
|
|
- self.assertRaises(ValueError,
|
|
|
+ self.assertRaises(ValueError,
|
|
|
datatypes.dict_of_key_value_pairs, 'foo=bar,baz=')
|
|
|
|
|
|
def test_logfile_name_returns_none_for_none_values(self):
|
|
|
for thing in datatypes.LOGFILE_NONES:
|
|
|
actual = datatypes.logfile_name(thing)
|
|
|
- self.assertEqual(actual, None)
|
|
|
-
|
|
|
+ self.assertEqual(actual, None)
|
|
|
+
|
|
|
def test_logfile_name_returns_none_for_uppered_none_values(self):
|
|
|
for thing in datatypes.LOGFILE_NONES:
|
|
|
if hasattr(thing, 'upper'):
|
|
@@ -153,14 +153,14 @@ class DatatypesTest(unittest.TestCase):
|
|
|
def test_logfile_name_returns_automatic_for_auto_values(self):
|
|
|
for thing in datatypes.LOGFILE_AUTOS:
|
|
|
actual = datatypes.logfile_name(thing)
|
|
|
- self.assertEqual(actual, datatypes.Automatic)
|
|
|
+ self.assertEqual(actual, datatypes.Automatic)
|
|
|
|
|
|
def test_logfile_name_returns_automatic_for_uppered_auto_values(self):
|
|
|
for thing in datatypes.LOGFILE_AUTOS:
|
|
|
if hasattr(thing, 'upper'):
|
|
|
thing = thing.upper()
|
|
|
actual = datatypes.logfile_name(thing)
|
|
|
- self.assertEqual(actual, datatypes.Automatic)
|
|
|
+ self.assertEqual(actual, datatypes.Automatic)
|
|
|
|
|
|
def test_logfile_name_returns_existing_dirpath_for_other_values(self):
|
|
|
func = datatypes.existing_dirpath
|
|
@@ -192,16 +192,16 @@ class DatatypesTest(unittest.TestCase):
|
|
|
|
|
|
def test_url_rejects_unix_scheme_with_no_slashes_or_path(self):
|
|
|
bad_url = "unix:"
|
|
|
- self.assertRaises(ValueError, datatypes.url, bad_url)
|
|
|
+ self.assertRaises(ValueError, datatypes.url, bad_url)
|
|
|
|
|
|
def test_url_rejects_unix_scheme_with_slashes_but_no_path(self):
|
|
|
bad_url = "unix://"
|
|
|
- self.assertRaises(ValueError, datatypes.url, bad_url)
|
|
|
-
|
|
|
+ self.assertRaises(ValueError, datatypes.url, bad_url)
|
|
|
+
|
|
|
class InetStreamSocketConfigTests(unittest.TestCase):
|
|
|
def _getTargetClass(self):
|
|
|
return datatypes.InetStreamSocketConfig
|
|
|
-
|
|
|
+
|
|
|
def _makeOne(self, *args, **kw):
|
|
|
return self._getTargetClass()(*args, **kw)
|
|
|
|
|
@@ -212,7 +212,7 @@ class InetStreamSocketConfigTests(unittest.TestCase):
|
|
|
def test___str__(self):
|
|
|
cfg = self._makeOne('localhost', 65531)
|
|
|
self.assertEqual(str(cfg), 'tcp://localhost:65531')
|
|
|
-
|
|
|
+
|
|
|
def test_repr(self):
|
|
|
conf = self._makeOne('127.0.0.1', 8675)
|
|
|
s = repr(conf)
|
|
@@ -229,7 +229,7 @@ class InetStreamSocketConfigTests(unittest.TestCase):
|
|
|
conf = self._makeOne('localhost', '5001')
|
|
|
addr = conf.addr()
|
|
|
self.assertEqual(addr, ('localhost', 5001))
|
|
|
-
|
|
|
+
|
|
|
def test_create_and_bind(self):
|
|
|
conf = self._makeOne('127.0.0.1', 8675)
|
|
|
sock = conf.create_and_bind()
|
|
@@ -237,7 +237,7 @@ class InetStreamSocketConfigTests(unittest.TestCase):
|
|
|
self.assertTrue(reuse)
|
|
|
self.assertEquals(conf.addr(), sock.getsockname()) #verifies that bind was called
|
|
|
sock.close()
|
|
|
-
|
|
|
+
|
|
|
def test_same_urls_are_equal(self):
|
|
|
conf1 = self._makeOne('localhost', 5001)
|
|
|
conf2 = self._makeOne('localhost', 5001)
|
|
@@ -254,23 +254,23 @@ class InetStreamSocketConfigTests(unittest.TestCase):
|
|
|
conf1 = self._makeOne('localhost', 5001)
|
|
|
conf2 = 'blah'
|
|
|
self.assertTrue(conf1 != conf2)
|
|
|
- self.assertFalse(conf1 == conf2)
|
|
|
-
|
|
|
+ self.assertFalse(conf1 == conf2)
|
|
|
+
|
|
|
class UnixStreamSocketConfigTests(unittest.TestCase):
|
|
|
def _getTargetClass(self):
|
|
|
return datatypes.UnixStreamSocketConfig
|
|
|
-
|
|
|
+
|
|
|
def _makeOne(self, *args, **kw):
|
|
|
return self._getTargetClass()(*args, **kw)
|
|
|
|
|
|
def test_url(self):
|
|
|
conf = self._makeOne('/tmp/foo.sock')
|
|
|
self.assertEqual(conf.url, 'unix:///tmp/foo.sock')
|
|
|
-
|
|
|
+
|
|
|
def test___str__(self):
|
|
|
cfg = self._makeOne('foo/bar')
|
|
|
self.assertEqual(str(cfg), 'unix://foo/bar')
|
|
|
-
|
|
|
+
|
|
|
def test_repr(self):
|
|
|
conf = self._makeOne('/tmp/foo.sock')
|
|
|
s = repr(conf)
|
|
@@ -282,13 +282,13 @@ class UnixStreamSocketConfigTests(unittest.TestCase):
|
|
|
conf = self._makeOne('/tmp/foo.sock')
|
|
|
addr = conf.addr()
|
|
|
self.assertEqual(addr, '/tmp/foo.sock')
|
|
|
-
|
|
|
+
|
|
|
def test_create_and_bind(self):
|
|
|
(tf_fd, tf_name) = tempfile.mkstemp()
|
|
|
owner = (sentinel.uid, sentinel.gid)
|
|
|
mode = sentinel.mode
|
|
|
conf = self._makeOne(tf_name, owner=owner, mode=mode)
|
|
|
-
|
|
|
+
|
|
|
#Patch os.chmod and os.chown functions with mocks
|
|
|
#objects so that the test does not depend on
|
|
|
#any specific system users or permissions
|
|
@@ -298,7 +298,7 @@ class UnixStreamSocketConfigTests(unittest.TestCase):
|
|
|
@patch('os.chmod', chmod_mock)
|
|
|
def call_create_and_bind(conf):
|
|
|
return conf.create_and_bind()
|
|
|
-
|
|
|
+
|
|
|
sock = call_create_and_bind(conf)
|
|
|
self.assertTrue(os.path.exists(tf_name))
|
|
|
self.assertEquals(conf.addr(), sock.getsockname()) #verifies that bind was called
|
|
@@ -319,25 +319,25 @@ class UnixStreamSocketConfigTests(unittest.TestCase):
|
|
|
mode_arg = chmod_mock.call_args[0][1]
|
|
|
self.assertEquals(tf_name, path_arg)
|
|
|
self.assertEquals(mode, mode_arg)
|
|
|
-
|
|
|
+
|
|
|
def test_same_paths_are_equal(self):
|
|
|
conf1 = self._makeOne('/tmp/foo.sock')
|
|
|
conf2 = self._makeOne('/tmp/foo.sock')
|
|
|
self.assertTrue(conf1 == conf2)
|
|
|
self.assertFalse(conf1 != conf2)
|
|
|
-
|
|
|
+
|
|
|
def test_diff_paths_are_not_equal(self):
|
|
|
conf1 = self._makeOne('/tmp/foo.sock')
|
|
|
conf2 = self._makeOne('/tmp/bar.sock')
|
|
|
self.assertTrue(conf1 != conf2)
|
|
|
self.assertFalse(conf1 == conf2)
|
|
|
-
|
|
|
+
|
|
|
def test_diff_objs_are_not_equal(self):
|
|
|
conf1 = self._makeOne('/tmp/foo.sock')
|
|
|
conf2 = 'blah'
|
|
|
self.assertTrue(conf1 != conf2)
|
|
|
- self.assertFalse(conf1 == conf2)
|
|
|
-
|
|
|
+ self.assertFalse(conf1 == conf2)
|
|
|
+
|
|
|
class RangeCheckedConversionTests(unittest.TestCase):
|
|
|
def _getTargetClass(self):
|
|
|
from supervisor.datatypes import RangeCheckedConversion
|
|
@@ -349,11 +349,11 @@ class RangeCheckedConversionTests(unittest.TestCase):
|
|
|
def test_below_lower_bound(self):
|
|
|
conversion = self._makeOne(lambda *arg: -1, 0)
|
|
|
self.assertRaises(ValueError, conversion, None)
|
|
|
-
|
|
|
+
|
|
|
def test_above_upper_lower_bound(self):
|
|
|
conversion = self._makeOne(lambda *arg: 1, 0, 0)
|
|
|
self.assertRaises(ValueError, conversion, None)
|
|
|
-
|
|
|
+
|
|
|
def test_passes(self):
|
|
|
conversion = self._makeOne(lambda *arg: 0, 0, 0)
|
|
|
self.assertEqual(conversion(0), 0)
|
|
@@ -373,7 +373,7 @@ class InetAddressTests(unittest.TestCase):
|
|
|
host, port = self._callFUT('*:8080')
|
|
|
self.assertEqual(host, '')
|
|
|
self.assertEqual(port, 8080)
|
|
|
-
|
|
|
+
|
|
|
def test_boring(self):
|
|
|
host, port = self._callFUT('localhost:80')
|
|
|
self.assertEqual(host, 'localhost')
|
|
@@ -392,13 +392,13 @@ class TestSocketAddress(unittest.TestCase):
|
|
|
addr = self._makeOne('/foo/bar')
|
|
|
self.assertEqual(addr.family, socket.AF_UNIX)
|
|
|
self.assertEqual(addr.address, '/foo/bar')
|
|
|
-
|
|
|
+
|
|
|
def test_inet_socket(self):
|
|
|
import socket
|
|
|
addr = self._makeOne('localhost:8080')
|
|
|
self.assertEqual(addr.family, socket.AF_INET)
|
|
|
self.assertEqual(addr.address, ('localhost', 8080))
|
|
|
-
|
|
|
+
|
|
|
class TestColonSeparatedUserGroup(unittest.TestCase):
|
|
|
def _callFUT(self, arg):
|
|
|
from supervisor.datatypes import colon_separated_user_group
|
|
@@ -410,11 +410,11 @@ class TestColonSeparatedUserGroup(unittest.TestCase):
|
|
|
def test_missinguser_username(self):
|
|
|
self.assertRaises(ValueError,
|
|
|
self._callFUT, 'godihopethisuserdoesntexist')
|
|
|
-
|
|
|
+
|
|
|
def test_missinguser_username_and_groupname(self):
|
|
|
self.assertRaises(ValueError,
|
|
|
self._callFUT, 'godihopethisuserdoesntexist:foo')
|
|
|
-
|
|
|
+
|
|
|
class TestOctalType(unittest.TestCase):
|
|
|
def _callFUT(self, arg):
|
|
|
from supervisor.datatypes import octal_type
|
|
@@ -425,8 +425,3 @@ class TestOctalType(unittest.TestCase):
|
|
|
|
|
|
def test_test_it_failure(self):
|
|
|
self.assertRaises(ValueError, self._callFUT, 'noo')
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|