openstack/octavia
Code Issues Proposed changes

Adding Python 3.4 Compatibility and Tox Check

Browse Source 
Change-Id: I2a2ec83baa13f4433e32c180d7a057553348a906
This commit is contained in:
Adam Harwell 2015-05-21 14:22:47 -07:00 committed by Michael Johnson
parent 907eaa54c5
commit e6de970a38
34 changed files with 240 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
octavia/__init__.py
View File

@ -15,5 +15,9 @@
import gettext
import six
gettext.install('octavia', unicode=1)
if six.PY2:
gettext.install('octavia', unicode=1)
else:
gettext.install('octavia')

2
octavia/amphorae/backends/health_daemon/config.py
View File

@ -15,7 +15,7 @@
import collections
import json
import singleton
import octavia.amphorae.backends.health_daemon.singleton as singleton
@singleton.singleton

4
octavia/amphorae/backends/health_daemon/status_message.py
View File

@ -20,7 +20,7 @@ import json
def encode(msg, key):
result = {}
src = json.dumps(msg)
hmc = hmac.new(key, src, hashlib.sha1)
hmc = hmac.new(key.encode('ascii'), src.encode('ascii'), hashlib.sha1)
result['msg'] = msg
result['hmac'] = hmc.hexdigest()
return json.dumps(result)
@ -29,5 +29,5 @@ def encode(msg, key):
def checkhmac(envelope_str, key):
envelope = json.loads(envelope_str)
src = json.dumps(envelope['msg'])
hmc = hmac.new(key, src, hashlib.sha1)
hmc = hmac.new(key.encode('ascii'), src.encode('ascii'), hashlib.sha1)
return hmc.hexdigest() == envelope['hmac']

6
octavia/amphorae/drivers/haproxy/ssh_driver.py
View File

