summaryrefslogtreecommitdiff
path: root/tests/unit
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unit')
-rw-r--r--tests/unit/__init__.py0
-rw-r--r--tests/unit/cache/__init__.py0
-rw-r--r--tests/unit/cache/base.py149
-rw-r--r--tests/unit/cache/test_cache.py38
-rw-r--r--tests/unit/cache/test_memory.py114
-rw-r--r--tests/unit/crypto/__init__.py0
-rw-r--r--tests/unit/crypto/test_utils.py189
-rw-r--r--tests/unit/fake/__init__.py23
-rw-r--r--tests/unit/fakes.py72
-rw-r--r--tests/unit/fixture/__init__.py0
-rw-r--r--tests/unit/scheduler/__init__.py0
-rw-r--r--tests/unit/scheduler/fake_hosts.py53
-rw-r--r--tests/unit/scheduler/test_base_filter.py168
-rw-r--r--tests/unit/scheduler/test_host_filters.py704
-rw-r--r--tests/unit/scheduler/test_weights.py65
-rw-r--r--tests/unit/test_cliutils.py699
-rw-r--r--tests/unit/test_imageutils.py199
-rw-r--r--tests/unit/test_memorycache.py55
18 files changed, 0 insertions, 2528 deletions
diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/tests/unit/__init__.py
+++ /dev/null
diff --git a/tests/unit/cache/__init__.py b/tests/unit/cache/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/tests/unit/cache/__init__.py
+++ /dev/null
diff --git a/tests/unit/cache/base.py b/tests/unit/cache/base.py
deleted file mode 100644
index 752bccd2..00000000
--- a/tests/unit/cache/base.py
+++ /dev/null
@@ -1,149 +0,0 @@
-# Copyright 2013 Red Hat, Inc.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import time
-
-from oslotest import base
-
-from openstack.common.cache import cache
-
-
-class CacheBaseTest(base.BaseTestCase):
-
- cache_url = None
-
- def setUp(self):
- super(CacheBaseTest, self).setUp()
- self.client = cache.get_cache(self.cache_url)
-
- def tearDown(self):
- self.client.clear()
- super(CacheBaseTest, self).tearDown()
-
- def test_set_get(self):
- self.client['foo'] = 'bar'
- self.assertEqual(self.client['foo'], 'bar')
-
- def test_get_keyerror(self):
- self.assertRaises(KeyError,
- self.client.__getitem__,
- "DoesNotExist")
-
- def test_set_not_exists_get(self):
- self.client.set('foo', 'bar', 10, not_exists=True)
- self.assertEqual(self.client.get('foo'), 'bar')
-
- def test_set_not_exists_false_get(self):
- self.client['foo'] = 'bar'
- ret = self.client.set('foo', 'baz',
- 0, not_exists=True)
- self.assertFalse(ret)
- self.assertEqual(self.client.get('foo'), 'bar')
-
- def test_set_unset(self):
- self.client['foo'] = 'bar'
- self.assertEqual(self.client['foo'], 'bar')
-
- del self.client['foo']
- self.assertIsNone(self.client.get('foo'))
-
- def test_incr(self):
- self.client['foo'] = 1
- self.assertEqual(self.client['foo'], 1)
-
- self.client.incr('foo', 2)
- self.assertEqual(self.client['foo'], 3)
-
- self.client.incr('foo', -3)
- self.assertEqual(self.client['foo'], 0)
-
- self.client.incr('foo', -3)
- self.assertEqual(self.client['foo'], -3)
-
- def test_append(self):
- self.client['foo'] = [1, 2]
- self.assertEqual(self.client['foo'], [1, 2])
-
- self.client.append('foo', 3)
- self.assertEqual(self.client['foo'], [1, 2, 3])
-
- self.client.append('foo', 4)
- self.assertEqual(self.client['foo'], [1, 2, 3, 4])
-
- def test_append_tail(self):
- self.client['foo'] = [1, 2]
- self.assertEqual(self.client['foo'], [1, 2])
-
- self.client.append_tail('foo', [3, 4])
- self.assertEqual(self.client['foo'], [1, 2, 3, 4])
-
- def test_set_many(self):
- self.client.set_many(dict(foo=0, bar=1))
- self.assertEqual(self.client['foo'], 0)
- self.assertEqual(self.client['bar'], 1)
-
- def test_unset_many(self):
- self.client['foo'] = 0
- self.client['bar'] = 1
- self.assertEqual(self.client['foo'], 0)
- self.assertEqual(self.client['bar'], 1)
-
- self.client.unset_many(['foo', 'bar'])
- self.assertIsNone(self.client.get('foo'))
- self.assertIsNone(self.client.get('bar'))
-
- def test_get_many(self):
- self.client['foo'] = 0
- self.client['bar'] = 1
- values = self.client.get_many(["foo", "bar"])
- self.assertEqual(list(values), [('foo', 0), ('bar', 1)])
-
- def test_timeout(self):
- self.client.set('foo', 'bar', ttl=1)
- self.assertEqual(self.client.get('foo'), 'bar')
-
- # NOTE(flaper87): It's not funny
- # to sleep tests but this test is
- # supposed to work for all backends.
- time.sleep(1)
- self.assertIsNone(self.client['foo'])
-
- def test_clear(self):
- self.client['foo'] = 0
- self.client['bar'] = 1
-
- self.client.clear()
-
- self.assertIsNone(self.client.get('foo'))
- self.assertIsNone(self.client.get('bar'))
-
- def test_exists(self):
- self.client['foo'] = 'bar'
- self.assertTrue('foo' in self.client)
-
- del self.client['foo']
- self.assertFalse('foo' in self.client)
-
- def test_update(self):
- self.client.update(foo='bar', bar='foo')
- self.assertEqual(self.client.get('foo'), 'bar')
- self.assertEqual(self.client.get('bar'), 'foo')
-
- def test_setdefault(self):
- ret = self.client.setdefault("foo", "bar")
- self.assertEqual(ret, "bar")
-
- ret = self.client.setdefault("foo", "baaaar")
- self.assertEqual(ret, "bar")
diff --git a/tests/unit/cache/test_cache.py b/tests/unit/cache/test_cache.py
deleted file mode 100644
index c962c6bb..00000000
--- a/tests/unit/cache/test_cache.py
+++ /dev/null
@@ -1,38 +0,0 @@
-# Copyright 2013 Red Hat, Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-from oslo_config import cfg
-from oslotest import base
-
-from openstack.common.cache._backends import memory
-from openstack.common.cache import cache
-
-
-class TestCacheModule(base.BaseTestCase):
-
- def test_oslo_config(self):
- conf = cfg.ConfigOpts()
- cache.register_oslo_configs(conf)
- driver = cache.get_cache(conf.cache_url)
- self.assertIsInstance(driver, memory.MemoryBackend)
-
- def test_get_cache(self):
- driver = cache.get_cache()
- self.assertIsInstance(driver, memory.MemoryBackend)
-
- def test_get_cache_options(self):
- driver = cache.get_cache('memory://?default_ttl=4')
- self.assertIsInstance(driver, memory.MemoryBackend)
- self.assertEqual(driver._options, {'default_ttl': '4'})
- self.assertEqual(driver._default_ttl, 4)
diff --git a/tests/unit/cache/test_memory.py b/tests/unit/cache/test_memory.py
deleted file mode 100644
index b50d4114..00000000
--- a/tests/unit/cache/test_memory.py
+++ /dev/null
@@ -1,114 +0,0 @@
-# Copyright 2013 Red Hat, Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import time
-
-import mock
-
-from tests.unit.cache import base
-
-
-class MemorycacheTest(base.CacheBaseTest):
- """Test memory backend
-
- Since it is the default driver, nothing
- has to be done here.
- """
-
- cache_url = 'memory://'
-
- def test_timeout(self):
- now = time.time()
- with mock.patch('time.time') as time_mock:
- time_mock.return_value = now
- self.client.set('foo', 'bar', ttl=3)
- time_mock.return_value = now + 1
- self.assertEqual(self.client.get('foo'), 'bar')
- time_mock.return_value = now + 3
- self.assertIsNone(self.client.get('foo'))
-
- def test_timeout_unset(self):
- now = time.time()
- with mock.patch('time.time') as time_mock:
- time_mock.return_value = now
- self.client.set('foo', 'bar', ttl=3)
- self.client.set('fooo', 'bar', ttl=4)
- self.client.set('foooo', 'bar', ttl=5)
- self.client.set('fooooo', 'bar', ttl=6)
- time_mock.return_value = now + 1
- self.assertEqual(self.client.get('foo'), 'bar')
- self.assertEqual(self.client.get('fooo'), 'bar')
- self.assertEqual(self.client.get('foooo'), 'bar')
- self.assertEqual(self.client.get('fooooo'), 'bar')
-
- time_mock.return_value = now + 5
- del self.client['foo']
- self.assertIsNone(self.client.get('foo'))
- self.assertIsNone(self.client.get('fooo'))
- self.assertIsNone(self.client.get('foooo'))
- self.assertEqual(self.client.get('fooooo'), 'bar')
-
- def test_timeout_unset_pop(self):
- now = time.time()
- with mock.patch('time.time') as time_mock:
- time_mock.return_value = now
- self.client.set('foo', 'bar', ttl=3)
- self.client.set('fooo', 'bar', ttl=4)
- self.client.set('foooo', 'bar', ttl=5)
- self.client.set('fooooo', 'bar', ttl=6)
- time_mock.return_value = now + 1
- self.assertEqual(self.client.get('foo'), 'bar')
- self.assertEqual(self.client.get('fooo'), 'bar')
- self.assertEqual(self.client.get('foooo'), 'bar')
- self.assertEqual(self.client.get('fooooo'), 'bar')
-
- time_mock.return_value = now + 4
-
- # NOTE(flaper87): Let unset delete foooo and timeout
- # expire foo and fooo.
- del self.client['foooo']
- self.assertIsNone(self.client.get('foo'))
- self.assertIsNone(self.client.get('fooo'))
- self.assertIsNone(self.client.get('foooo'))
- self.assertEqual(self.client.get('fooooo'), 'bar')
-
- def test_unset_keys_expires(self):
- # NOTE(kgriffs): The only way to verify the
- # side-effects of deleting a cache entry is
- # to use a white-box test. This test was
- # added to verify a fix for a bug that was
- # preventing keys from being removed from
- # _keys_expires when the value for the
- # key was deleted.
-
- # NOTE(kgriffs): _keys_expires is only used
- # to track entries with a non-zero TTL.
- ttl = 5
-
- now = int(time.time())
- expiration = now + ttl
-
- with mock.patch('time.time') as time_mock:
- time_mock.return_value = now
- self.client.set('foo', 'bar', ttl=ttl)
-
- expiration = now + 5
-
- keyset = self.client._keys_expires[expiration]
- self.assertEqual(len(keyset), 1)
-
- del self.client['foo']
-
- keyset = self.client._keys_expires[expiration]
- self.assertEqual(len(keyset), 0)
diff --git a/tests/unit/crypto/__init__.py b/tests/unit/crypto/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/tests/unit/crypto/__init__.py
+++ /dev/null
diff --git a/tests/unit/crypto/test_utils.py b/tests/unit/crypto/test_utils.py
deleted file mode 100644
index c4b98fbd..00000000
--- a/tests/unit/crypto/test_utils.py
+++ /dev/null
@@ -1,189 +0,0 @@
-# Copyright 2013 Red Hat, Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-"""
-Unit Tests for crypto utils.
-"""
-
-from oslotest import base as test_base
-import six
-
-from openstack.common.crypto import utils as cryptoutils
-
-bchr = six.int2byte
-
-
-class CryptoUtilsTestCase(test_base.BaseTestCase):
-
- # Uses Tests from RFC5869
- def _test_HKDF(self, ikm, prk, okm, length,
- salt=None, info=b'', hashtype='SHA256'):
- hkdf = cryptoutils.HKDF(hashtype=hashtype)
-
- tprk = hkdf.extract(ikm, salt=salt)
- self.assertEqual(prk, tprk)
-
- tokm = hkdf.expand(prk, info, length)
- self.assertEqual(okm, tokm)
-
- def test_HKDF_1(self):
- ikm = b'\x0b' * 22
- salt = b''.join(map(lambda x: bchr(x), range(0x00, 0x0d)))
- info = b''.join(map(lambda x: bchr(x), range(0xf0, 0xfa)))
- length = 42
-
- prk = (b'\x07\x77\x09\x36\x2c\x2e\x32\xdf\x0d\xdc\x3f\x0d\xc4\x7b'
- b'\xba\x63\x90\xb6\xc7\x3b\xb5\x0f\x9c\x31\x22\xec\x84\x4a'
- b'\xd7\xc2\xb3\xe5')
-
- okm = (b'\x3c\xb2\x5f\x25\xfa\xac\xd5\x7a\x90\x43\x4f\x64\xd0\x36'
- b'\x2f\x2a\x2d\x2d\x0a\x90\xcf\x1a\x5a\x4c\x5d\xb0\x2d\x56'
- b'\xec\xc4\xc5\xbf\x34\x00\x72\x08\xd5\xb8\x87\x18\x58\x65')
-
- self._test_HKDF(ikm, prk, okm, length, salt, info)
-
- def test_HKDF_2(self):
- ikm = b''.join(map(lambda x: bchr(x), range(0x00, 0x50)))
- salt = b''.join(map(lambda x: bchr(x), range(0x60, 0xb0)))
- info = b''.join(map(lambda x: bchr(x), range(0xb0, 0x100)))
- length = 82
-
- prk = (b'\x06\xa6\xb8\x8c\x58\x53\x36\x1a\x06\x10\x4c\x9c\xeb\x35'
- b'\xb4\x5c\xef\x76\x00\x14\x90\x46\x71\x01\x4a\x19\x3f\x40'
- b'\xc1\x5f\xc2\x44')
-
- okm = (b'\xb1\x1e\x39\x8d\xc8\x03\x27\xa1\xc8\xe7\xf7\x8c\x59\x6a'
- b'\x49\x34\x4f\x01\x2e\xda\x2d\x4e\xfa\xd8\xa0\x50\xcc\x4c'
- b'\x19\xaf\xa9\x7c\x59\x04\x5a\x99\xca\xc7\x82\x72\x71\xcb'
- b'\x41\xc6\x5e\x59\x0e\x09\xda\x32\x75\x60\x0c\x2f\x09\xb8'
- b'\x36\x77\x93\xa9\xac\xa3\xdb\x71\xcc\x30\xc5\x81\x79\xec'
- b'\x3e\x87\xc1\x4c\x01\xd5\xc1\xf3\x43\x4f\x1d\x87')
-
- self._test_HKDF(ikm, prk, okm, length, salt, info)
-
- def test_HKDF_3(self):
- ikm = b'\x0b' * 22
- length = 42
-
- prk = (b'\x19\xef\x24\xa3\x2c\x71\x7b\x16\x7f\x33\xa9\x1d\x6f\x64'
- b'\x8b\xdf\x96\x59\x67\x76\xaf\xdb\x63\x77\xac\x43\x4c\x1c'
- b'\x29\x3c\xcb\x04')
-
- okm = (b'\x8d\xa4\xe7\x75\xa5\x63\xc1\x8f\x71\x5f\x80\x2a\x06\x3c'
- b'\x5a\x31\xb8\xa1\x1f\x5c\x5e\xe1\x87\x9e\xc3\x45\x4e\x5f'
- b'\x3c\x73\x8d\x2d\x9d\x20\x13\x95\xfa\xa4\xb6\x1a\x96\xc8')
-
- self._test_HKDF(ikm, prk, okm, length)
-
- def test_HKDF_4(self):
- ikm = b'\x0b' * 11
- salt = b''.join(map(lambda x: bchr(x), range(0x00, 0x0d)))
- info = b''.join(map(lambda x: bchr(x), range(0xf0, 0xfa)))
- length = 42
-
- prk = (b'\x9b\x6c\x18\xc4\x32\xa7\xbf\x8f\x0e\x71\xc8\xeb\x88\xf4'
- b'\xb3\x0b\xaa\x2b\xa2\x43')
-
- okm = (b'\x08\x5a\x01\xea\x1b\x10\xf3\x69\x33\x06\x8b\x56\xef\xa5'
- b'\xad\x81\xa4\xf1\x4b\x82\x2f\x5b\x09\x15\x68\xa9\xcd\xd4'
- b'\xf1\x55\xfd\xa2\xc2\x2e\x42\x24\x78\xd3\x05\xf3\xf8\x96')
-
- self._test_HKDF(ikm, prk, okm, length, salt, info, hashtype='SHA')
-
- def test_HKDF_5(self):
- ikm = b''.join(map(lambda x: bchr(x), range(0x00, 0x50)))
- salt = b''.join(map(lambda x: bchr(x), range(0x60, 0xb0)))
- info = b''.join(map(lambda x: bchr(x), range(0xb0, 0x100)))
- length = 82
-
- prk = (b'\x8a\xda\xe0\x9a\x2a\x30\x70\x59\x47\x8d\x30\x9b\x26\xc4'
- b'\x11\x5a\x22\x4c\xfa\xf6')
-
- okm = (b'\x0b\xd7\x70\xa7\x4d\x11\x60\xf7\xc9\xf1\x2c\xd5\x91\x2a'
- b'\x06\xeb\xff\x6a\xdc\xae\x89\x9d\x92\x19\x1f\xe4\x30\x56'
- b'\x73\xba\x2f\xfe\x8f\xa3\xf1\xa4\xe5\xad\x79\xf3\xf3\x34'
- b'\xb3\xb2\x02\xb2\x17\x3c\x48\x6e\xa3\x7c\xe3\xd3\x97\xed'
- b'\x03\x4c\x7f\x9d\xfe\xb1\x5c\x5e\x92\x73\x36\xd0\x44\x1f'
- b'\x4c\x43\x00\xe2\xcf\xf0\xd0\x90\x0b\x52\xd3\xb4')
-
- self._test_HKDF(ikm, prk, okm, length, salt, info, hashtype='SHA')
-
- def test_HKDF_6(self):
- ikm = b'\x0b' * 22
- length = 42
-
- prk = (b'\xda\x8c\x8a\x73\xc7\xfa\x77\x28\x8e\xc6\xf5\xe7\xc2\x97'
- b'\x78\x6a\xa0\xd3\x2d\x01')
-
- okm = (b'\x0a\xc1\xaf\x70\x02\xb3\xd7\x61\xd1\xe5\x52\x98\xda\x9d'
- b'\x05\x06\xb9\xae\x52\x05\x72\x20\xa3\x06\xe0\x7b\x6b\x87'
- b'\xe8\xdf\x21\xd0\xea\x00\x03\x3d\xe0\x39\x84\xd3\x49\x18')
-
- self._test_HKDF(ikm, prk, okm, length, hashtype='SHA')
-
- def test_HKDF_7(self):
- ikm = b'\x0c' * 22
- length = 42
-
- prk = (b'\x2a\xdc\xca\xda\x18\x77\x9e\x7c\x20\x77\xad\x2e\xb1\x9d'
- b'\x3f\x3e\x73\x13\x85\xdd')
-
- okm = (b'\x2c\x91\x11\x72\x04\xd7\x45\xf3\x50\x0d\x63\x6a\x62\xf6'
- b'\x4f\x0a\xb3\xba\xe5\x48\xaa\x53\xd4\x23\xb0\xd1\xf2\x7e'
- b'\xbb\xa6\xf5\xe5\x67\x3a\x08\x1d\x70\xcc\xe7\xac\xfc\x48')
-
- self._test_HKDF(ikm, prk, okm, length, hashtype='SHA')
-
- def test_HKDF_8(self):
- ikm = b'\x0b' * 22
- prk = (b'\x19\xef\x24\xa3\x2c\x71\x7b\x16\x7f\x33\xa9\x1d\x6f\x64'
- b'\x8b\xdf\x96\x59\x67\x76\xaf\xdb\x63\x77\xac\x43\x4c\x1c'
- b'\x29\x3c\xcb\x04')
-
- # Just testing HKDFOutputLengthTooLong is returned
- try:
- self._test_HKDF(ikm, prk, None, 1000000)
- except cryptoutils.HKDFOutputLengthTooLong:
- pass
-
- def test_SymmetricCrypto_encrypt_string(self):
- msg = b'Plain Text'
-
- skc = cryptoutils.SymmetricCrypto()
- key = skc.new_key(16)
- cipher = skc.encrypt(key, msg)
- plain = skc.decrypt(key, cipher)
- self.assertEqual(msg, plain)
-
- def test_SymmetricCrypto_encrypt_blocks(self):
- cb = 16
- et = 'AES'
-
- skc = cryptoutils.SymmetricCrypto(enctype=et)
- key = skc.new_key(16)
- msg = skc.new_key(cb * 2)
-
- for i in range(cb * 2):
- cipher = skc.encrypt(key, msg[0:i], b64encode=False)
- plain = skc.decrypt(key, cipher, b64decode=False)
- self.assertEqual(msg[0:i], plain)
-
- def test_SymmetricCrypto_signing(self):
- msg = b'Authenticated Message'
- signature = b'KWjl6i30RMjc5PjnaccRwTPKTRCWM6sPpmGS2bxm5fQ='
- skey = b'L\xdd0\xf3\xb4\xc6\xe2p\xef\xc7\xbd\xaa\xc9eNC'
-
- skc = cryptoutils.SymmetricCrypto()
- validate = skc.sign(skey, msg)
- self.assertEqual(signature, validate)
diff --git a/tests/unit/fake/__init__.py b/tests/unit/fake/__init__.py
deleted file mode 100644
index d04431e3..00000000
--- a/tests/unit/fake/__init__.py
+++ /dev/null
@@ -1,23 +0,0 @@
-# Copyright 2012 IBM Corp.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-
-class FakeDriver(object):
- def __init__(self, first_arg=True):
- self.first_arg = first_arg
-
-
-class FakeDriver2(object):
- def __init__(self, first_arg):
- self.first_arg = first_arg
diff --git a/tests/unit/fakes.py b/tests/unit/fakes.py
deleted file mode 100644
index eab0d888..00000000
--- a/tests/unit/fakes.py
+++ /dev/null
@@ -1,72 +0,0 @@
-# Copyright 2012 Intel Inc, OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-"""
-Fakes For filter and weight tests.
-"""
-
-import gettext
-
-from openstack.common.scheduler import weights
-
-
-class FakeWeigher1(weights.BaseHostWeigher):
- def __init__(self):
- pass
-
-
-class FakeWeigher2(weights.BaseHostWeigher):
- def __init__(self):
- pass
-
-
-class FakeClass(object):
- def __init__(self):
- pass
-
-
-class FakeTranslations(gettext.GNUTranslations):
- """A test GNUTranslations class that takes a map of msg -> translations."""
-
- def __init__(self, translations):
- self.translations = translations
-
- # used by Python 3
- def gettext(self, msgid):
- return self.translations.get(msgid, msgid)
-
- # used by Python 2
- def ugettext(self, msgid):
- return self.translations.get(msgid, msgid)
-
- @staticmethod
- def translator(locales_map):
- """Returns a mock gettext.translation function that uses
- individual TestTranslations to translate in the given locales.
-
- :param locales_map: A map from locale name to a translations map.
- {
- 'es': {'Hi': 'Hola', 'Bye': 'Adios'},
- 'zh': {'Hi': 'Ni Hao', 'Bye': 'Zaijian'}
- }
- """
- def _translation(domain, localedir=None,
- languages=None, fallback=None):
- if languages:
- language = languages[0]
- if language in locales_map:
- return FakeTranslations(locales_map[language])
- return gettext.NullTranslations()
- return _translation
diff --git a/tests/unit/fixture/__init__.py b/tests/unit/fixture/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/tests/unit/fixture/__init__.py
+++ /dev/null
diff --git a/tests/unit/scheduler/__init__.py b/tests/unit/scheduler/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/tests/unit/scheduler/__init__.py
+++ /dev/null
diff --git a/tests/unit/scheduler/fake_hosts.py b/tests/unit/scheduler/fake_hosts.py
deleted file mode 100644
index 8cc988f0..00000000
--- a/tests/unit/scheduler/fake_hosts.py
+++ /dev/null
@@ -1,53 +0,0 @@
-# Copyright 2012 Intel Inc, OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-"""
-Fakes For filters tests.
-"""
-
-import six
-
-
-class FakeHostManager(object):
- """Defines fake hosts.
-
- host1: free_ram_mb=1024-512-512=0, free_disk_gb=1024-512-512=0
- host2: free_ram_mb=2048-512=1536 free_disk_gb=2048-512=1536
- host3: free_ram_mb=4096-1024=3072 free_disk_gb=4096-1024=3072
- host4: free_ram_mb=8192 free_disk_gb=8192
- """
-
- def __init__(self):
- self.service_states = {
- 'host1': {
- 'compute': {'host_memory_free': 1073741824},
- },
- 'host2': {
- 'compute': {'host_memory_free': 2147483648},
- },
- 'host3': {
- 'compute': {'host_memory_free': 3221225472},
- },
- 'host4': {
- 'compute': {'host_memory_free': 999999999},
- },
- }
-
-
-class FakeHostState(object):
- def __init__(self, host, attribute_dict):
- self.host = host
- for (key, val) in six.iteritems(attribute_dict):
- setattr(self, key, val)
diff --git a/tests/unit/scheduler/test_base_filter.py b/tests/unit/scheduler/test_base_filter.py
deleted file mode 100644
index a47b0db4..00000000
--- a/tests/unit/scheduler/test_base_filter.py
+++ /dev/null
@@ -1,168 +0,0 @@
-# Copyright (c) 2013 OpenStack Foundation.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import mock
-from oslotest import base as test_base
-from oslotest import moxstubout
-
-from openstack.common.scheduler import base_filter
-
-
-class TestBaseFilter(test_base.BaseTestCase):
-
- def setUp(self):
- super(TestBaseFilter, self).setUp()
- self.mox = self.useFixture(moxstubout.MoxStubout()).mox
- self.filter = base_filter.BaseFilter()
-
- def test_filter_one_is_called(self):
- filters = [1, 2, 3, 4]
- filter_properties = {'x': 'y'}
- self.mox.StubOutWithMock(self.filter, '_filter_one')
-
- self.filter._filter_one(1, filter_properties).AndReturn(False)
- self.filter._filter_one(2, filter_properties).AndReturn(True)
- self.filter._filter_one(3, filter_properties).AndReturn(True)
- self.filter._filter_one(4, filter_properties).AndReturn(False)
-
- self.mox.ReplayAll()
-
- result = list(self.filter.filter_all(filters, filter_properties))
- self.assertEqual([2, 3], result)
-
-
-class FakeExtension(object):
-
- def __init__(self, plugin):
- self.plugin = plugin
-
-
-class BaseFakeFilter(base_filter.BaseFilter):
- pass
-
-
-class FakeFilter1(BaseFakeFilter):
- """Derives from BaseFakeFilter and has a fake entry point defined.
-
- Entry point is returned by fake ExtensionManager.
- Should be included in the output of all_classes.
- """
- pass
-
-
-class FakeFilter2(BaseFakeFilter):
- """Derives from BaseFakeFilter but has no entry point.
-
- Should be not included in all_classes.
- """
- pass
-
-
-class FakeFilter3(base_filter.BaseFilter):
- """Does not derive from BaseFakeFilter.
-
- Should not be included.
- """
- pass
-
-
-class FakeFilter4(BaseFakeFilter):
- """Derives from BaseFakeFilter and has an entry point.
-
- Should be included.
- """
- pass
-
-
-class FakeFilter5(BaseFakeFilter):
- """Derives from BaseFakeFilter but has no entry point.
-
- Should not be included.
- """
- run_filter_once_per_request = True
- pass
-
-
-class FakeExtensionManager(list):
-
- def __init__(self, namespace):
- classes = [FakeFilter1, FakeFilter3, FakeFilter4]
- exts = map(FakeExtension, classes)
- super(FakeExtensionManager, self).__init__(exts)
- self.namespace = namespace
-
-
-class TestBaseFilterHandler(test_base.BaseTestCase):
-
- def setUp(self):
- super(TestBaseFilterHandler, self).setUp()
- self.stubs = self.useFixture(moxstubout.MoxStubout()).stubs
- self.stubs.Set(base_filter.base_handler.extension, 'ExtensionManager',
- FakeExtensionManager)
- self.handler = base_filter.BaseFilterHandler(BaseFakeFilter,
- 'fake_filters')
-
- def test_get_all_classes(self):
- # In order for a FakeFilter to be returned by get_all_classes, it has
- # to comply with these rules:
- # * It must be derived from BaseFakeFilter
- # AND
- # * It must have a python entrypoint assigned (returned by
- # FakeExtensionManager)
- expected = [FakeFilter1, FakeFilter4]
- result = self.handler.get_all_classes()
- self.assertEqual(expected, result)
-
- def _get_filtered_objects(self, filter_classes, index=0):
- filter_objs_initial = [1, 2, 3, 4]
- filter_properties = {'x': 'y'}
- return self.handler.get_filtered_objects(filter_classes,
- filter_objs_initial,
- filter_properties,
- index)
-
- @mock.patch.object(FakeFilter4, 'filter_all')
- @mock.patch.object(FakeFilter3, 'filter_all', return_value=None)
- def test_get_filtered_objects_return_none(self, fake3_filter_all,
- fake4_filter_all):
- filter_classes = [FakeFilter1, FakeFilter2, FakeFilter3, FakeFilter4]
- result = self._get_filtered_objects(filter_classes)
- self.assertIsNone(result)
- self.assertFalse(fake4_filter_all.called)
-
- def test_get_filtered_objects(self):
- filter_objs_expected = [1, 2, 3, 4]
- filter_classes = [FakeFilter1, FakeFilter2, FakeFilter3, FakeFilter4]
- result = self._get_filtered_objects(filter_classes)
- self.assertEqual(filter_objs_expected, result)
-
- def test_get_filtered_objects_with_filter_run_once(self):
- filter_objs_expected = [1, 2, 3, 4]
- filter_classes = [FakeFilter5]
-
- with mock.patch.object(FakeFilter5, 'filter_all',
- return_value=filter_objs_expected
- ) as fake5_filter_all:
- result = self._get_filtered_objects(filter_classes)
- self.assertEqual(filter_objs_expected, result)
- self.assertEqual(1, fake5_filter_all.call_count)
-
- result = self._get_filtered_objects(filter_classes, index=1)
- self.assertEqual(filter_objs_expected, result)
- self.assertEqual(1, fake5_filter_all.call_count)
-
- result = self._get_filtered_objects(filter_classes, index=2)
- self.assertEqual(filter_objs_expected, result)
- self.assertEqual(1, fake5_filter_all.call_count)
diff --git a/tests/unit/scheduler/test_host_filters.py b/tests/unit/scheduler/test_host_filters.py
deleted file mode 100644
index ac6b5dcf..00000000
--- a/tests/unit/scheduler/test_host_filters.py
+++ /dev/null
@@ -1,704 +0,0 @@
-# Copyright 2011 OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-"""
-Tests For Scheduler Host Filters.
-"""
-
-from oslo_context import context
-from oslo_serialization import jsonutils
-from oslotest import base as test_base
-
-from openstack.common.scheduler import filters
-from openstack.common.scheduler.filters import extra_specs_ops
-from tests.unit.scheduler import fake_hosts as fakes
-
-
-class TestFilter(filters.BaseHostFilter):
- pass
-
-
-class TestBogusFilter(object):
- """Class that doesn't inherit from BaseHostFilter."""
- pass
-
-
-class ExtraSpecsOpsTestCase(test_base.BaseTestCase):
- def _do_extra_specs_ops_test(self, value, req, matches):
- assertion = self.assertTrue if matches else self.assertFalse
- assertion(extra_specs_ops.match(value, req))
-
- def test_extra_specs_matches_simple(self):
- self._do_extra_specs_ops_test(
- value='1',
- req='1',
- matches=True)
-
- def test_extra_specs_fails_simple(self):
- self._do_extra_specs_ops_test(
- value='',
- req='1',
- matches=False)
-
- def test_extra_specs_fails_simple2(self):
- self._do_extra_specs_ops_test(
- value='3',
- req='1',
- matches=False)
-
- def test_extra_specs_fails_simple3(self):
- self._do_extra_specs_ops_test(
- value='222',
- req='2',
- matches=False)
-
- def test_extra_specs_fails_with_bogus_ops(self):
- self._do_extra_specs_ops_test(
- value='4',
- req='> 2',
- matches=False)
-
- def test_extra_specs_matches_with_op_eq(self):
- self._do_extra_specs_ops_test(
- value='123',
- req='= 123',
- matches=True)
-
- def test_extra_specs_matches_with_op_eq2(self):
- self._do_extra_specs_ops_test(
- value='124',
- req='= 123',
- matches=True)
-
- def test_extra_specs_fails_with_op_eq(self):
- self._do_extra_specs_ops_test(
- value='34',
- req='= 234',
- matches=False)
-
- def test_extra_specs_fails_with_op_eq3(self):
- self._do_extra_specs_ops_test(
- value='34',
- req='=',
- matches=False)
-
- def test_extra_specs_matches_with_op_seq(self):
- self._do_extra_specs_ops_test(
- value='123',
- req='s== 123',
- matches=True)
-
- def test_extra_specs_fails_with_op_seq(self):
- self._do_extra_specs_ops_test(
- value='1234',
- req='s== 123',
- matches=False)
-
- def test_extra_specs_matches_with_op_sneq(self):
- self._do_extra_specs_ops_test(
- value='1234',
- req='s!= 123',
- matches=True)
-
- def test_extra_specs_fails_with_op_sneq(self):
- self._do_extra_specs_ops_test(
- value='123',
- req='s!= 123',
- matches=False)
-
- def test_extra_specs_fails_with_op_sge(self):
- self._do_extra_specs_ops_test(
- value='1000',
- req='s>= 234',
- matches=False)
-
- def test_extra_specs_fails_with_op_sle(self):
- self._do_extra_specs_ops_test(
- value='1234',
- req='s<= 1000',
- matches=False)
-
- def test_extra_specs_fails_with_op_sl(self):
- self._do_extra_specs_ops_test(
- value='2',
- req='s< 12',
- matches=False)
-
- def test_extra_specs_fails_with_op_sg(self):
- self._do_extra_specs_ops_test(
- value='12',
- req='s> 2',
- matches=False)
-
- def test_extra_specs_matches_with_op_in(self):
- self._do_extra_specs_ops_test(
- value='12311321',
- req='<in> 11',
- matches=True)
-
- def test_extra_specs_matches_with_op_in2(self):
- self._do_extra_specs_ops_test(
- value='12311321',
- req='<in> 12311321',
- matches=True)
-
- def test_extra_specs_matches_with_op_in3(self):
- self._do_extra_specs_ops_test(
- value='12311321',
- req='<in> 12311321 <in>',
- matches=True)
-
- def test_extra_specs_fails_with_op_in(self):
- self._do_extra_specs_ops_test(
- value='12310321',
- req='<in> 11',
- matches=False)
-
- def test_extra_specs_fails_with_op_in2(self):
- self._do_extra_specs_ops_test(
- value='12310321',
- req='<in> 11 <in>',
- matches=False)
-
- def test_extra_specs_matches_with_op_is(self):
- self._do_extra_specs_ops_test(
- value=True,
- req='<is> True',
- matches=True)
-
- def test_extra_specs_matches_with_op_is2(self):
- self._do_extra_specs_ops_test(
- value=False,
- req='<is> False',
- matches=True)
-
- def test_extra_specs_matches_with_op_is3(self):
- self._do_extra_specs_ops_test(
- value=False,
- req='<is> Nonsense',
- matches=True)
-
- def test_extra_specs_fails_with_op_is(self):
- self._do_extra_specs_ops_test(
- value=True,
- req='<is> False',
- matches=False)
-
- def test_extra_specs_fails_with_op_is2(self):
- self._do_extra_specs_ops_test(
- value=False,
- req='<is> True',
- matches=False)
-
- def test_extra_specs_matches_with_op_or(self):
- self._do_extra_specs_ops_test(
- value='12',
- req='<or> 11 <or> 12',
- matches=True)
-
- def test_extra_specs_matches_with_op_or2(self):
- self._do_extra_specs_ops_test(
- value='12',
- req='<or> 11 <or> 12 <or>',
- matches=True)
-
- def test_extra_specs_fails_with_op_or(self):
- self._do_extra_specs_ops_test(
- value='13',
- req='<or> 11 <or> 12',
- matches=False)
-
- def test_extra_specs_fails_with_op_or2(self):
- self._do_extra_specs_ops_test(
- value='13',
- req='<or> 11 <or> 12 <or>',
- matches=False)
-
- def test_extra_specs_matches_with_op_le(self):
- self._do_extra_specs_ops_test(
- value='2',
- req='<= 10',
- matches=True)
-
- def test_extra_specs_fails_with_op_le(self):
- self._do_extra_specs_ops_test(
- value='3',
- req='<= 2',
- matches=False)
-
- def test_extra_specs_matches_with_op_ge(self):
- self._do_extra_specs_ops_test(
- value='3',
- req='>= 1',
- matches=True)
-
- def test_extra_specs_fails_with_op_ge(self):
- self._do_extra_specs_ops_test(
- value='2',
- req='>= 3',
- matches=False)
-
-
-class HostFiltersTestCase(test_base.BaseTestCase):
- """Test case for host filters."""
-
- def setUp(self):
- super(HostFiltersTestCase, self).setUp()
- self.context = context.RequestContext('fake', 'fake')
- self.json_query = jsonutils.dumps(
- ['and', ['>=', '$free_ram_mb', 1024],
- ['>=', '$free_disk_mb', 200 * 1024]])
- namespace = 'openstack.common.scheduler.filters'
- filter_handler = filters.HostFilterHandler(namespace)
- classes = filter_handler.get_all_classes()
- self.class_map = {}
- for cls in classes:
- self.class_map[cls.__name__] = cls
-
- def test_all_filters(self):
- # Double check at least a couple of known filters exist
- self.assertTrue('JsonFilter' in self.class_map)
- self.assertTrue('CapabilitiesFilter' in self.class_map)
- self.assertTrue('AvailabilityZoneFilter' in self.class_map)
- self.assertTrue('IgnoreAttemptedHostsFilter' in self.class_map)
-
- def _do_test_type_filter_extra_specs(self, ecaps, especs, passes):
- filt_cls = self.class_map['CapabilitiesFilter']()
- capabilities = {'enabled': True}
- capabilities.update(ecaps)
- service = {'disabled': False}
- filter_properties = {'resource_type': {'name': 'fake_type',
- 'extra_specs': especs}}
- host = fakes.FakeHostState('host1',
- {'free_capacity_gb': 1024,
- 'capabilities': capabilities,
- 'service': service})
- assertion = self.assertTrue if passes else self.assertFalse
- assertion(filt_cls.host_passes(host, filter_properties))
-
- def test_capability_filter_passes_extra_specs_simple(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'opt1': '1', 'opt2': '2'},
- especs={'opt1': '1', 'opt2': '2'},
- passes=True)
-
- def test_capability_filter_fails_extra_specs_simple(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'opt1': '1', 'opt2': '2'},
- especs={'opt1': '1', 'opt2': '222'},
- passes=False)
-
- def test_capability_filter_passes_extra_specs_complex(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'opt1': 10, 'opt2': 5},
- especs={'opt1': '>= 2', 'opt2': '<= 8'},
- passes=True)
-
- def test_capability_filter_fails_extra_specs_complex(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'opt1': 10, 'opt2': 5},
- especs={'opt1': '>= 2', 'opt2': '>= 8'},
- passes=False)
-
- def test_capability_filter_passes_scope_extra_specs(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'scope_lv1': {'opt1': 10}},
- especs={'capabilities:scope_lv1:opt1': '>= 2'},
- passes=True)
-
- def test_capability_filter_passes_fakescope_extra_specs(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'scope_lv1': {'opt1': 10}, 'opt2': 5},
- especs={'scope_lv1:opt1': '= 2', 'opt2': '>= 3'},
- passes=True)
-
- def test_capability_filter_fails_scope_extra_specs(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'scope_lv1': {'opt1': 10}},
- especs={'capabilities:scope_lv1:opt1': '<= 2'},
- passes=False)
-
- def test_capability_filter_passes_multi_level_scope_extra_specs(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'scope_lv0': {'scope_lv1':
- {'scope_lv2': {'opt1': 10}}}},
- especs={'capabilities:scope_lv0:scope_lv1:scope_lv2:opt1': '>= 2'},
- passes=True)
-
- def test_capability_filter_fails_wrong_scope_extra_specs(self):
- self._do_test_type_filter_extra_specs(
- ecaps={'scope_lv0': {'opt1': 10}},
- especs={'capabilities:scope_lv1:opt1': '>= 2'},
- passes=False)
-
- def test_json_filter_passes(self):
- filt_cls = self.class_map['JsonFilter']()
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'root_gb': 200,
- 'ephemeral_gb': 0},
- 'scheduler_hints': {'query': self.json_query}}
- capabilities = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 1024,
- 'free_disk_mb': 200 * 1024,
- 'capabilities': capabilities})
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_passes_with_no_query(self):
- filt_cls = self.class_map['JsonFilter']()
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'root_gb': 200,
- 'ephemeral_gb': 0}}
- capabilities = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 0,
- 'free_disk_mb': 0,
- 'capabilities': capabilities})
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_fails_on_memory(self):
- filt_cls = self.class_map['JsonFilter']()
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'root_gb': 200,
- 'ephemeral_gb': 0},
- 'scheduler_hints': {'query': self.json_query}}
- capabilities = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 1023,
- 'free_disk_mb': 200 * 1024,
- 'capabilities': capabilities})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_fails_on_disk(self):
- filt_cls = self.class_map['JsonFilter']()
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'root_gb': 200,
- 'ephemeral_gb': 0},
- 'scheduler_hints': {'query': self.json_query}}
- capabilities = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 1024,
- 'free_disk_mb': (200 * 1024) - 1,
- 'capabilities': capabilities})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_fails_on_caps_disabled(self):
- filt_cls = self.class_map['JsonFilter']()
- json_query = jsonutils.dumps(
- ['and', ['>=', '$free_ram_mb', 1024],
- ['>=', '$free_disk_mb', 200 * 1024],
- '$capabilities.enabled'])
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'root_gb': 200,
- 'ephemeral_gb': 0},
- 'scheduler_hints': {'query': json_query}}
- capabilities = {'enabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 1024,
- 'free_disk_mb': 200 * 1024,
- 'capabilities': capabilities})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_fails_on_service_disabled(self):
- filt_cls = self.class_map['JsonFilter']()
- json_query = jsonutils.dumps(
- ['and', ['>=', '$free_ram_mb', 1024],
- ['>=', '$free_disk_mb', 200 * 1024],
- ['not', '$service.disabled']])
- filter_properties = {'resource_type': {'memory_mb': 1024,
- 'local_gb': 200},
- 'scheduler_hints': {'query': json_query}}
- capabilities = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 1024,
- 'free_disk_mb': 200 * 1024,
- 'capabilities': capabilities})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_happy_day(self):
- """Test json filter more thoroughly."""
- filt_cls = self.class_map['JsonFilter']()
- raw = ['and',
- '$capabilities.enabled',
- ['=', '$capabilities.opt1', 'match'],
- ['or',
- ['and',
- ['<', '$free_ram_mb', 30],
- ['<', '$free_disk_mb', 300]],
- ['and',
- ['>', '$free_ram_mb', 30],
- ['>', '$free_disk_mb', 300]]]]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
-
- # Passes
- capabilities = {'enabled': True, 'opt1': 'match'}
- service = {'disabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 10,
- 'free_disk_mb': 200,
- 'capabilities': capabilities,
- 'service': service})
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- # Passes
- capabilities = {'enabled': True, 'opt1': 'match'}
- service = {'disabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 40,
- 'free_disk_mb': 400,
- 'capabilities': capabilities,
- 'service': service})
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- # Fails due to capabilities being disabled
- capabilities = {'enabled': False, 'opt1': 'match'}
- service = {'disabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 40,
- 'free_disk_mb': 400,
- 'capabilities': capabilities,
- 'service': service})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- # Fails due to being exact memory/disk we don't want
- capabilities = {'enabled': True, 'opt1': 'match'}
- service = {'disabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 30,
- 'free_disk_mb': 300,
- 'capabilities': capabilities,
- 'service': service})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- # Fails due to memory lower but disk higher
- capabilities = {'enabled': True, 'opt1': 'match'}
- service = {'disabled': False}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 20,
- 'free_disk_mb': 400,
- 'capabilities': capabilities,
- 'service': service})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- # Fails due to capabilities 'opt1' not equal
- capabilities = {'enabled': True, 'opt1': 'no-match'}
- service = {'enabled': True}
- host = fakes.FakeHostState('host1',
- {'free_ram_mb': 20,
- 'free_disk_mb': 400,
- 'capabilities': capabilities,
- 'service': service})
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_basic_operators(self):
- filt_cls = self.class_map['JsonFilter']()
- host = fakes.FakeHostState('host1',
- {'capabilities': {'enabled': True}})
- # (operator, arguments, expected_result)
- ops_to_test = [
- ['=', [1, 1], True],
- ['=', [1, 2], False],
- ['<', [1, 2], True],
- ['<', [1, 1], False],
- ['<', [2, 1], False],
- ['>', [2, 1], True],
- ['>', [2, 2], False],
- ['>', [2, 3], False],
- ['<=', [1, 2], True],
- ['<=', [1, 1], True],
- ['<=', [2, 1], False],
- ['>=', [2, 1], True],
- ['>=', [2, 2], True],
- ['>=', [2, 3], False],
- ['in', [1, 1], True],
- ['in', [1, 1, 2, 3], True],
- ['in', [4, 1, 2, 3], False],
- ['not', [True], False],
- ['not', [False], True],
- ['or', [True, False], True],
- ['or', [False, False], False],
- ['and', [True, True], True],
- ['and', [False, False], False],
- ['and', [True, False], False],
- # Nested ((True or False) and (2 > 1)) == Passes
- ['and', [['or', True, False], ['>', 2, 1]], True]]
-
- for (op, args, expected) in ops_to_test:
- raw = [op] + args
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertEqual(expected,
- filt_cls.host_passes(host, filter_properties))
-
- # This results in [False, True, False, True] and if any are True
- # then it passes...
- raw = ['not', True, False, True, False]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- # This results in [False, False, False] and if any are True
- # then it passes...which this doesn't
- raw = ['not', True, True, True]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_unknown_operator_raises(self):
- filt_cls = self.class_map['JsonFilter']()
- raw = ['!=', 1, 2]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- host = fakes.FakeHostState('host1',
- {'capabilities': {'enabled': True}})
- self.assertRaises(KeyError,
- filt_cls.host_passes, host, filter_properties)
-
- def test_json_filter_empty_filters_pass(self):
- filt_cls = self.class_map['JsonFilter']()
- host = fakes.FakeHostState('host1',
- {'capabilities': {'enabled': True}})
-
- raw = []
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
- raw = {}
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_invalid_num_arguments_fails(self):
- filt_cls = self.class_map['JsonFilter']()
- host = fakes.FakeHostState('host1',
- {'capabilities': {'enabled': True}})
-
- raw = ['>', ['and', ['or', ['not', ['<', ['>=', ['<=', ['in', ]]]]]]]]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- raw = ['>', 1]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
-
- def test_json_filter_unknown_variable_ignored(self):
- filt_cls = self.class_map['JsonFilter']()
- host = fakes.FakeHostState('host1',
- {'capabilities': {'enabled': True}})
-
- raw = ['=', '$........', 1, 1]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- raw = ['=', '$foo', 2, 2]
- filter_properties = {
- 'scheduler_hints': {
- 'query': jsonutils.dumps(raw),
- },
- }
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- @staticmethod
- def _make_zone_request(zone, is_admin=False):
- ctxt = context.RequestContext('fake', 'fake', is_admin=is_admin)
- return {
- 'context': ctxt,
- 'request_spec': {
- 'resource_properties': {
- 'availability_zone': zone
- }
- }
- }
-
- def test_availability_zone_filter_same(self):
- filt_cls = self.class_map['AvailabilityZoneFilter']()
- service = {'availability_zone': 'nova'}
- request = self._make_zone_request('nova')
- host = fakes.FakeHostState('host1',
- {'service': service})
- self.assertTrue(filt_cls.host_passes(host, request))
-
- def test_availability_zone_filter_different(self):
- filt_cls = self.class_map['AvailabilityZoneFilter']()
- service = {'availability_zone': 'nova'}
- request = self._make_zone_request('bad')
- host = fakes.FakeHostState('host1',
- {'service': service})
- self.assertFalse(filt_cls.host_passes(host, request))
-
- def test_availability_zone_filter_empty(self):
- filt_cls = self.class_map['AvailabilityZoneFilter']()
- service = {'availability_zone': 'nova'}
- request = {}
- host = fakes.FakeHostState('host1',
- {'service': service})
- self.assertTrue(filt_cls.host_passes(host, request))
-
- def test_ignore_attempted_hosts_filter_disabled(self):
- # Test case where re-scheduling is disabled.
- filt_cls = self.class_map['IgnoreAttemptedHostsFilter']()
- host = fakes.FakeHostState('host1', {})
- filter_properties = {}
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- def test_ignore_attempted_hosts_filter_pass(self):
- # Node not previously tried.
- filt_cls = self.class_map['IgnoreAttemptedHostsFilter']()
- host = fakes.FakeHostState('host1', {})
- attempted = dict(num_attempts=2, hosts=['host2'])
- filter_properties = dict(retry=attempted)
- self.assertTrue(filt_cls.host_passes(host, filter_properties))
-
- def test_ignore_attempted_hosts_filter_fail(self):
- # Node was already tried.
- filt_cls = self.class_map['IgnoreAttemptedHostsFilter']()
- host = fakes.FakeHostState('host1', {})
- attempted = dict(num_attempts=2, hosts=['host1'])
- filter_properties = dict(retry=attempted)
- self.assertFalse(filt_cls.host_passes(host, filter_properties))
diff --git a/tests/unit/scheduler/test_weights.py b/tests/unit/scheduler/test_weights.py
deleted file mode 100644
index 3ddadec0..00000000
--- a/tests/unit/scheduler/test_weights.py
+++ /dev/null
@@ -1,65 +0,0 @@
-# Copyright 2011-2012 OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-"""
-Tests For Scheduler weights.
-"""
-
-from oslotest import base as test_base
-
-from openstack.common.scheduler import base_weight
-from tests.unit import fakes
-
-
-class TestWeightHandler(test_base.BaseTestCase):
- def test_get_all_classes(self):
- namespace = "openstack.common.tests.fakes.weights"
- handler = base_weight.BaseWeightHandler(
- base_weight.BaseWeigher, namespace)
- classes = handler.get_all_classes()
- self.assertTrue(fakes.FakeWeigher1 in classes)
- self.assertTrue(fakes.FakeWeigher2 in classes)
- self.assertFalse(fakes.FakeClass in classes)
-
- def test_no_multiplier(self):
- class FakeWeigher(base_weight.BaseWeigher):
- def _weigh_object(self, *args, **kwargs):
- pass
-
- self.assertEqual(1.0,
- FakeWeigher().weight_multiplier())
-
- def test_no_weight_object(self):
- class FakeWeigher(base_weight.BaseWeigher):
- def weight_multiplier(self, *args, **kwargs):
- pass
- self.assertRaises(TypeError,
- FakeWeigher)
-
- def test_normalization(self):
- # weight_list, expected_result, minval, maxval
- map_ = (
- ((), (), None, None),
- ((0.0, 0.0), (0.0, 0.0), None, None),
- ((1.0, 1.0), (0.0, 0.0), None, None),
-
- ((20.0, 50.0), (0.0, 1.0), None, None),
- ((20.0, 50.0), (0.0, 0.375), None, 100.0),
- ((20.0, 50.0), (0.4, 1.0), 0.0, None),
- ((20.0, 50.0), (0.2, 0.5), 0.0, 100.0),
- )
- for seq, result, minval, maxval in map_:
- ret = base_weight.normalize(seq, minval=minval, maxval=maxval)
- self.assertEqual(result, tuple(ret))
diff --git a/tests/unit/test_cliutils.py b/tests/unit/test_cliutils.py
deleted file mode 100644
index 0fe69fc7..00000000
--- a/tests/unit/test_cliutils.py
+++ /dev/null
@@ -1,699 +0,0 @@
-# Copyright 2012 Red Hat, Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import sys
-
-import fixtures
-import mock
-from oslotest import base as test_base
-import six
-
-from openstack.common import cliutils
-
-
-class ValidateArgsTest(test_base.BaseTestCase):
-
- def test_lambda_no_args(self):
- cliutils.validate_args(lambda: None)
-
- def _test_lambda_with_args(self, *args, **kwargs):
- cliutils.validate_args(lambda x, y: None, *args, **kwargs)
-
- def test_lambda_positional_args(self):
- self._test_lambda_with_args(1, 2)
-
- def test_lambda_kwargs(self):
- self._test_lambda_with_args(x=1, y=2)
-
- def test_lambda_mixed_kwargs(self):
- self._test_lambda_with_args(1, y=2)
-
- def test_lambda_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_args)
-
- def test_lambda_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_args, 1)
-
- def test_lambda_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_args, y=2)
-
- def _test_lambda_with_default(self, *args, **kwargs):
- cliutils.validate_args(lambda x, y, z=3: None, *args, **kwargs)
-
- def test_lambda_positional_args_with_default(self):
- self._test_lambda_with_default(1, 2)
-
- def test_lambda_kwargs_with_default(self):
- self._test_lambda_with_default(x=1, y=2)
-
- def test_lambda_mixed_kwargs_with_default(self):
- self._test_lambda_with_default(1, y=2)
-
- def test_lambda_positional_args_all_with_default(self):
- self._test_lambda_with_default(1, 2, 3)
-
- def test_lambda_kwargs_all_with_default(self):
- self._test_lambda_with_default(x=1, y=2, z=3)
-
- def test_lambda_mixed_kwargs_all_with_default(self):
- self._test_lambda_with_default(1, y=2, z=3)
-
- def test_lambda_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_default)
-
- def test_lambda_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_default, 1)
-
- def test_lambda_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_default, y=2)
-
- def test_lambda_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_lambda_with_default, y=2, z=3)
-
- def test_function_no_args(self):
- def func():
- pass
- cliutils.validate_args(func)
-
- def _test_function_with_args(self, *args, **kwargs):
- def func(x, y):
- pass
- cliutils.validate_args(func, *args, **kwargs)
-
- def test_function_positional_args(self):
- self._test_function_with_args(1, 2)
-
- def test_function_kwargs(self):
- self._test_function_with_args(x=1, y=2)
-
- def test_function_mixed_kwargs(self):
- self._test_function_with_args(1, y=2)
-
- def test_function_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_args)
-
- def test_function_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_args, 1)
-
- def test_function_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_args, y=2)
-
- def _test_function_with_default(self, *args, **kwargs):
- def func(x, y, z=3):
- pass
- cliutils.validate_args(func, *args, **kwargs)
-
- def test_function_positional_args_with_default(self):
- self._test_function_with_default(1, 2)
-
- def test_function_kwargs_with_default(self):
- self._test_function_with_default(x=1, y=2)
-
- def test_function_mixed_kwargs_with_default(self):
- self._test_function_with_default(1, y=2)
-
- def test_function_positional_args_all_with_default(self):
- self._test_function_with_default(1, 2, 3)
-
- def test_function_kwargs_all_with_default(self):
- self._test_function_with_default(x=1, y=2, z=3)
-
- def test_function_mixed_kwargs_all_with_default(self):
- self._test_function_with_default(1, y=2, z=3)
-
- def test_function_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_default)
-
- def test_function_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_default, 1)
-
- def test_function_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_default, y=2)
-
- def test_function_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_function_with_default, y=2, z=3)
-
- def test_bound_method_no_args(self):
- class Foo(object):
- def bar(self):
- pass
- cliutils.validate_args(Foo().bar)
-
- def _test_bound_method_with_args(self, *args, **kwargs):
- class Foo(object):
- def bar(self, x, y):
- pass
- cliutils.validate_args(Foo().bar, *args, **kwargs)
-
- def test_bound_method_positional_args(self):
- self._test_bound_method_with_args(1, 2)
-
- def test_bound_method_kwargs(self):
- self._test_bound_method_with_args(x=1, y=2)
-
- def test_bound_method_mixed_kwargs(self):
- self._test_bound_method_with_args(1, y=2)
-
- def test_bound_method_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_args)
-
- def test_bound_method_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_args, 1)
-
- def test_bound_method_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_args, y=2)
-
- def _test_bound_method_with_default(self, *args, **kwargs):
- class Foo(object):
- def bar(self, x, y, z=3):
- pass
- cliutils.validate_args(Foo().bar, *args, **kwargs)
-
- def test_bound_method_positional_args_with_default(self):
- self._test_bound_method_with_default(1, 2)
-
- def test_bound_method_kwargs_with_default(self):
- self._test_bound_method_with_default(x=1, y=2)
-
- def test_bound_method_mixed_kwargs_with_default(self):
- self._test_bound_method_with_default(1, y=2)
-
- def test_bound_method_positional_args_all_with_default(self):
- self._test_bound_method_with_default(1, 2, 3)
-
- def test_bound_method_kwargs_all_with_default(self):
- self._test_bound_method_with_default(x=1, y=2, z=3)
-
- def test_bound_method_mixed_kwargs_all_with_default(self):
- self._test_bound_method_with_default(1, y=2, z=3)
-
- def test_bound_method_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_default)
-
- def test_bound_method_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_default, 1)
-
- def test_bound_method_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_default, y=2)
-
- def test_bound_method_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_bound_method_with_default, y=2, z=3)
-
- def test_unbound_method_no_args(self):
- class Foo(object):
- def bar(self):
- pass
- cliutils.validate_args(Foo.bar, Foo())
-
- def _test_unbound_method_with_args(self, *args, **kwargs):
- class Foo(object):
- def bar(self, x, y):
- pass
- cliutils.validate_args(Foo.bar, Foo(), *args, **kwargs)
-
- def test_unbound_method_positional_args(self):
- self._test_unbound_method_with_args(1, 2)
-
- def test_unbound_method_kwargs(self):
- self._test_unbound_method_with_args(x=1, y=2)
-
- def test_unbound_method_mixed_kwargs(self):
- self._test_unbound_method_with_args(1, y=2)
-
- def test_unbound_method_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_args)
-
- def test_unbound_method_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_args, 1)
-
- def test_unbound_method_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_args, y=2)
-
- def _test_unbound_method_with_default(self, *args, **kwargs):
- class Foo(object):
- def bar(self, x, y, z=3):
- pass
- cliutils.validate_args(Foo.bar, Foo(), *args, **kwargs)
-
- def test_unbound_method_positional_args_with_default(self):
- self._test_unbound_method_with_default(1, 2)
-
- def test_unbound_method_kwargs_with_default(self):
- self._test_unbound_method_with_default(x=1, y=2)
-
- def test_unbound_method_mixed_kwargs_with_default(self):
- self._test_unbound_method_with_default(1, y=2)
-
- def test_unbound_method_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_default)
-
- def test_unbound_method_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_default, 1)
-
- def test_unbound_method_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_default, y=2)
-
- def test_unbound_method_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_unbound_method_with_default, y=2, z=3)
-
- def test_class_method_no_args(self):
- class Foo(object):
- @classmethod
- def bar(cls):
- pass
- cliutils.validate_args(Foo.bar)
-
- def _test_class_method_with_args(self, *args, **kwargs):
- class Foo(object):
- @classmethod
- def bar(cls, x, y):
- pass
- cliutils.validate_args(Foo.bar, *args, **kwargs)
-
- def test_class_method_positional_args(self):
- self._test_class_method_with_args(1, 2)
-
- def test_class_method_kwargs(self):
- self._test_class_method_with_args(x=1, y=2)
-
- def test_class_method_mixed_kwargs(self):
- self._test_class_method_with_args(1, y=2)
-
- def test_class_method_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_args)
-
- def test_class_method_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_args, 1)
-
- def test_class_method_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_args, y=2)
-
- def _test_class_method_with_default(self, *args, **kwargs):
- class Foo(object):
- @classmethod
- def bar(cls, x, y, z=3):
- pass
- cliutils.validate_args(Foo.bar, *args, **kwargs)
-
- def test_class_method_positional_args_with_default(self):
- self._test_class_method_with_default(1, 2)
-
- def test_class_method_kwargs_with_default(self):
- self._test_class_method_with_default(x=1, y=2)
-
- def test_class_method_mixed_kwargs_with_default(self):
- self._test_class_method_with_default(1, y=2)
-
- def test_class_method_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_default)
-
- def test_class_method_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_default, 1)
-
- def test_class_method_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_default, y=2)
-
- def test_class_method_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_class_method_with_default, y=2, z=3)
-
- def test_static_method_no_args(self):
- class Foo(object):
- @staticmethod
- def bar():
- pass
- cliutils.validate_args(Foo.bar)
-
- def _test_static_method_with_args(self, *args, **kwargs):
- class Foo(object):
- @staticmethod
- def bar(x, y):
- pass
- cliutils.validate_args(Foo.bar, *args, **kwargs)
-
- def test_static_method_positional_args(self):
- self._test_static_method_with_args(1, 2)
-
- def test_static_method_kwargs(self):
- self._test_static_method_with_args(x=1, y=2)
-
- def test_static_method_mixed_kwargs(self):
- self._test_static_method_with_args(1, y=2)
-
- def test_static_method_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_args)
-
- def test_static_method_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_args, 1)
-
- def test_static_method_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_args, y=2)
-
- def _test_static_method_with_default(self, *args, **kwargs):
- class Foo(object):
- @staticmethod
- def bar(x, y, z=3):
- pass
- cliutils.validate_args(Foo.bar, *args, **kwargs)
-
- def test_static_method_positional_args_with_default(self):
- self._test_static_method_with_default(1, 2)
-
- def test_static_method_kwargs_with_default(self):
- self._test_static_method_with_default(x=1, y=2)
-
- def test_static_method_mixed_kwargs_with_default(self):
- self._test_static_method_with_default(1, y=2)
-
- def test_static_method_with_default_missing_args1(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_default)
-
- def test_static_method_with_default_missing_args2(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_default, 1)
-
- def test_static_method_with_default_missing_args3(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_default, y=2)
-
- def test_static_method_with_default_missing_args4(self):
- self.assertRaises(cliutils.MissingArgs,
- self._test_static_method_with_default, y=2, z=3)
-
-
-class _FakeResult(object):
- def __init__(self, name, value):
- self.name = name
- self.value = value
-
-
-class PrintResultTestCase(test_base.BaseTestCase):
-
- def setUp(self):
- super(PrintResultTestCase, self).setUp()
- self.mock_add_row = mock.MagicMock()
- self.useFixture(fixtures.MonkeyPatch(
- "prettytable.PrettyTable.add_row",
- self.mock_add_row))
- self.mock_get_string = mock.MagicMock(return_value="")
- self.useFixture(fixtures.MonkeyPatch(
- "prettytable.PrettyTable.get_string",
- self.mock_get_string))
- self.mock_init = mock.MagicMock(return_value=None)
- self.useFixture(fixtures.MonkeyPatch(
- "prettytable.PrettyTable.__init__",
- self.mock_init))
- # NOTE(dtantsur): won't work with mocked __init__
- self.useFixture(fixtures.MonkeyPatch(
- "prettytable.PrettyTable.align",
- mock.MagicMock()))
-
- def test_print_list_sort_by_str(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k3", 2),
- _FakeResult("k2", 3)]
-
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=0)
-
- self.assertEqual(self.mock_add_row.call_args_list,
- [mock.call(["k1", 1]),
- mock.call(["k3", 2]),
- mock.call(["k2", 3])])
- self.mock_get_string.assert_called_with(sortby="Name")
- self.mock_init.assert_called_once_with(["Name", "Value"])
-
- def test_print_list_sort_by_integer(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k2", 3),
- _FakeResult("k3", 2)]
-
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=1)
-
- self.assertEqual(self.mock_add_row.call_args_list,
- [mock.call(["k1", 1]),
- mock.call(["k2", 3]),
- mock.call(["k3", 2])])
- self.mock_get_string.assert_called_with(sortby="Value")
- self.mock_init.assert_called_once_with(["Name", "Value"])
-
- def test_print_list_sort_by_none(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k3", 3),
- _FakeResult("k2", 2)]
-
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=None)
-
- self.assertEqual(self.mock_add_row.call_args_list,
- [mock.call(["k1", 1]),
- mock.call(["k3", 3]),
- mock.call(["k2", 2])])
- self.mock_get_string.assert_called_with()
- self.mock_init.assert_called_once_with(["Name", "Value"])
-
- def test_print_dict(self):
- cliutils.print_dict({"K": "k", "Key": "Value"})
- cliutils.print_dict({"K": "k", "Key": "Long\\nValue"})
- self.mock_add_row.assert_has_calls([
- mock.call(["K", "k"]),
- mock.call(["Key", "Value"]),
- mock.call(["K", "k"]),
- mock.call(["Key", "Long"]),
- mock.call(["", "Value"])],
- any_order=True)
-
- def test_print_list_field_labels(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k3", 3),
- _FakeResult("k2", 2)]
- field_labels = ["Another Name", "Another Value"]
-
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=None,
- field_labels=field_labels)
-
- self.assertEqual(self.mock_add_row.call_args_list,
- [mock.call(["k1", 1]),
- mock.call(["k3", 3]),
- mock.call(["k2", 2])])
- self.mock_init.assert_called_once_with(field_labels)
-
- def test_print_list_field_labels_sort(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k3", 3),
- _FakeResult("k2", 2)]
- field_labels = ["Another Name", "Another Value"]
-
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=0,
- field_labels=field_labels)
-
- self.assertEqual(self.mock_add_row.call_args_list,
- [mock.call(["k1", 1]),
- mock.call(["k3", 3]),
- mock.call(["k2", 2])])
- self.mock_init.assert_called_once_with(field_labels)
- self.mock_get_string.assert_called_with(sortby="Another Name")
-
- def test_print_list_field_labels_too_many(self):
- objs = [_FakeResult("k1", 1),
- _FakeResult("k3", 3),
- _FakeResult("k2", 2)]
- field_labels = ["Another Name", "Another Value", "Redundant"]
-
- self.assertRaises(ValueError, cliutils.print_list,
- objs, ["Name", "Value"], sortby_index=None,
- field_labels=field_labels)
-
-
-class PrintResultStringTestCase(test_base.BaseTestCase):
-
- def test_print_list_string(self):
- objs = [_FakeResult("k1", 1)]
- field_labels = ["Another Name", "Another Value"]
-
- orig = sys.stdout
- sys.stdout = six.StringIO()
- cliutils.print_list(objs, ["Name", "Value"], sortby_index=0,
- field_labels=field_labels)
- out = sys.stdout.getvalue()
- sys.stdout.close()
- sys.stdout = orig
- expected = '''\
-+--------------+---------------+
-| Another Name | Another Value |
-+--------------+---------------+
-| k1 | 1 |
-+--------------+---------------+
-'''
- self.assertEqual(expected, out)
-
- def test_print_dict_string(self):
- orig = sys.stdout
- sys.stdout = six.StringIO()
- cliutils.print_dict({"K": "k", "Key": "Value"})
- out = sys.stdout.getvalue()
- sys.stdout.close()
- sys.stdout = orig
- expected = '''\
-+----------+-------+
-| Property | Value |
-+----------+-------+
-| K | k |
-| Key | Value |
-+----------+-------+
-'''
- self.assertEqual(expected, out)
-
- def test_print_dict_string_custom_headers(self):
- orig = sys.stdout
- sys.stdout = six.StringIO()
- cliutils.print_dict({"K": "k", "Key": "Value"}, dict_property='Foo',
- dict_value='Bar')
- out = sys.stdout.getvalue()
- sys.stdout.close()
- sys.stdout = orig
- expected = '''\
-+-----+-------+
-| Foo | Bar |
-+-----+-------+
-| K | k |
-| Key | Value |
-+-----+-------+
-'''
- self.assertEqual(expected, out)
-
- def test_print_dict_string_sorted(self):
- orig = sys.stdout
- sys.stdout = six.StringIO()
- cliutils.print_dict({"Foo": "k", "Bar": "Value"})
- out = sys.stdout.getvalue()
- sys.stdout.close()
- sys.stdout = orig
- expected = '''\
-+----------+-------+
-| Property | Value |
-+----------+-------+
-| Bar | Value |
-| Foo | k |
-+----------+-------+
-'''
- self.assertEqual(expected, out)
-
-
-class DecoratorsTestCase(test_base.BaseTestCase):
-
- def test_arg(self):
- func_args = [("--image", ), ("--flavor", )]
- func_kwargs = [dict(default=None,
- metavar="<image>"),
- dict(default=None,
- metavar="<flavor>")]
-
- @cliutils.arg(*func_args[1], **func_kwargs[1])
- @cliutils.arg(*func_args[0], **func_kwargs[0])
- def dummy_func():
- pass
-
- self.assertTrue(hasattr(dummy_func, "arguments"))
- self.assertEqual(len(dummy_func.arguments), 2)
- for args_kwargs in zip(func_args, func_kwargs):
- self.assertIn(args_kwargs, dummy_func.arguments)
-
- def test_unauthenticated(self):
- def dummy_func():
- pass
-
- self.assertFalse(cliutils.isunauthenticated(dummy_func))
- dummy_func = cliutils.unauthenticated(dummy_func)
- self.assertTrue(cliutils.isunauthenticated(dummy_func))
-
-
-class EnvTestCase(test_base.BaseTestCase):
-
- def test_env(self):
- env = {"alpha": "a", "beta": "b"}
- self.useFixture(fixtures.MonkeyPatch("os.environ", env))
- self.assertEqual(cliutils.env("beta"), env["beta"])
- self.assertEqual(cliutils.env("beta", "alpha"), env["beta"])
- self.assertEqual(cliutils.env("alpha", "beta"), env["alpha"])
- self.assertEqual(cliutils.env("gamma", "beta"), env["beta"])
- self.assertEqual(cliutils.env("gamma"), "")
- self.assertEqual(cliutils.env("gamma", default="c"), "c")
-
-
-class GetPasswordTestCase(test_base.BaseTestCase):
-
- def setUp(self):
- super(GetPasswordTestCase, self).setUp()
-
- class FakeFile(object):
- def isatty(self):
- return True
-
- self.useFixture(fixtures.MonkeyPatch("sys.stdin", FakeFile()))
-
- def test_get_password(self):
- self.useFixture(fixtures.MonkeyPatch("getpass.getpass",
- lambda prompt: "mellon"))
- self.assertEqual(cliutils.get_password(), "mellon")
-
- def test_get_password_verify(self):
- env = {"OS_VERIFY_PASSWORD": "True"}
- self.useFixture(fixtures.MonkeyPatch("os.environ", env))
- self.useFixture(fixtures.MonkeyPatch("getpass.getpass",
- lambda prompt: "mellon"))
- self.assertEqual(cliutils.get_password(), "mellon")
-
- def test_get_password_verify_failure(self):
- env = {"OS_VERIFY_PASSWORD": "True"}
- self.useFixture(fixtures.MonkeyPatch("os.environ", env))
- self.useFixture(fixtures.MonkeyPatch("getpass.getpass",
- lambda prompt: prompt))
- self.assertIsNone(cliutils.get_password())
diff --git a/tests/unit/test_imageutils.py b/tests/unit/test_imageutils.py
deleted file mode 100644
index 94bd13f8..00000000
--- a/tests/unit/test_imageutils.py
+++ /dev/null
@@ -1,199 +0,0 @@
-# Copyright (C) 2012 Yahoo! Inc.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-from oslotest import base as test_base
-import testscenarios
-
-from openstack.common import imageutils
-
-load_tests = testscenarios.load_tests_apply_scenarios
-
-
-class ImageUtilsRawTestCase(test_base.BaseTestCase):
-
- _image_name = [
- ('disk_config', dict(image_name='disk.config')),
- ]
-
- _file_format = [
- ('raw', dict(file_format='raw')),
- ]
-
- _virtual_size = [
- ('64M', dict(virtual_size='64M',
- exp_virtual_size=67108864)),
- ('64M_with_byte_hint', dict(virtual_size='64M (67108844 bytes)',
- exp_virtual_size=67108844)),
- ('64M_byte', dict(virtual_size='67108844',
- exp_virtual_size=67108844)),
- ('4.4M', dict(virtual_size='4.4M',
- exp_virtual_size=4613735)),
- ('4.4M_with_byte_hint', dict(virtual_size='4.4M (4592640 bytes)',
- exp_virtual_size=4592640)),
- ('2K', dict(virtual_size='2K',
- exp_virtual_size=2048)),
- ('2K_with_byte_hint', dict(virtual_size='2K (2048 bytes)',
- exp_virtual_size=2048)),
- ]
-
- _disk_size = [
- ('96K', dict(disk_size='96K',
- exp_disk_size=98304)),
- ('96K_byte', dict(disk_size='963434',
- exp_disk_size=963434)),
- ('3.1M', dict(disk_size='3.1G',
- exp_disk_size=3328599655)),
- ]
-
- _garbage_before_snapshot = [
- ('no_garbage', dict(garbage_before_snapshot=None)),
- ('garbage_before_snapshot_list', dict(garbage_before_snapshot=False)),
- ('garbage_after_snapshot_list', dict(garbage_before_snapshot=True)),
- ]
-
- _snapshot_count = [
- ('no_snapshots', dict(snapshot_count=None)),
- ('one_snapshots', dict(snapshot_count=1)),
- ('three_snapshots', dict(snapshot_count=3)),
- ]
-
- @classmethod
- def generate_scenarios(cls):
- cls.scenarios = testscenarios.multiply_scenarios(
- cls._image_name,
- cls._file_format,
- cls._virtual_size,
- cls._disk_size,
- cls._garbage_before_snapshot,
- cls._snapshot_count)
-
- def _initialize_img_info(self):
- return ('image: %s' % self.image_name,
- 'file_format: %s' % self.file_format,
- 'virtual_size: %s' % self.virtual_size,
- 'disk_size: %s' % self.disk_size)
-
- def _insert_snapshots(self, img_info):
- img_info = img_info + ('Snapshot list:',)
- img_info = img_info + ('ID '
- 'TAG '
- 'VM SIZE '
- 'DATE '
- 'VM CLOCK',)
- for i in range(self.snapshot_count):
- img_info = img_info + ('%d '
- 'd9a9784a500742a7bb95627bb3aace38 '
- '0 2012-08-20 10:52:46 '
- '00:00:00.000' % (i + 1),)
- return img_info
-
- def _base_validation(self, image_info):
- self.assertEqual(image_info.image, self.image_name)
- self.assertEqual(image_info.file_format, self.file_format)
- self.assertEqual(image_info.virtual_size, self.exp_virtual_size)
- self.assertEqual(image_info.disk_size, self.exp_disk_size)
- if self.snapshot_count is not None:
- self.assertEqual(len(image_info.snapshots), self.snapshot_count)
-
- def test_qemu_img_info(self):
- img_info = self._initialize_img_info()
- if self.garbage_before_snapshot is True:
- img_info = img_info + ('blah BLAH: bb',)
- if self.snapshot_count is not None:
- img_info = self._insert_snapshots(img_info)
- if self.garbage_before_snapshot is False:
- img_info = img_info + ('junk stuff: bbb',)
- example_output = '\n'.join(img_info)
- image_info = imageutils.QemuImgInfo(example_output)
- self._base_validation(image_info)
-
-ImageUtilsRawTestCase.generate_scenarios()
-
-
-class ImageUtilsQemuTestCase(ImageUtilsRawTestCase):
-
- _file_format = [
- ('qcow2', dict(file_format='qcow2')),
- ]
-
- _qcow2_cluster_size = [
- ('65536', dict(cluster_size='65536', exp_cluster_size=65536)),
- ]
-
- _qcow2_encrypted = [
- ('no_encryption', dict(encrypted=None)),
- ('encrypted', dict(encrypted='yes')),
- ]
-
- _qcow2_backing_file = [
- ('no_backing_file', dict(backing_file=None)),
- ('backing_file_path',
- dict(backing_file='/var/lib/nova/a328c7998805951a_2',
- exp_backing_file='/var/lib/nova/a328c7998805951a_2')),
- ('backing_file_path_with_actual_path',
- dict(backing_file='/var/lib/nova/a328c7998805951a_2 '
- '(actual path: /b/3a988059e51a_2)',
- exp_backing_file='/b/3a988059e51a_2')),
- ]
-
- @classmethod
- def generate_scenarios(cls):
- cls.scenarios = testscenarios.multiply_scenarios(
- cls._image_name,
- cls._file_format,
- cls._virtual_size,
- cls._disk_size,
- cls._garbage_before_snapshot,
- cls._snapshot_count,
- cls._qcow2_cluster_size,
- cls._qcow2_encrypted,
- cls._qcow2_backing_file)
-
- def test_qemu_img_info(self):
- img_info = self._initialize_img_info()
- img_info = img_info + ('cluster_size: %s' % self.cluster_size,)
- if self.backing_file is not None:
- img_info = img_info + ('backing file: %s' %
- self.backing_file,)
- if self.encrypted is not None:
- img_info = img_info + ('encrypted: %s' % self.encrypted,)
- if self.garbage_before_snapshot is True:
- img_info = img_info + ('blah BLAH: bb',)
- if self.snapshot_count is not None:
- img_info = self._insert_snapshots(img_info)
- if self.garbage_before_snapshot is False:
- img_info = img_info + ('junk stuff: bbb',)
- example_output = '\n'.join(img_info)
- image_info = imageutils.QemuImgInfo(example_output)
- self._base_validation(image_info)
- self.assertEqual(image_info.cluster_size, self.exp_cluster_size)
- if self.backing_file is not None:
- self.assertEqual(image_info.backing_file,
- self.exp_backing_file)
- if self.encrypted is not None:
- self.assertEqual(image_info.encrypted, self.encrypted)
-
-ImageUtilsQemuTestCase.generate_scenarios()
-
-
-class ImageUtilsBlankTestCase(test_base.BaseTestCase):
- def test_qemu_img_info_blank(self):
- example_output = '\n'.join(['image: None', 'file_format: None',
- 'virtual_size: None', 'disk_size: None',
- 'cluster_size: None',
- 'backing_file: None'])
- image_info = imageutils.QemuImgInfo()
- self.assertEqual(str(image_info), example_output)
- self.assertEqual(len(image_info.snapshots), 0)
diff --git a/tests/unit/test_memorycache.py b/tests/unit/test_memorycache.py
deleted file mode 100644
index eae6cd0a..00000000
--- a/tests/unit/test_memorycache.py
+++ /dev/null
@@ -1,55 +0,0 @@
-# Copyright 2013 Nebula, Inc.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import time
-
-import mock
-from oslotest import base as test_base
-
-from openstack.common import memorycache
-
-
-class MemorycacheTest(test_base.BaseTestCase):
- def setUp(self):
- self.client = memorycache.get_client()
- super(MemorycacheTest, self).setUp()
-
- def test_set_get(self):
- self.client.set('foo', 'bar')
- self.assertEqual(self.client.get('foo'), 'bar')
-
- def test_add_get(self):
- self.client.add('foo', 'bar')
- self.assertEqual(self.client.get('foo'), 'bar')
-
- def test_set_add_get(self):
- self.client.set('foo', 'bar')
- self.client.add('foo', 'baz')
- self.assertEqual(self.client.get('foo'), 'bar')
-
- def test_set_delete(self):
- self.client.set('foo', 'bar')
- self.client.delete('foo')
- self.assertIsNone(self.client.get('foo'))
-
- def test_timeout(self):
- now = time.time()
- with mock.patch('time.time') as time_mock:
- time_mock.return_value = now
- self.client.set('foo', 'bar', time=3)
- time_mock.return_value = now + 1
- self.assertEqual(self.client.get('foo'), 'bar')
- time_mock.return_value = now + 3
- self.assertIsNone(self.client.get('foo'))
View Lorry Controller Status