openstack
/
octavia
Code Issues Proposed changes
octavia/octavia/tests/unit/db/test_models.py

764 lines
32 KiB
Python
Raw Blame History

# Copyright 2014 Rackspace
#
# 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 octavia.common import constants
from octavia.common import data_models
from octavia.db import models
from octavia.tests.unit.db import base
class ModelTestMixin(object):
FAKE_UUID_1 = '0123456789012345678901234567890123456'
FAKE_UUID_2 = '1234567890123456789012345678901234567'
def _insert(self, session, model_cls, model_kwargs):
with session.begin():
model = model_cls(**model_kwargs)
session.add(model)
return model
def associate_amphora(self, load_balancer, amphora):
load_balancer.amphorae.append(amphora)
def create_listener(self, session, **overrides):
kwargs = {'tenant_id': self.FAKE_UUID_1,
'id': self.FAKE_UUID_1,
'protocol': constants.PROTOCOL_HTTP,
'protocol_port': 80,
'provisioning_status': constants.ACTIVE,
'operating_status': constants.ONLINE,
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.Listener, kwargs)
def create_listener_statistics(self, session, listener_id, **overrides):
kwargs = {'listener_id': listener_id,
'bytes_in': 0,
'bytes_out': 0,
'active_connections': 0,
'total_connections': 0}
kwargs.update(overrides)
return self._insert(session, models.ListenerStatistics, kwargs)
def create_pool(self, session, **overrides):
kwargs = {'tenant_id': self.FAKE_UUID_1,
'id': self.FAKE_UUID_1,
'protocol': constants.PROTOCOL_HTTP,
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN,
'operating_status': constants.ONLINE,
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.Pool, kwargs)
def create_session_persistence(self, session, pool_id, **overrides):
kwargs = {'pool_id': pool_id,
'type': constants.SESSION_PERSISTENCE_HTTP_COOKIE}
kwargs.update(overrides)
return self._insert(session, models.SessionPersistence, kwargs)
def create_health_monitor(self, session, pool_id, **overrides):
kwargs = {'tenant_id': self.FAKE_UUID_1,
'id': self.FAKE_UUID_1,
'pool_id': pool_id,
'type': constants.HEALTH_MONITOR_HTTP,
'delay': 1,
'timeout': 1,
'fall_threshold': 1,
'rise_threshold': 1,
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.HealthMonitor, kwargs)
def create_member(self, session, pool_id, **overrides):
kwargs = {'tenant_id': self.FAKE_UUID_1,
'id': self.FAKE_UUID_1,
'pool_id': pool_id,
'ip_address': '10.0.0.1',
'protocol_port': 80,
'operating_status': constants.ONLINE,
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.Member, kwargs)
def create_load_balancer(self, session, **overrides):
kwargs = {'tenant_id': self.FAKE_UUID_1,
'id': self.FAKE_UUID_1,
'provisioning_status': constants.ACTIVE,
'operating_status': constants.ONLINE,
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.LoadBalancer, kwargs)
def create_vip(self, session, load_balancer_id, **overrides):
kwargs = {'load_balancer_id': load_balancer_id}
kwargs.update(overrides)
return self._insert(session, models.Vip, kwargs)
def create_sni(self, session, **overrides):
kwargs = {'listener_id': self.FAKE_UUID_1,
'tls_container_id': self.FAKE_UUID_1}
kwargs.update(overrides)
return self._insert(session, models.SNI, kwargs)
def create_amphora(self, session, **overrides):
kwargs = {'id': self.FAKE_UUID_1,
'host_id': self.FAKE_UUID_1,
'status': constants.ONLINE}
kwargs.update(overrides)
return self._insert(session, models.Amphora, kwargs)
class PoolModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def test_create(self):
self.create_pool(self.session)
def test_update(self):
pool = self.create_pool(self.session)
id = pool.id
pool.name = 'test1'
new_pool = self.session.query(
models.Pool).filter_by(id=id).first()
self.assertEqual('test1', new_pool.name)
def test_delete(self):
pool = self.create_pool(self.session)
id = pool.id
with self.session.begin():
self.session.delete(pool)
self.session.flush()
new_pool = self.session.query(
models.Pool).filter_by(id=id).first()
self.assertIsNone(new_pool)
def test_member_relationship(self):
pool = self.create_pool(self.session)
self.create_member(self.session, pool.id, id=self.FAKE_UUID_1,
ip_address="10.0.0.1")
self.create_member(self.session, pool.id, id=self.FAKE_UUID_2,
ip_address="10.0.0.2")
new_pool = self.session.query(
models.Pool).filter_by(id=pool.id).first()
self.assertIsNotNone(new_pool.members)
self.assertEqual(2, len(new_pool.members))
self.assertTrue(isinstance(new_pool.members[0], models.Member))
def test_health_monitor_relationship(self):
pool = self.create_pool(self.session)
self.create_health_monitor(self.session, pool.id)
new_pool = self.session.query(models.Pool).filter_by(
id=pool.id).first()
self.assertIsNotNone(new_pool.health_monitor)
self.assertTrue(isinstance(new_pool.health_monitor,
models.HealthMonitor))
def test_session_persistence_relationship(self):
pool = self.create_pool(self.session)
self.create_session_persistence(self.session, pool_id=pool.id)
new_pool = self.session.query(models.Pool).filter_by(
id=pool.id).first()
self.assertIsNotNone(new_pool.session_persistence)
self.assertTrue(isinstance(new_pool.session_persistence,
models.SessionPersistence))
def test_listener_relationship(self):
pool = self.create_pool(self.session)
self.create_listener(self.session, default_pool_id=pool.id)
new_pool = self.session.query(models.Pool).filter_by(
id=pool.id).first()
self.assertIsNotNone(new_pool.listener)
self.assertTrue(isinstance(new_pool.listener, models.Listener))
class MemberModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(MemberModelTest, self).setUp()
self.pool = self.create_pool(self.session)
def test_create(self):
self.create_member(self.session, self.pool.id)
def test_update(self):
member = self.create_member(self.session, self.pool.id)
member_id = member.id
member.name = 'test1'
new_member = self.session.query(
models.Member).filter_by(id=member_id).first()
self.assertEqual('test1', new_member.name)
def test_delete(self):
member = self.create_member(self.session, self.pool.id)
member_id = member.id
with self.session.begin():
self.session.delete(member)
self.session.flush()
new_member = self.session.query(
models.Member).filter_by(id=member_id).first()
self.assertIsNone(new_member)
def test_pool_relationship(self):
member = self.create_member(self.session, self.pool.id,
id=self.FAKE_UUID_1,
ip_address="10.0.0.1")
self.create_member(self.session, self.pool.id, id=self.FAKE_UUID_2,
ip_address="10.0.0.2")
new_member = self.session.query(models.Member).filter_by(
id=member.id).first()
self.assertIsNotNone(new_member.pool)
self.assertTrue(isinstance(new_member.pool, models.Pool))
class SessionPersistenceModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(SessionPersistenceModelTest, self).setUp()
self.pool = self.create_pool(self.session)
def test_create(self):
self.create_session_persistence(self.session, self.pool.id)
def test_update(self):
session_persistence = self.create_session_persistence(self.session,
self.pool.id)
session_persistence.name = 'test1'
new_session_persistence = self.session.query(
models.SessionPersistence).filter_by(pool_id=self.pool.id).first()
self.assertEqual('test1', new_session_persistence.name)
def test_delete(self):
session_persistence = self.create_session_persistence(self.session,
self.pool.id)
with self.session.begin():
self.session.delete(session_persistence)
self.session.flush()
new_session_persistence = self.session.query(
models.SessionPersistence).filter_by(pool_id=self.pool.id).first()
self.assertIsNone(new_session_persistence)
def test_pool_relationship(self):
self.create_session_persistence(self.session, self.pool.id)
new_persistence = self.session.query(
models.SessionPersistence).filter_by(pool_id=self.pool.id).first()
self.assertIsNotNone(new_persistence.pool)
self.assertTrue(isinstance(new_persistence.pool, models.Pool))
class ListenerModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def test_create(self):
self.create_listener(self.session)
def test_update(self):
listener = self.create_listener(self.session)
listener_id = listener.id
listener.name = 'test1'
new_listener = self.session.query(
models.Listener).filter_by(id=listener_id).first()
self.assertEqual('test1', new_listener.name)
def test_delete(self):
listener = self.create_listener(self.session)
listener_id = listener.id
with self.session.begin():
self.session.delete(listener)
self.session.flush()
new_listener = self.session.query(
models.Listener).filter_by(id=listener_id).first()
self.assertIsNone(new_listener)
def test_load_balancer_relationship(self):
lb = self.create_load_balancer(self.session)
listener = self.create_listener(self.session, load_balancer_id=lb.id)
new_listener = self.session.query(
models.Listener).filter_by(id=listener.id).first()
self.assertIsNotNone(new_listener.load_balancer)
self.assertTrue(isinstance(new_listener.load_balancer,
models.LoadBalancer))
def test_listener_statistics_relationship(self):
listener = self.create_listener(self.session)
self.create_listener_statistics(self.session, listener_id=listener.id)
new_listener = self.session.query(models.Listener).filter_by(
id=listener.id).first()
self.assertIsNotNone(new_listener.stats)
self.assertTrue(isinstance(new_listener.stats,
models.ListenerStatistics))
def test_pool_relationship(self):
pool = self.create_pool(self.session)
listener = self.create_listener(self.session, default_pool_id=pool.id)
new_listener = self.session.query(models.Listener).filter_by(
id=listener.id).first()
self.assertIsNotNone(new_listener.default_pool)
self.assertTrue(isinstance(new_listener.default_pool, models.Pool))
def test_sni_relationship(self):
listener = self.create_listener(self.session)
self.create_sni(self.session, listener_id=listener.id,
tls_container_id=self.FAKE_UUID_1)
self.create_sni(self.session, listener_id=listener.id,
tls_container_id=self.FAKE_UUID_2)
new_listener = self.session.query(models.Listener).filter_by(
id=listener.id).first()
self.assertIsNotNone(new_listener.sni_containers)
self.assertEqual(2, len(new_listener.sni_containers))
class ListenerStatisticsModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(ListenerStatisticsModelTest, self).setUp()
self.listener = self.create_listener(self.session)
def test_create(self):
self.create_listener_statistics(self.session, self.listener.id)
def test_update(self):
stats = self.create_listener_statistics(self.session, self.listener.id)
stats.name = 'test1'
new_stats = self.session.query(models.ListenerStatistics).filter_by(
listener_id=self.listener.id).first()
self.assertEqual('test1', new_stats.name)
def test_delete(self):
stats = self.create_listener_statistics(self.session, self.listener.id)
with self.session.begin():
self.session.delete(stats)
self.session.flush()
new_stats = self.session.query(models.ListenerStatistics).filter_by(
listener_id=self.listener.id).first()
self.assertIsNone(new_stats)
def test_listener_relationship(self):
self.create_listener_statistics(self.session, self.listener.id)
new_stats = self.session.query(models.ListenerStatistics).filter_by(
listener_id=self.listener.id).first()
self.assertIsNotNone(new_stats.listener)
self.assertTrue(isinstance(new_stats.listener, models.Listener))
class HealthMonitorModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(HealthMonitorModelTest, self).setUp()
self.pool = self.create_pool(self.session)
def test_create(self):
self.create_health_monitor(self.session, self.pool.id)
def test_update(self):
health_monitor = self.create_health_monitor(self.session, self.pool.id)
health_monitor_id = health_monitor.id
health_monitor.name = 'test1'
new_health_monitor = self.session.query(
models.HealthMonitor).filter_by(id=health_monitor_id).first()
self.assertEqual('test1', new_health_monitor.name)
def test_delete(self):
health_monitor = self.create_health_monitor(self.session, self.pool.id)
health_monitor_id = health_monitor.id
with self.session.begin():
self.session.delete(health_monitor)
self.session.flush()
new_health_monitor = self.session.query(
models.HealthMonitor).filter_by(id=health_monitor_id).first()
self.assertIsNone(new_health_monitor)
def test_pool_relationship(self):
health_monitor = self.create_health_monitor(self.session, self.pool.id)
new_health_monitor = self.session.query(
models.HealthMonitor).filter_by(id=health_monitor.id).first()
self.assertIsNotNone(new_health_monitor.pool)
self.assertTrue(isinstance(new_health_monitor.pool, models.Pool))
class LoadBalancerModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def test_create(self):
self.create_load_balancer(self.session)
def test_update(self):
load_balancer = self.create_load_balancer(self.session)
lb_id = load_balancer.id
load_balancer.name = 'test1'
new_load_balancer = self.session.query(
models.LoadBalancer).filter_by(id=lb_id).first()
self.assertEqual('test1', new_load_balancer.name)
def test_delete(self):
load_balancer = self.create_load_balancer(self.session)
lb_id = load_balancer.id
with self.session.begin():
self.session.delete(load_balancer)
self.session.flush()
new_load_balancer = self.session.query(
models.LoadBalancer).filter_by(id=lb_id).first()
self.assertIsNone(new_load_balancer)
def test_listener_relationship(self):
load_balancer = self.create_load_balancer(self.session)
self.create_listener(self.session, load_balancer_id=load_balancer.id)
new_load_balancer = self.session.query(
models.LoadBalancer).filter_by(id=load_balancer.id).first()
self.assertIsNotNone(new_load_balancer.listeners)
self.assertEqual(1, len(new_load_balancer.listeners))
def test_load_balancer_amphora_relationship(self):
load_balancer = self.create_load_balancer(self.session)
amphora = self.create_amphora(self.session)
self.associate_amphora(load_balancer, amphora)
new_load_balancer = self.session.query(
models.LoadBalancer).filter_by(id=load_balancer.id).first()
self.assertIsNotNone(new_load_balancer.amphorae)
self.assertEqual(1, len(new_load_balancer.amphorae))
def test_load_balancer_vip_relationship(self):
load_balancer = self.create_load_balancer(self.session)
self.create_vip(self.session, load_balancer.id)
new_load_balancer = self.session.query(
models.LoadBalancer).filter_by(id=load_balancer.id).first()
self.assertIsNotNone(new_load_balancer.vip)
self.assertIsInstance(new_load_balancer.vip, models.Vip)
class VipModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(VipModelTest, self).setUp()
self.load_balancer = self.create_load_balancer(self.session)
def test_create(self):
self.create_vip(self.session, self.load_balancer.id)
def test_update(self):
vip = self.create_vip(self.session, self.load_balancer.id)
vip.ip_address = "10.0.0.1"
new_vip = self.session.query(models.Vip).filter_by(
load_balancer_id=self.load_balancer.id).first()
self.assertEqual("10.0.0.1", new_vip.ip_address)
def test_delete(self):
vip = self.create_vip(self.session, self.load_balancer.id)
with self.session.begin():
self.session.delete(vip)
self.session.flush()
new_vip = self.session.query(models.Vip).filter_by(
load_balancer_id=vip.load_balancer_id).first()
self.assertIsNone(new_vip)
def test_vip_load_balancer_relationship(self):
self.create_vip(self.session, self.load_balancer.id)
new_vip = self.session.query(models.Vip).filter_by(
load_balancer_id=self.load_balancer.id).first()
self.assertIsNotNone(new_vip.load_balancer)
self.assertTrue(isinstance(new_vip.load_balancer, models.LoadBalancer))
class SNIModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(SNIModelTest, self).setUp()
self.listener = self.create_listener(self.session)
def test_create(self):
self.create_sni(self.session, listener_id=self.listener.id)
def test_update(self):
sni = self.create_sni(self.session, listener_id=self.listener.id)
sni.listener_id = self.FAKE_UUID_2
new_sni = self.session.query(
models.SNI).filter_by(listener_id=self.FAKE_UUID_2).first()
self.assertEqual(self.FAKE_UUID_2, new_sni.listener_id)
def test_delete(self):
sni = self.create_sni(self.session, listener_id=self.listener.id)
with self.session.begin():
self.session.delete(sni)
self.session.flush()
new_sni = self.session.query(
models.SNI).filter_by(listener_id=self.listener.id).first()
self.assertIsNone(new_sni)
def test_sni_relationship(self):
self.create_sni(self.session, listener_id=self.listener.id)
new_sni = self.session.query(models.SNI).filter_by(
listener_id=self.listener.id).first()
self.assertIsNotNone(new_sni.listener)
self.assertTrue(isinstance(new_sni.listener, models.Listener))
class AmphoraModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(AmphoraModelTest, self).setUp()
self.load_balancer = self.create_load_balancer(self.session)
def test_create(self):
self.create_amphora(self.session)
def test_update(self):
amphora = self.create_amphora(
self.session)
amphora.amphora_id = self.FAKE_UUID_2
new_amphora = self.session.query(models.Amphora).filter_by(
id=amphora.id).first()
self.assertEqual(self.FAKE_UUID_2, new_amphora.amphora_id)
def test_delete(self):
amphora = self.create_amphora(
self.session)
with self.session.begin():
self.session.delete(amphora)
self.session.flush()
new_amphora = self.session.query(
models.Amphora).filter_by(id=amphora.id).first()
self.assertIsNone(new_amphora)
def test_load_balancer_relationship(self):
amphora = self.create_amphora(self.session)
self.associate_amphora(self.load_balancer, amphora)
new_amphora = self.session.query(models.Amphora).filter_by(
id=amphora.id).first()
self.assertIsNotNone(new_amphora.load_balancer)
self.assertIsInstance(new_amphora.load_balancer, models.LoadBalancer)
class DataModelConversionTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(DataModelConversionTest, self).setUp()
self.pool = self.create_pool(self.session)
self.hm = self.create_health_monitor(self.session, self.pool.id)
self.member = self.create_member(self.session, self.pool.id,
id=self.FAKE_UUID_1,
ip_address='10.0.0.1')
self.sp = self.create_session_persistence(self.session, self.pool.id)
self.lb = self.create_load_balancer(self.session)
self.vip = self.create_vip(self.session, self.lb.id)
self.listener = self.create_listener(self.session,
default_pool_id=self.pool.id,
load_balancer_id=self.lb.id)
self.stats = self.create_listener_statistics(self.session,
self.listener.id)
self.sni = self.create_sni(self.session, listener_id=self.listener.id)
def test_load_balancer_tree(self):
lb_db = self.session.query(models.LoadBalancer).filter_by(
id=self.lb.id).first()
self.check_load_balancer(lb_db.to_data_model())
def test_vip_tree(self):
vip_db = self.session.query(models.Vip).filter_by(
load_balancer_id=self.lb.id).first()
self.check_vip(vip_db.to_data_model())
def test_listener_tree(self):
listener_db = self.session.query(models.Listener).filter_by(
id=self.listener.id).first()
self.check_listener(listener_db.to_data_model())
def test_sni_tree(self):
sni_db = self.session.query(models.SNI).filter_by(
listener_id=self.listener.id).first()
self.check_sni(sni_db.to_data_model())
def test_listener_statistics_tree(self):
stats_db = self.session.query(models.ListenerStatistics).filter_by(
listener_id=self.listener.id).first()
self.check_listener_statistics(stats_db.to_data_model())
def test_pool_tree(self):
pool_db = self.session.query(models.Pool).filter_by(
id=self.pool.id).first()
self.check_pool(pool_db.to_data_model())
def test_session_persistence_tree(self):
sp_db = self.session.query(models.SessionPersistence).filter_by(
pool_id=self.pool.id).first()
self.check_session_persistence(sp_db.to_data_model())
def test_health_monitor_tree(self):
hm_db = self.session.query(models.HealthMonitor).filter_by(
id=self.hm.id).first()
self.check_health_monitor(hm_db.to_data_model())
def test_member_tree(self):
member_db = self.session.query(models.Member).filter_by(
id=self.member.id).first()
self.check_member(member_db.to_data_model())
def check_load_balancer(self, lb, check_listeners=True,
check_amphorae=True, check_vip=True):
self.assertIsInstance(lb, data_models.LoadBalancer)
self.check_load_balancer_data_model(lb)
self.assertIsInstance(lb.listeners, list)
self.assertIsInstance(lb.amphorae, list)
if check_listeners:
for listener in lb.listeners:
self.check_listener(listener, check_lb=False)
if check_amphorae:
for amphora in lb.amphorae:
self.check_amphora(amphora, check_load_balancer=False)
if check_vip:
self.check_vip(lb.vip, check_lb=False)
def check_vip(self, vip, check_lb=True):
self.assertIsInstance(vip, data_models.Vip)
self.check_vip_data_model(vip)
if check_lb:
self.check_load_balancer(vip.load_balancer, check_vip=False)
def check_sni(self, sni, check_listener=True):
self.assertIsInstance(sni, data_models.SNI)
self.check_sni_data_model(sni)
if check_listener:
self.check_listener(sni.listener, check_sni=False)
def check_listener_statistics(self, stats, check_listener=True):
self.assertIsInstance(stats, data_models.ListenerStatistics)
self.check_listener_statistics_data_model(stats)
if check_listener:
self.check_listener(stats.listener, check_statistics=False)
def check_amphora(self, amphora, check_load_balancer=True):
self.assertIsInstance(amphora, data_models.Amphora)
self.check_amphora_data_model(amphora)
if check_load_balancer:
self.check_load_balancer(amphora.load_balancer)
def check_listener(self, listener, check_sni=True, check_pool=True,
check_lb=True, check_statistics=True):
self.assertIsInstance(listener, data_models.Listener)
self.check_listener_data_model(listener)
if check_lb:
self.check_load_balancer(listener.load_balancer)
if check_sni:
c_containers = listener.sni_containers
self.assertIsInstance(c_containers, list)
for sni in c_containers:
self.check_sni(sni, check_listener=False)
if check_pool:
self.check_pool(listener.default_pool, check_listener=False)
if check_statistics:
self.check_listener_statistics(listener.stats,
check_listener=False)
def check_session_persistence(self, session_persistence, check_pool=True):
self.assertIsInstance(session_persistence,
data_models.SessionPersistence)
self.check_session_persistence_data_model(session_persistence)
if check_pool:
self.check_pool(session_persistence.pool, check_sp=False)
def check_member(self, member, check_pool=True):
self.assertIsInstance(member, data_models.Member)
self.check_member_data_model(member)
if check_pool:
self.check_pool(member.pool)
def check_health_monitor(self, health_monitor, check_pool=True):
self.assertIsInstance(health_monitor, data_models.HealthMonitor)
self.check_health_monitor_data_model(health_monitor)
if check_pool:
self.check_pool(health_monitor.pool, check_hm=False)
def check_pool(self, pool, check_listener=True, check_sp=True,
check_hm=True, check_members=True):
self.assertIsInstance(pool, data_models.Pool)
self.check_pool_data_model(pool)
if check_listener:
self.check_listener(pool.listener, check_pool=False)
if check_sp:
self.check_session_persistence(pool.session_persistence,
check_pool=False)
if check_members:
c_members = pool.members
self.assertIsNotNone(c_members)
self.assertEqual(1, len(c_members))
for c_member in c_members:
self.check_member(c_member, check_pool=False)
if check_hm:
self.check_health_monitor(pool.health_monitor, check_pool=False)
def check_load_balancer_data_model(self, lb):
self.assertEqual(self.FAKE_UUID_1, lb.tenant_id)
self.assertEqual(self.FAKE_UUID_1, lb.id)
self.assertEqual(constants.ACTIVE, lb.provisioning_status)
self.assertEqual(True, lb.enabled)
def check_vip_data_model(self, vip):
self.assertEqual(self.FAKE_UUID_1, vip.load_balancer_id)
def check_listener_data_model(self, listener):
self.assertEqual(self.FAKE_UUID_1, listener.tenant_id)
self.assertEqual(self.FAKE_UUID_1, listener.id)
self.assertEqual(constants.PROTOCOL_HTTP, listener.protocol)
self.assertEqual(80, listener.protocol_port)
self.assertEqual(constants.ACTIVE, listener.provisioning_status)
self.assertEqual(constants.ONLINE, listener.operating_status)
self.assertEqual(True, listener.enabled)
def check_sni_data_model(self, sni):
self.assertEqual(self.FAKE_UUID_1, sni.listener_id)
self.assertEqual(self.FAKE_UUID_1, sni.tls_container_id)
def check_listener_statistics_data_model(self, stats):
self.assertEqual(self.listener.id, stats.listener_id)
self.assertEqual(0, stats.bytes_in)
self.assertEqual(0, stats.bytes_out)
self.assertEqual(0, stats.active_connections)
self.assertEqual(0, stats.total_connections)
def check_pool_data_model(self, pool):
self.assertEqual(self.FAKE_UUID_1, pool.tenant_id)
self.assertEqual(self.FAKE_UUID_1, pool.id)
self.assertEqual(constants.PROTOCOL_HTTP, pool.protocol)
self.assertEqual(constants.LB_ALGORITHM_ROUND_ROBIN, pool.lb_algorithm)
self.assertEqual(constants.ONLINE, pool.operating_status)
self.assertEqual(True, pool.enabled)
def check_session_persistence_data_model(self, sp):
self.assertEqual(self.pool.id, sp.pool_id)
self.assertEqual(constants.SESSION_PERSISTENCE_HTTP_COOKIE, sp.type)
def check_health_monitor_data_model(self, hm):
self.assertEqual(self.FAKE_UUID_1, hm.tenant_id)
self.assertEqual(self.FAKE_UUID_1, hm.id)
self.assertEqual(constants.HEALTH_MONITOR_HTTP, hm.type)
self.assertEqual(1, hm.delay)
self.assertEqual(1, hm.timeout)
self.assertEqual(1, hm.fall_threshold)
self.assertEqual(1, hm.rise_threshold)
self.assertEqual(True, hm.enabled)
def check_member_data_model(self, member):
self.assertEqual(self.FAKE_UUID_1, member.tenant_id)
self.assertEqual(self.FAKE_UUID_1, member.id)
self.assertEqual(self.pool.id, member.pool_id)
self.assertEqual('10.0.0.1', member.ip_address)
self.assertEqual(80, member.protocol_port)
self.assertEqual(constants.ONLINE, member.operating_status)
self.assertEqual(True, member.enabled)
def check_amphora_data_model(self, amphora):
self.assertEqual(self.FAKE_UUID_1, amphora.id)
self.assertEqual(self.FAKE_UUID_1, amphora.host_id)
self.assertEqual(constants.ONLINE, amphora.status)
def check_load_balancer_amphora_data_model(self, amphora):
self.assertEqual(self.FAKE_UUID_1, amphora.amphora_id)
self.assertEqual(self.FAKE_UUID_1, amphora.load_balancer_id)
View Git Blame Copy Permalink