@ -17,6 +17,7 @@ import time
from oslo_log import log as logging
import paramiko
import six
from octavia.amphorae.driver_exceptions import exceptions as exc
from octavia.amphorae.drivers import driver_base as driver_base
@ -198,7 +199,8 @@ class HaproxyManager(driver_base.AmphoraLoadBalancerDriver):
return stdout, stderr
def _connect(self, hostname):
for attempts in xrange(self.amp_config.connection_max_retries):
for attempts in six.moves.xrange(
self.amp_config.connection_max_retries):
try:
self.client.connect(hostname=hostname,
username=self.amp_config.username,
@ -274,7 +276,7 @@ class HaproxyManager(driver_base.AmphoraLoadBalancerDriver):
# Write data to temp file to prepare for upload
for datum in data:
temp = tempfile.NamedTemporaryFile(delete=True)
temp.write(datum)
temp.write(datum.encode('ascii'))
temp.flush()
temps.append(temp)

4
octavia/api/v1/types/base.py
View File

@ -12,8 +12,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from wsme import types as wtypes
if six.PY3:
unicode = str
class IPAddressType(wtypes.UserType):
basetype = unicode

36
octavia/common/tls_utils/cert_parser.py
View File

@ -13,10 +13,10 @@
# License for the specific language governing permissions and limitations
# under the License.
from cryptography.hazmat import backends
from cryptography import x509
from OpenSSL import crypto
from OpenSSL import SSL
import pyasn1.codec.der.decoder as decoder
import pyasn1_modules.rfc2459 as rfc2459
import six
import octavia.common.exceptions as exceptions
@ -100,25 +100,21 @@ def get_host_names(certificate):
'dns_names' is a list of dNSNames (possibly empty) from
the SubjectAltNames of the certificate.
"""
try:
cert = x509.load_pem_x509_certificate(certificate,
backends.default_backend())
ext = cert.extensions.get_extension_for_oid(
x509.OID_SUBJECT_ALTERNATIVE_NAME
)
cn = cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME)[0]
x509 = _get_x509_from_pem_bytes(certificate)
hostnames = {}
if hasattr(x509.get_subject(), 'CN'):
hostnames['cn'] = x509.get_subject().CN
hostnames['dns_names'] = []
num_exts = x509.get_extension_count()
for i in range(0, num_exts):
ext = x509.get_extension(i)
if ext.get_short_name() == 'subjectAltName':
data = ext.get_data()
general_names_container = decoder.decode(
data, asn1Spec=rfc2459.GeneralNames())
for general_names in general_names_container[0]:
if general_names.getName() == 'dNSName':
octets = general_names.getComponent().asOctets()
hostnames['dns_names'].append(octets.encode('utf-8'))
return hostnames
host_names = {
'cn': cn.value.lower(),
'dns_names': ext.value.get_values_for_type(x509.DNSName)
}
return host_names
except Exception:
raise exceptions.UnreadableCert
def _get_x509_from_pem_bytes(certificate_pem):

4
octavia/common/utils.py
View File

@ -43,7 +43,9 @@ def get_random_string(length):
rndstr = ""
random.seed(datetime.datetime.now().microsecond)
while len(rndstr) < length:
rndstr += hashlib.sha224(str(random.random())).hexdigest()
rndstr += hashlib.sha224(
str(random.random()).encode('ascii')
).hexdigest()
return rndstr[0:length]

4
octavia/controller/worker/tasks/network_tasks.py
View File

@ -186,7 +186,7 @@ class UnPlugNetworks(BaseNetworkTask):
except Exception as e:
LOG.error(
_LE("Unable to unplug network - exception: %s"),
e.message)
str(e))
pass # Todo(german) follow up if that makes sense
@ -215,7 +215,7 @@ class HandleNetworkDeltas(BaseNetworkTask):
except Exception as e:
LOG.error(
_LE("Unable to unplug network - exception: %s"),
e.message)
str(e))
pass
def revert(self, deltas):

3
octavia/hacking/checks.py
View File

@ -15,6 +15,7 @@
import re
import pep8
import six
"""
Guidelines for writing new hacking checks
@ -47,7 +48,7 @@ _all_log_levels = {
'exception': '_LE',
}
log_translation_hints = []
for level, hint in _all_log_levels.iteritems():
for level, hint in six.iteritems(_all_log_levels):
r = "(.)*LOG\.%(level)s\(\s*((%(wrong_hints)s)\(|'|\")" % {
'level': level,
'wrong_hints': '|'.join(_all_hints - set([hint])),

2
octavia/network/drivers/neutron/allowed_address_pairs.py
View File

@ -174,7 +174,7 @@ class AllowedAddressPairsDriver(base.AbstractNetworkDriver):
except neutron_client_exceptions.PortNotFoundClient as e:
raise base.PortNotFound(e.message)
except Exception as e:
raise base.PlugVIPException(e.message)
raise base.PlugVIPException(str(e))
def _add_vip_security_group_to_amphorae(self, load_balancer_id, amphora):
sec_grp = self._get_lb_security_group(load_balancer_id)

9
octavia/tests/functional/api/v1/base.py
View File

@ -14,9 +14,9 @@
import logging
import mock
import pecan
import pecan.testing
import six
from octavia.api import config as pconfig
from octavia.common import constants
@ -24,6 +24,11 @@ from octavia.db import api as db_api
from octavia.db import repositories
from octavia.tests.functional.db import base as base_db_test
if six.PY2:
import mock
else:
import unittest.mock as mock
LOG = logging.getLogger(__name__)
@ -213,4 +218,4 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
self.assertEqual(provisioning_status,
api_listener.get('provisioning_status'))
self.assertEqual(operating_status,
api_listener.get('operating_status'))
api_listener.get('operating_status'))

14
octavia/tests/unit/amphorae/backends/health_daemon/test_config.py
View File

@ -34,13 +34,17 @@ class TestConfig(base.TestCase):
matchers.raises(IOError))
def test_config(self):
def check_update():
self.assertFalse(self.update_called)
self.update_called = True
cfg = config.JSONFileConfig()
cfg.set_filename(self.sampleconfig[1])
# Check the singleton decorator
self.assertIs(cfg, config.JSONFileConfig())
cfg.add_observer(self.check_update)
cfg.add_observer(check_update)
self.update_called = False
cfg.check_update()
@ -60,13 +64,13 @@ class TestConfig(base.TestCase):
self.assertIs(cfg['delay'], 5)
# Check for removing an observer - Thanks Stephen
cfg.remove_observer(self.check_update)
cfg.remove_observer(check_update)
self.update_called = False
cfg.check_update()
self.assertFalse(self.update_called)
# Better add it back for the next test
cfg.add_observer(self.check_update)
cfg.add_observer(check_update)
# Next, replace the file (new inode)
self.remove_config_file()
@ -80,10 +84,6 @@ class TestConfig(base.TestCase):
self.assertIs(cfg['delay'], 3)
def check_update(self):
self.assertFalse(self.update_called)
self.update_called = True
def setup_config_file(self):
self.sampleconfig = tempfile.mkstemp()
conffile = os.fdopen(self.sampleconfig[0], 'w+')

8
octavia/tests/unit/amphorae/drivers/haproxy/test_ssh_driver.py
View File

@ -11,10 +11,11 @@
# 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 oslo_log import log
from oslo_utils import uuidutils
import paramiko
import six
from octavia.amphorae.drivers.haproxy.jinja import jinja_cfg
from octavia.amphorae.drivers.haproxy import ssh_driver
@ -25,6 +26,11 @@ from octavia.db import models as models
from octavia.tests.unit import base
from octavia.tests.unit.common.sample_configs import sample_configs
if six.PY2:
import mock
else:
import unittest.mock as mock
LOG = log.getLogger(__name__)

7
octavia/tests/unit/api/v1/handlers/queue/test_producer.py
View File

@ -24,9 +24,9 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_config import fixture
import oslo_messaging as messaging
import six
from octavia.api.v1.handlers.queue import producer
from octavia.api.v1.types import health_monitor
@ -38,6 +38,11 @@ from octavia.common import config
from octavia.common import data_models
from octavia.tests.unit import base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestProducer(base.TestCase):
def setUp(self):

7
octavia/tests/unit/certificates/common/test_barbican.py
View File

@ -13,12 +13,17 @@
# under the License.
from barbicanclient import client as barbican_client
import mock
import six
import octavia.certificates.common.barbican as barbican_common
from octavia.common import keystone
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestBarbicanAuth(base.TestCase):

7
octavia/tests/unit/certificates/generator/test_barbican.py
View File

@ -13,12 +13,17 @@
# under the License.
import uuid
import mock
from OpenSSL import crypto
import six
import octavia.certificates.generator.barbican as barbican_cert_gen
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestBarbicanGenerator(base.TestCase):

11
octavia/tests/unit/certificates/generator/test_local.py
View File

@ -38,7 +38,7 @@ class TestLocalGenerator(base.TestCase):
ca_key = crypto.PKey()
ca_key.generate_key(crypto.TYPE_RSA, 2048)
self.ca_private_key_passphrase = "Testing"
self.ca_private_key_passphrase = b"Testing"
self.ca_private_key = crypto.dump_privatekey(
crypto.FILETYPE_PEM,
ca_key,
@ -77,14 +77,17 @@ class TestLocalGenerator(base.TestCase):
ca_digest=self.signing_digest
)
self.assertIn("-----BEGIN CERTIFICATE-----", signed_cert)
self.assertIn("-----BEGIN CERTIFICATE-----",
signed_cert.decode('ascii'))
# Load the cert for specific tests
cert = crypto.load_certificate(crypto.FILETYPE_PEM, signed_cert)
# Make sure expiry time is accurate
expires = datetime.datetime.strptime(cert.get_notAfter(),
'%Y%m%d%H%M%SZ')
expires = datetime.datetime.strptime(
cert.get_notAfter().decode('ascii'),
'%Y%m%d%H%M%SZ'
)
should_expire = (datetime.datetime.utcnow() +
datetime.timedelta(seconds=2 * 365 * 24 * 60 * 60))
diff = should_expire - expires

7
octavia/tests/unit/certificates/manager/test_barbican.py
View File

@ -15,13 +15,18 @@ import uuid
from barbicanclient import containers
from barbicanclient import secrets
import mock
import six
import octavia.certificates.common.barbican as barbican_common
import octavia.certificates.common.cert as cert
import octavia.certificates.manager.barbican as barbican_cert_mgr
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestBarbicanManager(base.TestCase):

20
octavia/tests/unit/certificates/manager/test_local.py
View File

@ -13,16 +13,26 @@
# under the License.
import os
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
import six
import octavia.certificates.common.cert as cert
import octavia.certificates.manager.local as local_cert_mgr
import octavia.tests.unit.base as base
if six.PY2:
import __builtin__ as builtins
class TestLocalGenerator(base.TestCase):
import mock
else:
import builtins
import unittest.mock as mock
class TestLocalManager(base.TestCase):
def setUp(self):
self.certificate = "My Certificate"
@ -33,12 +43,12 @@ class TestLocalGenerator(base.TestCase):
conf = oslo_fixture.Config(cfg.CONF)
conf.config(group="certificates", storage_path="/tmp/")
super(TestLocalGenerator, self).setUp()
super(TestLocalManager, self).setUp()
def _store_cert(self):
file_mock = mock.mock_open()
# Attempt to store the cert
with mock.patch('__builtin__.open', file_mock, create=True):
with mock.patch.object(builtins, 'open', file_mock):
cert_id = local_cert_mgr.LocalCertManager.store_cert(
certificate=self.certificate,
intermediates=self.intermediates,
@ -70,7 +80,7 @@ class TestLocalGenerator(base.TestCase):
def _get_cert(self, cert_id):
file_mock = mock.mock_open()
# Attempt to retrieve the cert
with mock.patch('__builtin__.open', file_mock, create=True):
with mock.patch.object(builtins, 'open', file_mock):
data = local_cert_mgr.LocalCertManager.get_cert(cert_id)
# Verify the correct files were opened

7
octavia/tests/unit/common/test_base_taskflow.py
View File

@ -14,14 +14,19 @@
#
import concurrent.futures
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
import six
from taskflow import engines as tf_engines
from octavia.common import base_taskflow
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
MAX_WORKERS = 1
_engine_mock = mock.MagicMock()

101
octavia/tests/unit/common/tls_utils/test_cert_parser.py
View File

@ -13,53 +13,54 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
import octavia.common.exceptions as exceptions
import octavia.common.tls_utils.cert_parser as cert_parser
from octavia.tests.unit import base
ALT_EXT_CRT = """-----BEGIN CERTIFICATE-----
MIIGxDCCBaygAwIBAgIGAUp0fCElMA0GCSqGSIb3DQEBDQUAMIGLMQswCQYDVQQG
EwJVUzEOMAwGA1UECAwFVGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4wHAYD
VQQKDBVPcGVuU3RhY2sgRXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24gTGJh
YXMxHjAcBgNVBAMMFXd3dy5DTkZyb21TdWJqZWN0Lm9yZzAeFw0xNDExMjIwMDEx
MzlaFw0yMjEyMjEwMDExMzlaMIGLMQswCQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4
YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4wHAYDVQQKDBVPcGVuU3RhY2sgRXhw
ZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24gTGJhYXMxHjAcBgNVBAMMFXd3dy5D
TkZyb21TdWJqZWN0Lm9yZzCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
ALL1nmbDPUDps84i1sM3rhHrc+Dlu0N/wKQWKZFeiWUtF/pot19V3o0yXDpsg7W5
RkLMTFkZEcnQpyGdpAGjTjzmNXMZw99EzxsmrR3l6hUEISifVbvEuftYZT6jPxM5
ML6WAjFNaBEZPWtZi8CgX5xdjdrDNndwyHob49n7Nc/h1kVqqBqMILabTqC6yEcx
S/B+DugVuuYbEdYYYElQUMfM+mUdULrSqIVl2n5AvvSFjWzWzfgPyp4QKn+f7HVR
T62bh/XjQ88n1tMYNAEqixRZTPgqY1LFl9VJVgRp9fdL6ttMurOR3C0STJ5qCdKB
L7LrpbY4u8dEragRC6YAyI8CAwEAAaOCAyowggMmMAwGA1UdEwEB/wQCMAAwDgYD
VR0PAQH/BAQDAgO4MIIDBAYDVR0RBIIC+zCCAveCGXd3dy5ob3N0RnJvbV9kTlNO
YW1lMS5jb22CGXd3dy5ob3N0RnJvbV9kTlNOYW1lMi5jb22CGXd3dy5ob3N0RnJv
bV9kTlNOYW1lMy5jb22BEW5vb25lQG5vd2hlcmUub3JnpIGPMIGMMQswCQYDVQQG
EwJVUzEOMAwGA1UECAwFVGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4wHAYD
VQQKDBVPcGVuU3RhY2sgRXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24gTGJh
YXMxHzAdBgNVBAMMFnd3dy5jbkZyb21BbHROYW1lMS5vcmekgY8wgYwxCzAJBgNV
BAYTAlVTMQ4wDAYDVQQIDAVUZXhhczEUMBIGA1UEBwwLU2FuIEFudG9uaW8xHjAc
BgNVBAoMFU9wZW5TdGFjayBFeHBlcmltZW50czEWMBQGA1UECwwNTmV1dHJvbiBM
YmFhczEfMB0GA1UEAwwWd3d3LmNuRnJvbUFsdE5hbWUyLm9yZ6SBjzCBjDELMAkG
A1UEBhMCVVMxDjAMBgNVBAgMBVRleGFzMRQwEgYDVQQHDAtTYW4gQW50b25pbzEe
MBwGA1UECgwVT3BlblN0YWNrIEV4cGVyaW1lbnRzMRYwFAYDVQQLDA1OZXV0cm9u
IExiYWFzMR8wHQYDVQQDDBZ3d3cuY25Gcm9tQWx0TmFtZTMub3JnpIGPMIGMMQsw
CQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlv
MR4wHAYDVQQKDBVPcGVuU3RhY2sgRXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRy
b24gTGJhYXMxHzAdBgNVBAMMFnd3dy5jbkZyb21BbHROYW1lNC5vcmeHBAoBAgOH
EAEjRWeJq83v97PVkeaixICGFmh0dHA6Ly93d3cuZXhhbXBsZS5jb22CGXd3dy5o
b3N0RnJvbV9kTlNOYW1lNC5jb20wDQYJKoZIhvcNAQENBQADggEBAICUCDMhDf0f
cvX5mVnq4Q3+SM/nl03Gse6J0JdpFivS4hl+uZs0TAFYpfEPpAa7KKxD229kbCiQ
kyxf8fzADSl77RQbL6Lxa8K/c66mVNiuVvTHV4r/nDNcRYN9fGArw/Ho7VX+HVQ6
UW1t/uvXeyg695t7kzZmvg0ChD5kS848d2rXu2MhwHwXA8rbuK6gxVY97fbzBNlj
aiPJUAb8lqZMShd+3yVCgMmV0J20u2b5pSdO+LHQ7NfVqURk2pcHD8slfHzXT58q
YB90v0pSVP6mzHGyLxETZZz0nhaH9EjOyFkQI84ORT8Kmd5Y04gSI0LTKKF1eMNE
TyNC+MtsRdA=
ALT_EXT_CRT = b"""-----BEGIN CERTIFICATE-----
MIIGqjCCBZKgAwIBAgIJAIApBg8slSSiMA0GCSqGSIb3DQEBBQUAMIGLMQswCQYD
VQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4w
HAYDVQQKDBVPcGVuU3RhY2sgRXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24g
TGJhYXMxHjAcBgNVBAMMFXd3dy5DTkZyb21TdWJqZWN0Lm9yZzAeFw0xNTA1MjEy
MDMzMjNaFw0yNTA1MTgyMDMzMjNaMIGLMQswCQYDVQQGEwJVUzEOMAwGA1UECAwF
VGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4wHAYDVQQKDBVPcGVuU3RhY2sg
RXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24gTGJhYXMxHjAcBgNVBAMMFXd3
dy5DTkZyb21TdWJqZWN0Lm9yZzCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoC
ggEBALL1nmbDPUDps84i1sM3rhHrc+Dlu0N/wKQWKZFeiWUtF/pot19V3o0yXDps
g7W5RkLMTFkZEcnQpyGdpAGjTjzmNXMZw99EzxsmrR3l6hUEISifVbvEuftYZT6j
PxM5ML6WAjFNaBEZPWtZi8CgX5xdjdrDNndwyHob49n7Nc/h1kVqqBqMILabTqC6
yEcxS/B+DugVuuYbEdYYYElQUMfM+mUdULrSqIVl2n5AvvSFjWzWzfgPyp4QKn+f
7HVRT62bh/XjQ88n1tMYNAEqixRZTPgqY1LFl9VJVgRp9fdL6ttMurOR3C0STJ5q
CdKBL7LrpbY4u8dEragRC6YAyI8CAwEAAaOCAw0wggMJMAkGA1UdEwQCMAAwCwYD
VR0PBAQDAgXgMIIC7QYDVR0RBIIC5DCCAuCCGHd3dy5ob3N0RnJvbUROU05hbWUx
LmNvbYIYd3d3Lmhvc3RGcm9tRE5TTmFtZTIuY29tghh3d3cuaG9zdEZyb21ETlNO
YW1lMy5jb22CGHd3dy5ob3N0RnJvbUROU05hbWU0LmNvbYcECgECA4cQASNFZ4mr
ze/3s9WR5qLEgIYWaHR0cDovL3d3dy5leGFtcGxlLmNvbaSBjzCBjDELMAkGA1UE
BhMCVVMxDjAMBgNVBAgMBVRleGFzMRQwEgYDVQQHDAtTYW4gQW50b25pbzEeMBwG
A1UECgwVT3BlblN0YWNrIEV4cGVyaW1lbnRzMRYwFAYDVQQLDA1OZXV0cm9uIExi
YWFzMR8wHQYDVQQDDBZ3d3cuY25Gcm9tQWx0TmFtZTEub3JnpIGPMIGMMQswCQYD
VQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxFDASBgNVBAcMC1NhbiBBbnRvbmlvMR4w
HAYDVQQKDBVPcGVuU3RhY2sgRXhwZXJpbWVudHMxFjAUBgNVBAsMDU5ldXRyb24g
TGJhYXMxHzAdBgNVBAMMFnd3dy5jbkZyb21BbHROYW1lMi5vcmekgY8wgYwxCzAJ
BgNVBAYTAlVTMQ4wDAYDVQQIDAVUZXhhczEUMBIGA1UEBwwLU2FuIEFudG9uaW8x
HjAcBgNVBAoMFU9wZW5TdGFjayBFeHBlcmltZW50czEWMBQGA1UECwwNTmV1dHJv
biBMYmFhczEfMB0GA1UEAwwWd3d3LmNuRnJvbUFsdE5hbWUzLm9yZ6SBjzCBjDEL
MAkGA1UEBhMCVVMxDjAMBgNVBAgMBVRleGFzMRQwEgYDVQQHDAtTYW4gQW50b25p
bzEeMBwGA1UECgwVT3BlblN0YWNrIEV4cGVyaW1lbnRzMRYwFAYDVQQLDA1OZXV0
cm9uIExiYWFzMR8wHQYDVQQDDBZ3d3cuY25Gcm9tQWx0TmFtZTQub3JnMA0GCSqG
SIb3DQEBBQUAA4IBAQCS6iDn6R3C+qJLZibaqrBSkM9yu5kwRsQ6lQ+DODvVYGWq
eGkkh5o2c6WbJlH44yF280+HvnJcuISD7epPHJN0vUM9+WMtXfEli9avFHgu2JxP
3P0ixK2kaJnqKQkSEdnA/v/eWP1Cd2v6rbKCIo9d2gSP0cnpdtlX9Zk3SzEh0V7s
RjSdfZoAvz0aAnpDHlTerLcz5T2aiRae2wSt/RLA3qDO1Ji05tWvQBmKuepxS6A1
tL4Drm+OCXJwTrE7ClTMCwcrZnLl4tI+Z+X3DV92WQB8ldST/QFjz1hgs/4zrADA
elu2c/X7MR4ObOjhDfaVGQ8kMhYf5hx69qyNDsGi
-----END CERTIFICATE-----
"""
SOME_OTHER_RSA_KEY = """
SOME_OTHER_RSA_KEY = b"""
-----BEGIN RSA PRIVATE KEY-----
MIICWwIBAAKBgQDDnJL9dAdDpjoq4tksTJmdM0AjIHa7Y2yc8XwU7YkgrOR0m4Po
r7El0NwWf5i/LFudX1cOkfwemMIPwQ+67k0BVu/W3SR+g9ZzVKZtTBJnDoqMZ4RJ
@ -77,7 +78,7 @@ q3NRSsf/nRLY1NtMdwJAVKOdUCwZKGpGyOUZPRbZZAPlojIff2CxJ6E2Pr0RbShD
-----END RSA PRIVATE KEY-----
"""
ALT_EXT_CRT_KEY = """
ALT_EXT_CRT_KEY = b"""
-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEAsvWeZsM9QOmzziLWwzeuEetz4OW7Q3/ApBYpkV6JZS0X+mi3
X1XejTJcOmyDtblGQsxMWRkRydCnIZ2kAaNOPOY1cxnD30TPGyatHeXqFQQhKJ9V
@ -109,7 +110,7 @@ iMwJYgm98P27s4TEMdhlPNVJrj1FrD+4VrgpOsoM20EkZnTvel9s
ENCRYPTED_PKCS8_CRT_KEY_PASSPHRASE = "test_passphrase"
ENCRYPTED_PKCS8_CRT_KEY = """-----BEGIN ENCRYPTED PRIVATE KEY-----
ENCRYPTED_PKCS8_CRT_KEY = b"""-----BEGIN ENCRYPTED PRIVATE KEY-----
MIIE6TAbBgkqhkiG9w0BBQMwDgQIT04zko6pmJICAggABIIEyL/79sqzTQ7BsEjY
ao2Uhh3//mpNJfCDhjSZOmWL7s4+161cEqpxrfxo4bHH8fkZ60VZUQP8CjwwQUhP
4iwpv2bYbQwzlttZwTC6s28wh7FRtgVoVPTwvXJa6fl2zAjLtsjwLZ/556ez9xIJ
@ -140,7 +141,7 @@ p7cuYY1cAyI7gFfl5A==
-----END ENCRYPTED PRIVATE KEY-----
"""
UNENCRYPTED_PKCS8_CRT_KEY = """-----BEGIN PRIVATE KEY-----
UNENCRYPTED_PKCS8_CRT_KEY = b"""-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCy9Z5mwz1A6bPO
ItbDN64R63Pg5btDf8CkFimRXollLRf6aLdfVd6NMlw6bIO1uUZCzExZGRHJ0Kch
naQBo0485jVzGcPfRM8bJq0d5eoVBCEon1W7xLn7WGU+oz8TOTC+lgIxTWgRGT1r
@ -172,7 +173,7 @@ P7hWuCk6ygzbQSRmdO96X2w=
EXPECTED_IMD_SUBJS = ["IMD3", "IMD2", "IMD1"]
X509_IMDS = """Junk
X509_IMDS = b"""Junk
-----BEGIN CERTIFICATE-----
MIIBhDCCAS6gAwIBAgIGAUo7hO/eMA0GCSqGSIb3DQEBCwUAMA8xDTALBgNVBAMT
BElNRDIwHhcNMTQxMjExMjI0MjU1WhcNMjUxMTIzMjI0MjU1WjAPMQ0wCwYDVQQD
@ -213,11 +214,11 @@ uJIQ
class TestTLSParseUtils(base.TestCase):
def test_alt_subject_name_parses(self):
hosts = cert_parser.get_host_names(ALT_EXT_CRT)
self.assertEqual(hosts['cn'], 'www.CNFromSubject.org')
self.assertEqual(hosts['dns_names'][0], 'www.hostFrom_dNSName1.com')
self.assertEqual(hosts['dns_names'][1], 'www.hostFrom_dNSName2.com')
self.assertEqual(hosts['dns_names'][2], 'www.hostFrom_dNSName3.com')
self.assertEqual(hosts['dns_names'][3], 'www.hostFrom_dNSName4.com')
self.assertIn('www.cnfromsubject.org', hosts['cn'])
self.assertIn('www.hostfromdnsname1.com', hosts['dns_names'])
self.assertIn('www.hostfromdnsname2.com', hosts['dns_names'])
self.assertIn('www.hostfromdnsname3.com', hosts['dns_names'])
self.assertIn('www.hostfromdnsname4.com', hosts['dns_names'])
def test_x509_parses(self):
self.assertRaises(exceptions.UnreadableCert,
@ -252,5 +253,5 @@ class TestTLSParseUtils(base.TestCase):
for x509Pem in cert_parser._split_x509s(X509_IMDS):
imds.append(cert_parser._get_x509_from_pem_bytes(x509Pem))
for i in xrange(0, len(imds)):
self.assertEqual(EXPECTED_IMD_SUBJS[i], imds[i].get_subject().CN)
for i in six.moves.xrange(0, len(imds)):
self.assertEqual(EXPECTED_IMD_SUBJS[i], imds[i].get_subject().CN)

6
octavia/tests/unit/compute/drivers/test_nova_driver.py
View File

@ -13,10 +13,10 @@
# under the License.
from keystoneclient import session
import mock
import novaclient.v2
from oslo_config import cfg
from oslo_utils import uuidutils
import six
from octavia.common import constants
from octavia.common import data_models as models
@ -25,6 +25,10 @@ from octavia.common import keystone
import octavia.compute.drivers.nova_driver as nova_common
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
CONF = cfg.CONF

7
octavia/tests/unit/controller/queue/test_consumer.py
View File

@ -12,14 +12,19 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_config import cfg
import oslo_messaging as messaging
import six
from octavia.controller.queue import consumer
from octavia.controller.queue import endpoint
from octavia.tests.unit import base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestConsumer(base.TestCase):

7
octavia/tests/unit/controller/queue/test_endpoint.py
View File

@ -12,12 +12,17 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
import six
from octavia.controller.queue import endpoint
from octavia.controller.worker import controller_worker
from octavia.tests.unit import base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestEndpoint(base.TestCase):

19
octavia/tests/unit/controller/test_health_mixin.py
View File

@ -12,13 +12,18 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_utils import uuidutils
import six
from octavia.common import constants
from octavia.controller.healthmanager import update_health_mixin as healthmixin
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestUpdateHealthMixin(base.TestCase):
FAKE_UUID_1 = uuidutils.generate_uuid()
@ -59,12 +64,14 @@ class TestUpdateHealthMixin(base.TestCase):
self.assertTrue(self.amphora_health_repo.update.called)
# test listener, member
for listener_id, listener in health.get('listeners', {}).iteritems():
for listener_id, listener in six.iteritems(
health.get('listeners', {})):
self.listener_repo.update.assert_any_call(
'blah', listener_id, operating_status=constants.ONLINE)
for member_id, member in listener.get('members', {}).iteritems():
for member_id, member in six.iteritems(
listener.get('members', {})):
self.member_repo.update.assert_any_call(
'blah', id=member_id, operating_status=constants.ONLINE)
@ -92,11 +99,13 @@ class TestUpdateHealthMixin(base.TestCase):
self.assertFalse(self.amphora_health_repo.update.called)
# test listener, member
for listener_id, listener in health.get('listeners', {}).iteritems():
for listener_id, listener in six.iteritems(
health.get('listeners', {})):
self.listener_repo.update.assert_any_call(
'blah', listener_id, operating_status=constants.ERROR)
for member_id, member in listener.get('members', {}).iteritems():
for member_id, member in six.iteritems(
listener.get('members', {})):
self.member_repo.update.assert_any_call(
'blah', id=member_id, operating_status=constants.ERROR)

7
octavia/tests/unit/controller/worker/tasks/test_amphora_driver_tasks.py
View File

@ -13,14 +13,19 @@
# under the License.
#
import mock
from oslo_utils import uuidutils
import six
from octavia.common import constants
from octavia.controller.worker.tasks import amphora_driver_tasks
from octavia.db import repositories as repo
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMP_ID = uuidutils.generate_uuid()
LISTENER_ID = uuidutils.generate_uuid()
LB_ID = uuidutils.generate_uuid()

9
octavia/tests/unit/controller/worker/tasks/test_compute_tasks.py
View File

@ -15,16 +15,21 @@
import time
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from oslo_utils import uuidutils
import six
from octavia.common import constants
from octavia.common import exceptions
from octavia.controller.worker.tasks import compute_tasks
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMP_FLAVOR_ID = 10
AMP_IMAGE_ID = 11
AMP_SSH_KEY_NAME = None
@ -152,4 +157,4 @@ class TestComputeTasks(base.TestCase):
delete_compute = compute_tasks.ComputeDelete()
delete_compute.execute(COMPUTE_ID)
mock_driver.delete.assert_called_once_with(amphora_id=COMPUTE_ID)
mock_driver.delete.assert_called_once_with(amphora_id=COMPUTE_ID)

7
octavia/tests/unit/controller/worker/tasks/test_controller_tasks.py
View File

@ -13,14 +13,19 @@
# under the License.
#
import mock
from oslo_utils import uuidutils
import six
from octavia.controller.worker import controller_worker
from octavia.controller.worker.tasks import controller_tasks
from octavia.db import repositories as repo
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMP_ID = uuidutils.generate_uuid()
LB1_ID = uuidutils.generate_uuid()
LB2_ID = uuidutils.generate_uuid()

7
octavia/tests/unit/controller/worker/tasks/test_database_tasks.py
View File

@ -13,8 +13,8 @@
# under the License.
#
import mock
from oslo_utils import uuidutils
import six
from taskflow.types import failure
from octavia.common import constants
@ -23,6 +23,11 @@ from octavia.controller.worker.tasks import database_tasks
from octavia.db import repositories as repo
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMP_ID = uuidutils.generate_uuid()
COMPUTE_ID = uuidutils.generate_uuid()
HM_ID = uuidutils.generate_uuid()

7
octavia/tests/unit/controller/worker/tasks/test_model_tasks.py
View File

@ -13,11 +13,16 @@
# under the License.
#
import mock
import six
from octavia.controller.worker.tasks import model_tasks
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
class TestObjectUpdateTasks(base.TestCase):

7
octavia/tests/unit/controller/worker/tasks/test_network_tasks.py
View File

@ -13,10 +13,10 @@
# under the License.
#
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from oslo_utils import uuidutils
import six
from octavia.common import data_models as o_data_models
from octavia.controller.worker.tasks import network_tasks
@ -24,6 +24,11 @@ from octavia.network import base as net_base
from octavia.network import data_models
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMPHORA_ID = 7
COMPUTE_ID = uuidutils.generate_uuid()

7
octavia/tests/unit/controller/worker/test_controller_worker.py
View File

@ -13,8 +13,8 @@
# under the License.
#
import mock
from oslo_utils import uuidutils
import six
from octavia.common import base_taskflow
from octavia.common import constants
@ -22,6 +22,11 @@ from octavia.common import exceptions
from octavia.controller.worker import controller_worker
import octavia.tests.unit.base as base
if six.PY2:
import mock
else:
import unittest.mock as mock
AMP_ID = uuidutils.generate_uuid()
LB_ID = uuidutils.generate_uuid()
POOL_ID = uuidutils.generate_uuid()

1
requirements.txt
View File

@ -45,3 +45,4 @@ networkx>=1.8
Flask<1.0,>=0.10
netifaces>=0.10.4
cryptography>=0.9

2
tox.ini
View File

@ -1,6 +1,6 @@
[tox]
minversion = 1.6
envlist = docs,py27,pep8,specs
envlist = docs,py27,py34,pep8,specs
skipsdist = True
[testenv]