openstack
/
networking-l2gw
Code Issues Proposed changes

handle connection from multiple ovsdb servers

Browse Source 
ovsdb server initiates the connection to l2gw agent with the entries in
manager table in ovsdb hardware vtep schema and with flag 'enable_manager'
set to True in l2gateway_agent.ini.
multiple ovsdb can connect to same l2gatway agent with the l2gw agent ip in
respective manager table.
This patchset addresses the handling of connections from multiple
ovsdb servers.

Change-Id: Iffe6a9e8599ffaed55c3f8b749c4d6f7d0e52c79
Closes-Bug:1495005
changes/41/222841/1
vikas 8 years ago
parent 0211f43e11
commit ed5f17fec6
8 changed files with 274 additions and 194 deletions
Show Stats Download Patch File Download Diff File

73
networking_l2gw/services/l2gateway/agent/ovsdb/base_connection.py
Unescape Escape View File

@ -46,10 +46,11 @@ class BaseConnection(object):
self.enable_manager = cfg.CONF.ovsdb.enable_manager
if self.enable_manager:
self.s = None
self.c_sock = None
self.addr = None
self.check_c_sock = None
self.check_sock_rcv = False
eventlet.greenthread.spawn(self._rcv_socket)
self.ovsdb_dicts = {}
self.ovsdb_fd_states = {}
else:
self.gw_config = gw_config
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@ -96,30 +97,72 @@ class BaseConnection(object):
self.s.listen(5) # Now wait for client connection.
while True:
# Establish connection with client.
self.c_sock, ip_addr = self.s.accept()
self.addr = ip_addr[0]
self.c_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
LOG.debug("Got connection from %s ", self.addr)
c_sock, ip_addr = self.s.accept()
addr = ip_addr[0]
c_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
LOG.debug("Got connection from %s ", addr)
self.connected = True
self.ovsdb_dicts[addr] = c_sock
eventlet.greenthread.spawn(self._common_sock_rcv_thread, addr)
def _echo_response(self):
def _common_sock_rcv_thread(self, addr):
chunks = []
lc = rc = 0
prev_char = None
self._echo_response(addr)
if self.enable_manager and self.check_c_sock:
while self.read_on:
response = self.ovsdb_dicts.get(addr).recv(n_const.BUFFER_SIZE)
self.ovsdb_fd_states[addr] = 'connected'
eventlet.greenthread.sleep(0)
self.check_sock_rcv = True
if response:
response = response.decode('utf8')
message_mark = 0
for i, c in enumerate(response):
if c == '{' and not (prev_char and
prev_char == '\\'):
lc += 1
elif c == '}' and not (prev_char and
prev_char == '\\'):
rc += 1
if rc > lc:
raise Exception(_LE("json string not valid"))
elif lc == rc and lc is not 0:
chunks.append(response[message_mark:i + 1])
message = "".join(chunks)
eventlet.greenthread.spawn_n(
self._on_remote_message, message, addr)
eventlet.greenthread.sleep(0)
lc = rc = 0
message_mark = i + 1
chunks = []
prev_char = c
chunks.append(response[message_mark:])
else:
self.read_on = False
self.disconnect(addr)
self.ovsdb_fd_states[addr] = 'disconnected'
def _echo_response(self, addr):
while True:
try:
if self.enable_manager:
eventlet.greenthread.sleep(0)
response = self.c_sock.recv(n_const.BUFFER_SIZE)
response = self.ovsdb_dicts.get(addr).recv(
n_const.BUFFER_SIZE)
sock_json_m = jsonutils.loads(response)
sock_handler_method = sock_json_m.get('method', None)
if sock_handler_method == 'echo':
self.check_c_sock = True
self.c_sock.send(jsonutils.dumps(
self.ovsdb_dicts.get(addr).send(jsonutils.dumps(
{"result": sock_json_m.get("params", None),
"error": None, "id": sock_json_m['id']}))
break
except Exception:
continue
def send(self, message, callback=None):
def send(self, message, callback=None, addr=None):
"""Sends a message to the OVSDB server."""
if callback:
self.callbacks[message['id']] = callback
@ -128,7 +171,8 @@ class BaseConnection(object):
while retry_count <= n_const.MAX_RETRIES:
try:
if self.enable_manager:
bytes_sent = self.c_sock.send(jsonutils.dumps(message))
bytes_sent = self.ovsdb_dicts.get(addr).send(
jsonutils.dumps(message))
else:
bytes_sent = self.socket.send(jsonutils.dumps(message))
if bytes_sent:
@ -140,13 +184,14 @@ class BaseConnection(object):
LOG.warning(_LW("Could not send message to the "
"OVSDB server."))
self.disconnect()
self.disconnect(addr)
return False
def disconnect(self):
def disconnect(self, addr=None):
"""disconnects the connection from the OVSDB server."""
if self.enable_manager:
self.c_sock.close()
self.ovsdb_dicts.get(addr).close()
del self.ovsdb_dicts[addr]
else:
self.socket.close()
self.connected = False

65
networking_l2gw/services/l2gateway/agent/ovsdb/manager.py
Unescape Escape View File

@ -45,6 +45,9 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
self.enable_manager = cfg.CONF.ovsdb.enable_manager
if self.enable_manager:
self.ovsdb_fd = None
self._sock_open_connection()
self.looping_task_ovsdb_states = (
loopingcall.FixedIntervalLoopingCall(self._send_ovsdb_states))
else:
self.looping_task = loopingcall.FixedIntervalLoopingCall(
self._connect_to_ovsdb_server)
@ -148,6 +151,10 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
self._stop_looping_task()
self._disconnect_all_ovsdb_servers()
def _send_ovsdb_states(self):
self.plugin_rpc.notify_ovsdb_states(ctx.get_admin_context(),
self.ovsdb_fd.ovsdb_fd_states)
def _disconnect_all_ovsdb_servers(self):
if self.gateways:
for key, gateway in self.gateways.items():
@ -171,14 +178,32 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
elif self.enable_manager and self.l2gw_agent_type == n_const.MONITOR:
if self.ovsdb_fd is None:
self._sock_open_connection()
elif self.ovsdb_fd and not self.ovsdb_fd.check_monitor_thread:
self.ovsdb_fd._spawn_monitor_thread()
elif not self.enable_manager:
elif ((self.ovsdb_fd) and not (
self.ovsdb_fd.check_monitor_table_thread) and (
self.ovsdb_fd.check_sock_rcv)):
for key in self.ovsdb_fd.ovsdb_dicts.keys():
eventlet.greenthread.spawn_n(
self.ovsdb_fd._spawn_monitor_table_thread, key)
self._start_looping_task_ovsdb_states()
elif ((self.enable_manager) and not (
self.l2gw_agent_type == n_const.MONITOR)):
self._stop_looping_task_ovsdb_states()
elif (not (self.l2gw_agent_type == n_const.MONITOR) and not (
self.enable_manager)):
# Otherwise, stop monitoring the OVSDB servers
# and close the open connections if any.
self._stop_looping_task()
self._disconnect_all_ovsdb_servers()
def _stop_looping_task_ovsdb_states(self):
if self.looping_task_ovsdb_states._running:
self.looping_task_ovsdb_states.stop()
def _start_looping_task_ovsdb_states(self):
if not self.looping_task_ovsdb_states._running:
self.looping_task_ovsdb_states.start(
interval=self.conf.ovsdb.periodic_interval)
def _stop_looping_task(self):
if self.looping_task._running:
self.looping_task.stop()
@ -218,13 +243,15 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
def delete_network(self, context, ovsdb_identifier, logical_switch_uuid):
"""Handle RPC cast from plugin to delete a network."""
if self.enable_manager and self.l2gw_agent_type == n_const.MONITOR:
self.ovsdb_fd.delete_logical_switch(logical_switch_uuid, False)
self.ovsdb_fd.delete_logical_switch(logical_switch_uuid,
ovsdb_identifier, False)
elif ((self.enable_manager) and (
not self.l2gw_agent_type == n_const.MONITOR)):
self._sock_open_connection()
self.ovsdb_fd._echo_response()
self.ovsdb_fd._echo_response(ovsdb_identifier)
if self.ovsdb_fd.check_c_sock:
self.ovsdb_fd.delete_logical_switch(logical_switch_uuid,
ovsdb_identifier,
False)
elif not self.enable_manager:
if self._is_valid_request(ovsdb_identifier):
@ -238,16 +265,17 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
if self.enable_manager and self.l2gw_agent_type == n_const.MONITOR:
self.ovsdb_fd.insert_ucast_macs_remote(logical_switch_dict,
locator_dict,
mac_dict, False)
mac_dict, ovsdb_identifier,
False)
elif ((self.enable_manager) and (
not self.l2gw_agent_type == n_const.MONITOR)):
self._sock_open_connection()
self.ovsdb_fd._echo_response()
self.ovsdb_fd._echo_response(ovsdb_identifier)
if self.ovsdb_fd.check_c_sock:
self.ovsdb_fd.insert_ucast_macs_remote(
logical_switch_dict,
locator_dict,
mac_dict)
mac_dict, ovsdb_identifier)
elif not self.enable_manager:
if self._is_valid_request(ovsdb_identifier):
with self._open_connection(ovsdb_identifier) as ovsdb_fd:
@ -260,15 +288,16 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
"""Handle RPC cast from plugin to delete neutron port MACs."""
if self.enable_manager and self.l2gw_agent_type == n_const.MONITOR:
self.ovsdb_fd.delete_ucast_macs_remote(logical_switch_uuid, mac,
ovsdb_identifier,
False)
elif ((self.enable_manager) and (
not self.l2gw_agent_type == n_const.MONITOR)):
self._sock_open_connection()
self.ovsdb_fd._echo_response()
self.ovsdb_fd._echo_response(ovsdb_identifier)
if self.ovsdb_fd.check_c_sock:
self.ovsdb_fd.delete_ucast_macs_remote(
logical_switch_uuid,
mac)
mac, ovsdb_identifier)
elif not self.enable_manager:
if self._is_valid_request(ovsdb_identifier):
with self._open_connection(ovsdb_identifier) as ovsdb_fd:
@ -282,14 +311,16 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
"""
if self.enable_manager and self.l2gw_agent_type == n_const.MONITOR:
self.ovsdb_fd.update_ucast_macs_remote(locator_dict,
mac_dict, False)
mac_dict, ovsdb_identifier,
False)
elif ((self.enable_manager) and (
not self.l2gw_agent_type == n_const.MONITOR)):
self._sock_open_connection()
self.ovsdb_fd._echo_response()
self.ovsdb_fd._echo_response(ovsdb_identifier)
if self.ovsdb_fd.check_c_sock:
self.ovsdb_fd.update_ucast_macs_remote(locator_dict,
mac_dict)
mac_dict,
ovsdb_identifier)
elif not self.enable_manager:
if self._is_valid_request(ovsdb_identifier):
with self._open_connection(ovsdb_identifier) as ovsdb_fd:
@ -308,17 +339,19 @@ class OVSDBManager(base_agent_manager.BaseAgentManager):
self.ovsdb_fd.update_connection_to_gateway(logical_switch_dict,
locator_dicts,
mac_dicts,
port_dicts, False)
port_dicts,
ovsdb_identifier,
False)
elif ((self.enable_manager) and (
not self.l2gw_agent_type == n_const.MONITOR)):
self._sock_open_connection()
self.ovsdb_fd._echo_response()
self.ovsdb_fd._echo_response(ovsdb_identifier)
if self.ovsdb_fd.check_c_sock:
self.ovsdb_fd.update_connection_to_gateway(
logical_switch_dict,
locator_dicts,
mac_dicts,
port_dicts)
port_dicts, ovsdb_identifier)
elif not self.enable_manager:
if self._is_valid_request(ovsdb_identifier):
with self._open_connection(ovsdb_identifier) as ovsdb_fd:

82
networking_l2gw/services/l2gateway/agent/ovsdb/ovsdb_monitor.py
Unescape Escape View File

@ -41,14 +41,13 @@ class OVSDBMonitor(base_connection.BaseConnection):
self.read_on = True
self.handlers = {"echo": self._default_echo_handler}
if self.enable_manager:
self.check_c_sock = self.c_sock
self.check_monitor_thread = False
self.check_monitor_table_thread = False
if not self.enable_manager:
eventlet.greenthread.spawn(self._rcv_thread)
def _spawn_monitor_thread(self):
eventlet.greenthread.spawn(self._sock_rcv_thread)
self.check_monitor_thread = True
def _spawn_monitor_table_thread(self, addr):
self.set_monitor_response_handler(addr)
self.check_monitor_table_thread = True
def _initialize_data_dict(self):
data_dict = {'new_local_macs': [],
@ -91,7 +90,7 @@ class OVSDBMonitor(base_connection.BaseConnection):
self._process_physical_locator_set
}
def set_monitor_response_handler(self):
def set_monitor_response_handler(self, addr=None):
"""Monitor OVSDB tables to receive events for any changes in OVSDB."""
if self.connected:
op_id = str(random.getrandbits(128))
@ -113,7 +112,7 @@ class OVSDBMonitor(base_connection.BaseConnection):
'Physical_Locator_Set': [props]}
]}
self._set_handler("update", self._update_event_handler)
if not self.send(monitor_message):
if not self.send(monitor_message, addr=addr):
# Return so that this will retried in the next iteration
return
try:
@ -121,15 +120,15 @@ class OVSDBMonitor(base_connection.BaseConnection):
except exceptions.OVSDBError:
with excutils.save_and_reraise_exception():
if self.enable_manager:
self.check_monitor_thread = False
self.check_monitor_table_thread = False
LOG.exception(_LE("Exception while receiving the "
"response for the monitor message"))
self._process_monitor_msg(response_result)
self._process_monitor_msg(response_result, addr)
def _update_event_handler(self, message):
self._process_update_event(message)
def _update_event_handler(self, message, addr):
self._process_update_event(message, addr)
def _process_update_event(self, message):
def _process_update_event(self, message, addr):
"""Process update event that is triggered by the OVSDB server."""
LOG.debug("_process_update_event: message = %s ", str(message))
data_dict = self._initialize_data_dict()
@ -137,7 +136,7 @@ class OVSDBMonitor(base_connection.BaseConnection):
params_list = message.get('params')
param_dict = params_list[1]
self._process_tables(param_dict, data_dict)
self.rpc_callback(self._form_ovsdb_data(data_dict))
self.rpc_callback(self._form_ovsdb_data(data_dict, addr))
def _process_tables(self, param_dict, data_dict):
# Process all the tables one by one.
@ -177,64 +176,27 @@ class OVSDBMonitor(base_connection.BaseConnection):
)
return result
def _default_echo_handler(self, message):
def _default_echo_handler(self, message, addr):
"""Message handler for the OVSDB server's echo request."""
self.send({"result": message.get("params", None),
"error": None, "id": message['id']})
"error": None, "id": message['id']}, addr=addr)
def _set_handler(self, method_name, handler):
self.handlers[method_name] = handler
def _on_remote_message(self, message):
def _on_remote_message(self, message, addr=None):
"""Processes the message received on the socket."""
try:
json_m = jsonutils.loads(message)
handler_method = json_m.get('method', None)
if handler_method:
self.handlers.get(handler_method)(json_m)
self.handlers.get(handler_method)(json_m, addr)
else:
self.responses.append(json_m)
except Exception as e:
LOG.exception(_LE("Exception [%s] while handling "
"message"), e)
def _sock_rcv_thread(self):
chunks = []
lc = rc = 0
prev_char = None
self._echo_response()
if self.enable_manager and self.check_c_sock:
eventlet.greenthread.spawn_n(self.set_monitor_response_handler)
while self.read_on:
response = self.c_sock.recv(n_const.BUFFER_SIZE)
eventlet.greenthread.sleep(0)
if response:
response = response.decode('utf8')
message_mark = 0
for i, c in enumerate(response):
if c == '{' and not (prev_char and
prev_char == '\\'):
lc += 1
elif c == '}' and not (prev_char and
prev_char == '\\'):
rc += 1
if rc > lc:
raise Exception(_LE("json string not valid"))
elif lc == rc and lc is not 0:
chunks.append(response[message_mark:i + 1])
message = "".join(chunks)
eventlet.greenthread.spawn_n(
self._on_remote_message, message)
eventlet.greenthread.sleep(0)
lc = rc = 0
message_mark = i + 1
chunks = []
prev_char = c
chunks.append(response[message_mark:])
else:
self.read_on = False
self.disconnect()
def _rcv_thread(self):
chunks = []
lc = rc = 0
@ -274,26 +236,26 @@ class OVSDBMonitor(base_connection.BaseConnection):
LOG.exception(_LE("Exception [%s] occurred while receiving"
"message from the OVSDB server"), ex)
def disconnect(self):
def disconnect(self, addr=None):
"""disconnects the connection from the OVSDB server."""
self.read_on = False
super(OVSDBMonitor, self).disconnect()
super(OVSDBMonitor, self).disconnect(addr)
def _process_monitor_msg(self, message):
def _process_monitor_msg(self, message, addr=None):
"""Process initial set of records in the OVSDB at startup."""
result_dict = message.get('result')
data_dict = self._initialize_data_dict()
try:
self._process_tables(result_dict, data_dict)
self.rpc_callback(self._form_ovsdb_data(data_dict))
self.rpc_callback(self._form_ovsdb_data(data_dict, addr))
except Exception as e:
LOG.exception(_LE("_process_monitor_msg:ERROR %s "), e)
def _get_list(self, resource_list):
return [element.__dict__ for element in resource_list]
def _form_ovsdb_data(self, data_dict):
return {n_const.OVSDB_IDENTIFIER: str(self.addr) if (
def _form_ovsdb_data(self, data_dict, addr):
return {n_const.OVSDB_IDENTIFIER: str(addr) if (
self.enable_manager) else (self.gw_config.ovsdb_identifier),
'new_logical_switches': self._get_list(
data_dict.get('new_logical_switches')),

44
networking_l2gw/services/l2gateway/agent/ovsdb/ovsdb_writer.py
Unescape Escape View File

@ -53,10 +53,10 @@ class OVSDBWriter(base_connection.BaseConnection):
message="Error from the OVSDB server: %s" % error)
return result
def _get_reply(self, operation_id):
def _get_reply(self, operation_id, ovsdb_identifier):
count = 0
while count <= n_const.MAX_RETRIES:
response = self._recv_data()
response = self._recv_data(ovsdb_identifier)
LOG.debug("Response from OVSDB server = %s", str(response))
if response:
try:
@ -64,9 +64,10 @@ class OVSDBWriter(base_connection.BaseConnection):
self.responses.append(json_m)
method_type = json_m.get('method', None)
if method_type == "echo" and self.enable_manager:
self.c_sock.send(jsonutils.dumps(
{"result": json_m.get("params", None),
"error": None, "id": json_m['id']}))
self.ovsdb_dicts.get(ovsdb_identifier).send(
jsonutils.dumps(
{"result": json_m.get("params", None),
"error": None, "id": json_m['id']}))
else:
if self._process_response(operation_id):
return True
@ -80,13 +81,15 @@ class OVSDBWriter(base_connection.BaseConnection):
LOG.error(_LE("Could not obtain response from the OVSDB server "
"for the request"))
def _send_and_receive(self, query, operation_id, rcv_required):
if not self.send(query):
def _send_and_receive(self, query, operation_id, ovsdb_identifier,
rcv_required):
if not self.send(query, addr=ovsdb_identifier):
return
if rcv_required:
self._get_reply(operation_id)
self._get_reply(operation_id, ovsdb_identifier)
def delete_logical_switch(self, logical_switch_uuid, rcv_required=True):
def delete_logical_switch(self, logical_switch_uuid, ovsdb_identifier,
rcv_required=True):
"""Delete an entry from Logical_Switch OVSDB table."""
commit_dict = {"op": "commit", "durable": True}
op_id = str(random.getrandbits(128))
@ -99,10 +102,11 @@ class OVSDBWriter(base_connection.BaseConnection):
commit_dict],
"id": op_id}
LOG.debug("delete_logical_switch: query: %s", query)
self._send_and_receive(query, op_id, rcv_required)
self._send_and_receive(query, op_id, ovsdb_identifier, rcv_required)
def insert_ucast_macs_remote(self, l_switch_dict, locator_dict,
mac_dict, rcv_required=True):
mac_dict, ovsdb_identifier,
rcv_required=True):
"""Insert an entry in Ucast_Macs_Remote OVSDB table."""
# To insert an entry in Ucast_Macs_Remote table, it requires
# corresponding entry in Physical_Locator (Compute node VTEP IP)
@ -146,9 +150,10 @@ class OVSDBWriter(base_connection.BaseConnection):
"params": params,
"id": op_id}
LOG.debug("insert_ucast_macs_remote: query: %s", query)
self._send_and_receive(query, op_id, rcv_required)
self._send_and_receive(query, op_id, ovsdb_identifier, rcv_required)
def update_ucast_macs_remote(self, locator_dict, mac_dict,
ovsdb_identifier,
rcv_required=True):
"""Update an entry in Ucast_Macs_Remote OVSDB table."""
# It is possible that the locator may not exist already.
@ -181,9 +186,10 @@ class OVSDBWriter(base_connection.BaseConnection):
"params": params,
"id": op_id}
LOG.debug("update_ucast_macs_remote: query: %s", query)
self._send_and_receive(query, op_id, rcv_required)
self._send_and_receive(query, op_id, ovsdb_identifier, rcv_required)
def delete_ucast_macs_remote(self, logical_switch_uuid, macs,
ovsdb_identifier,
rcv_required=True):
"""Delete entries from Ucast_Macs_Remote OVSDB table."""
commit_dict = {"op": "commit", "durable": True}
@ -205,11 +211,12 @@ class OVSDBWriter(base_connection.BaseConnection):
"params": params,
"id": op_id}
LOG.debug("delete_ucast_macs_remote: query: %s", query)
self._send_and_receive(query, op_id, rcv_required)
self._send_and_receive(query, op_id, ovsdb_identifier, rcv_required)
def update_connection_to_gateway(self, logical_switch_dict,
locator_dicts, mac_dicts,
port_dicts, rcv_required=True):
port_dicts, ovsdb_identifier,
rcv_required=True):
"""Updates Physical Port's VNI to VLAN binding."""
# Form the JSON Query so as to update the physical port with the
# vni-vlan (logical switch uuid to vlan) binding
@ -222,16 +229,17 @@ class OVSDBWriter(base_connection.BaseConnection):
"params": update_dicts,
"id": op_id}
LOG.debug("update_connection_to_gateway: query = %s", query)
self._send_and_receive(query, op_id, rcv_required)
self._send_and_receive(query, op_id, ovsdb_identifier, rcv_required)
def _recv_data(self):
def _recv_data(self, ovsdb_identifier):
chunks = []
lc = rc = 0
prev_char = None
while True:
try:
if self.enable_manager:
response = self.c_sock.recv(n_const.BUFFER_SIZE)
response = self.ovsdb_dicts.get(ovsdb_identifier).recv(
n_const.BUFFER_SIZE)
else:
response = self.socket.recv(n_const.BUFFER_SIZE)
if response:

50
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_base_connection.py
Unescape Escape View File

@ -75,6 +75,9 @@ class SocketClass(object):
raise self.recv_error
return self.rcv_data
def close(self):
pass
class TestBaseConnection(base.BaseTestCase):
def setUp(self):
@ -181,17 +184,22 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_ovsdb_conn = base_connection.BaseConnection(mock.Mock(),
self.conf)
self.l2gw_ovsdb_conn.c_sock = mock.patch('socket.socket').start()
self.mock_sock = mock.patch('socket.socket').start()
self.l2gw_ovsdb_conn.s = mock.patch('socket.socket').start()
self.fakesocket = SocketClass()
self.fake_ip = 'fake_ip'
self.l2gw_ovsdb_conn.ovsdb_dicts = {'fake_ip': self.fakesocket}
def test_init_with_enable_manager(self):
fake_dict = {}
with mock.patch.object(eventlet.greenthread,
'spawn') as (mock_thread):
self.l2gw_ovsdb_conn.__init__(mock.Mock(), self.conf)
self.assertEqual(self.l2gw_ovsdb_conn.s, mock.ANY)
self.assertIsNone(self.l2gw_ovsdb_conn.c_sock)
self.assertIsNone(self.l2gw_ovsdb_conn.addr)
self.assertFalse(self.l2gw_ovsdb_conn.check_sock_rcv)
self.assertIsNone(self.l2gw_ovsdb_conn.check_c_sock)
self.assertEqual(self.l2gw_ovsdb_conn.ovsdb_dicts, fake_dict)
self.assertEqual(self.l2gw_ovsdb_conn.ovsdb_fd_states, fake_dict)
self.assertTrue(mock_thread.called)
def test_echo_response(self):
@ -202,23 +210,45 @@ class TestBaseConnection_with_enable_manager(base.BaseTestCase):
with contextlib.nested(
mock.patch.object(eventlet.greenthread, 'sleep'),
mock.patch.object(jsonutils, 'loads', return_value=fake_resp),
mock.patch.object(self.l2gw_ovsdb_conn.c_sock,
mock.patch.object(self.fakesocket,
'recv',
return_value=fake_resp),
mock.patch.object(self.l2gw_ovsdb_conn.c_sock,
'send')) as (
mock.patch.object(self.fakesocket,
'send')
) as (
fake_thread, mock_loads,
mock_sock_rcv,
mock_sock_send):
self.l2gw_ovsdb_conn._echo_response()
self.l2gw_ovsdb_conn._echo_response(self.fake_ip)
self.assertTrue(fake_thread.called)
self.assertTrue(mock_sock_rcv.called)
mock_loads.assert_called_with(fake_resp)
self.assertTrue(self.l2gw_ovsdb_conn.check_c_sock)
self.assertTrue(mock_sock_send.called)
def test_common_sock_rcv_thread_none(self):
with contextlib.nested(
mock.patch.object(base_connection.BaseConnection,
'_echo_response'),
mock.patch.object(eventlet.greenthread, 'sleep'),
mock.patch.object(self.fakesocket,
'recv', return_value=None),
mock.patch.object(base_connection.BaseConnection,
'disconnect')) as (
mock_resp, green_thrd_sleep,
mock_rcv, mock_disconnect):
self.l2gw_ovsdb_conn.check_c_sock = True
self.l2gw_ovsdb_conn.read_on = True
self.l2gw_ovsdb_conn._common_sock_rcv_thread(self.fake_ip)
self.assertTrue(mock_resp.called)
self.assertTrue(green_thrd_sleep.called)
self.assertTrue(mock_rcv.called)
self.assertTrue(mock_disconnect.called)
self.assertFalse(self.l2gw_ovsdb_conn.connected)
self.assertFalse(self.l2gw_ovsdb_conn.read_on)
def test_disconnect_with_enable_manager(self):
with mock.patch.object(self.l2gw_ovsdb_conn.c_sock,
'close') as mock_close:
self.l2gw_ovsdb_conn.disconnect()
with mock.patch.object(self.fakesocket,
'close') as (mock_close):
self.l2gw_ovsdb_conn.disconnect(self.fake_ip)
self.assertTrue(mock_close.called)

66
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_manager.py
Unescape Escape View File

@ -225,8 +225,14 @@ class TestManager(base.BaseTestCase):
def test_init_with_enable_manager(self):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
self.l2gw_agent_manager.__init__()
self.assertIsNone(self.l2gw_agent_manager.ovsdb_fd)
with contextlib.nested(
mock.patch.object(manager.OVSDBManager,
'_sock_open_connection'),
mock.patch.object(loopingcall, 'FixedIntervalLoopingCall')) as (
mock_sock_open_conn, mock_loop):
self.l2gw_agent_manager.__init__()
self.assertTrue(mock_sock_open_conn.called)
self.assertTrue(mock_loop.called)
def test_sock_open_connection(self):
cfg.CONF.set_override('enable_manager', True, 'ovsdb')
@ -253,14 +259,23 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.conf.host = 'fake_host'
self.l2gw_agent_manager.__init__()
self.l2gw_agent_manager.l2gw_agent_type = n_const.MONITOR
with mock.patch.object(ovsdb_common_class,
'OVSDB_commom_class') as mock_ovsdb_common:
with contextlib.nested(
mock.patch.object(ovsdb_common_class,
'OVSDB_commom_class'),
mock.patch.object(eventlet.greenthread,
'spawn_n'),
mock.patch.object(self.l2gw_agent_manager,
'_start_looping_task_ovsdb_states')) as (
mock_ovsdb_common, mock_thread, mock_looping):
self.l2gw_agent_manager.ovsdb_fd = mock_ovsdb_common.return_value
self.l2gw_agent_manager.ovsdb_fd.check_monitor_thread = False
self.l2gw_agent_manager.ovsdb_fd.check_monitor_table_thread = False
self.l2gw_agent_manager.ovsdb_fd.check_sock_rcv = True
self.l2gw_agent_manager.ovsdb_fd.ovsdb_dicts = {
"fake_ip": "fake_sock"}
self.l2gw_agent_manager.set_monitor_agent(self.context,
'fake_host')
(self.l2gw_agent_manager.ovsdb_fd._spawn_monitor_thread.
assert_called_with())
self.assertTrue(mock_thread.called)
self.assertTrue(mock_looping.called)
def test_update_connection_to_gateway_for_monitor_agent(self):
"""Test case to test update_connection_to_gateway for
@ -277,7 +292,8 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), mock.Mock(), mock.Mock(),
mock.Mock(), mock.Mock())
(self.l2gw_agent_manager.ovsdb_fd.update_connection_to_gateway.
assert_called_with(mock.ANY, mock.ANY, mock.ANY, mock.ANY, False))
assert_called_with(mock.ANY, mock.ANY, mock.ANY, mock.ANY,
mock.ANY, False))
def test_update_connection_to_gateway_for_transact_agent(self):
"""Test case to test update_connection_to_gateway
@ -298,10 +314,11 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), mock.Mock(), mock.Mock(),
mock.Mock(), mock.Mock())
(self.l2gw_agent_manager.ovsdb_fd._echo_response.
assert_called_with())
assert_called_with(mock.ANY))
self.assertTrue(mock_open_conn.called)
(self.l2gw_agent_manager.ovsdb_fd.update_connection_to_gateway.
assert_called_with(mock.ANY, mock.ANY, mock.ANY, mock.ANY))
assert_called_with(mock.ANY, mock.ANY, mock.ANY, mock.ANY,
mock.ANY))
def test_delete_network_for_monitor_agent(self):
"""Test case to test delete_network with enable_manager."""
@ -314,7 +331,7 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.delete_network(
self.context, mock.Mock(), "fake_logical_switch_uuid")
(self.l2gw_agent_manager.ovsdb_fd.delete_logical_switch.
assert_called_with("fake_logical_switch_uuid", False))
assert_called_with("fake_logical_switch_uuid", mock.ANY, False))
def test_delete_network_for_transact_agent(self):
"""Test case to test delete_network with enable_manager."""
@ -331,10 +348,10 @@ class TestManager(base.BaseTestCase):
self.l2gw_agent_manager.delete_network(
self.context, mock.Mock(), "fake_logical_switch_uuid")
(self.l2gw_agent_manager.ovsdb_fd._echo_response.
assert_called_with())
assert_called_with(mock.ANY))
self.assertTrue(mock_open_conn.called)
(self.l2gw_agent_manager.ovsdb_fd.delete_logical_switch.
assert_called_with("fake_logical_switch_uuid", False))
assert_called_with("fake_logical_switch_uuid", mock.ANY, False))
def test_add_vif_to_gateway_for_monitor_agent(self):
"""Test case to test add_vif_to_gateway with enable_manager."""
@ -349,7 +366,8 @@ class TestManager(base.BaseTestCase):
"fake_locator_dict", "fake_mac_dict")
(self.l2gw_agent_manager.ovsdb_fd.insert_ucast_macs_remote.
assert_called_with("fake_logical_switch_dict",
"fake_locator_dict", "fake_mac_dict", False))
"fake_locator_dict", "fake_mac_dict",
mock.ANY, False))
def test_add_vif_to_gateway_for_transact_agent(self):
"""Test case to test add_vif_to_gateway with enable_manager."""
@ -367,11 +385,12 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), "fake_logical_switch_dict",
"fake_locator_dict", "fake_mac_dict")
(self.l2gw_agent_manager.ovsdb_fd._echo_response.
assert_called_with())
assert_called_with(mock.ANY))
self.assertTrue(mock_open_conn.called)
(self.l2gw_agent_manager.ovsdb_fd.insert_ucast_macs_remote.
assert_called_with("fake_logical_switch_dict",
"fake_locator_dict", "fake_mac_dict"))
"fake_locator_dict", "fake_mac_dict",
mock.ANY))
def test_delete_vif_from_gateway_for_monitor_agent(self):
"""Test case to test delete_vif_to_gateway with enable_manager."""
@ -385,7 +404,8 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), "fake_logical_switch_uuid",
"fake_mac")
(self.l2gw_agent_manager.ovsdb_fd.delete_ucast_macs_remote.
assert_called_with("fake_logical_switch_uuid", "fake_mac", False))
assert_called_with("fake_logical_switch_uuid", "fake_mac",
mock.ANY, False))
def test_delete_vif_to_gateway_for_transact_agent(self):
"""Test case to test delete_vif_to_gateway with enable_manager."""
@ -403,10 +423,11 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), "fake_logical_switch_uuid",
"fake_mac")
(self.l2gw_agent_manager.ovsdb_fd._echo_response.
assert_called_with())
assert_called_with(mock.ANY))
self.assertTrue(mock_open_conn.called)
(self.l2gw_agent_manager.ovsdb_fd.delete_ucast_macs_remote.
assert_called_with("fake_logical_switch_uuid", "fake_mac"))
assert_called_with("fake_logical_switch_uuid", "fake_mac",
mock.ANY))
def test_update_vif_from_gateway_for_monitor_agent(self):
"""Test case to test update_vif_to_gateway with enable_manager."""
@ -421,7 +442,7 @@ class TestManager(base.BaseTestCase):
"fake_logical_switch_uuid", "fake_mac")
(self.l2gw_agent_manager.ovsdb_fd.update_ucast_macs_remote.
assert_called_with(
"fake_logical_switch_uuid", "fake_mac", False))
"fake_logical_switch_uuid", "fake_mac", mock.ANY, False))
def test_update_vif_to_gateway_for_transact_agent(self):
"""Test case to test update_vif_to_gateway
@ -442,7 +463,8 @@ class TestManager(base.BaseTestCase):
self.context, mock.Mock(), "fake_logical_switch_uuid",
"fake_mac")
(self.l2gw_agent_manager.ovsdb_fd._echo_response.
assert_called_with())
assert_called_with(mock.ANY))
self.assertTrue(mock_open_conn.called)
(self.l2gw_agent_manager.ovsdb_fd.update_ucast_macs_remote.
assert_called_with("fake_logical_switch_uuid", "fake_mac"))
assert_called_with("fake_logical_switch_uuid", "fake_mac",
mock.ANY))

39
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_ovsdb_monitor.py
Unescape Escape View File

@ -23,7 +23,6 @@ import mock
from neutron.tests import base
from networking_l2gw.services.l2gateway.agent import l2gateway_config as conf
from networking_l2gw.services.l2gateway.agent.ovsdb import base_connection
from networking_l2gw.services.l2gateway.agent.ovsdb import ovsdb_monitor
from networking_l2gw.services.l2gateway.common import config
from networking_l2gw.services.l2gateway.common import constants as n_const
@ -169,8 +168,8 @@ class TestOVSDBMonitor(base.BaseTestCase):
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'_process_update_event'
) as process_update_event:
self.l2gw_ovsdb._update_event_handler(self.msg)
process_update_event.assert_called_once_with(self.msg)
self.l2gw_ovsdb._update_event_handler(self.msg, mock.ANY)
process_update_event.assert_called_once_with(self.msg, mock.ANY)
def test_process_update_event(self):
"""Test case to test _process_update_event."""
@ -200,7 +199,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
proc_mcast_mac_local,
proc_phys_loc_set):
self.l2gw_ovsdb._setup_dispatch_table()
self.l2gw_ovsdb._process_update_event(self.msg2)
self.l2gw_ovsdb._process_update_event(self.msg2, mock.ANY)
self.assertTrue(proc_phy_port.called)
self.assertTrue(proc_phy_switch.called)
self.assertTrue(proc_logic_switch.called)
@ -238,7 +237,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
'id': 'fake_id'}
with mock.patch.object(ovsdb_monitor.OVSDBMonitor,
'send') as send:
self.l2gw_ovsdb._default_echo_handler(dummy_msg)
self.l2gw_ovsdb._default_echo_handler(dummy_msg, mock.ANY)
self.assertTrue(send.called)
def test_set_handler(self):
@ -321,7 +320,7 @@ class TestOVSDBMonitor(base.BaseTestCase):
'_get_list',
return_value=some_value
):
result = self.l2gw_ovsdb._form_ovsdb_data(mock.Mock())
result = self.l2gw_ovsdb._form_ovsdb_data(mock.Mock(), mock.ANY)
self.assertEqual(expect, result)
def test_process_physical_port(self):
@ -762,35 +761,9 @@ class TestOVSDBMonitor_with_enable_manager(base.BaseTestCase):
self.l2gw_ovsdb = ovsdb_monitor.OVSDBMonitor(mock.Mock(),
self.conf,
self.callback)
fakesocket = SocketClass()
self.l2gw_ovsdb.c_sock = fakesocket
def test_init_with_enable_manager(self):
self.l2gw_ovsdb.__init__(mock.Mock(), self.conf,
self.callback)
self.assertTrue(self.l2gw_ovsdb.enable_manager)
self.assertFalse(self.l2gw_ovsdb.check_monitor_thread)
self.assertIsNone(self.l2gw_ovsdb.check_c_sock)
def test_sock_rcv_thread_none(self):
with contextlib.nested(
mock.patch.object(base_connection.BaseConnection,
'_echo_response'),
mock.patch.object(eventlet.greenthread, 'spawn_n'),
mock.patch.object(eventlet.greenthread, 'sleep'),
mock.patch.object(self.l2gw_ovsdb.c_sock,
'recv', return_value=None),
mock.patch.object(base_connection.BaseConnection,
'disconnect')) as (
mock_resp, green_thrd_spawn, green_thrd_sleep,
mock_rcv, mock_disconnect):
self.l2gw_ovsdb.check_c_sock = True
self.l2gw_ovsdb.read_on = True
self.l2gw_ovsdb._sock_rcv_thread()
self.assertTrue(mock_resp.called)
self.assertTrue(green_thrd_spawn.called)
self.assertTrue(green_thrd_sleep.called)
self.assertTrue(mock_rcv.called)
self.assertTrue(mock_disconnect.called)
self.assertFalse(self.l2gw_ovsdb.connected)
self.assertFalse(self.l2gw_ovsdb.read_on)
self.assertFalse(self.l2gw_ovsdb.check_monitor_table_thread)

49
networking_l2gw/tests/unit/services/l2gateway/agent/ovsdb/test_ovsdb_writer.py
Unescape Escape View File

@ -123,7 +123,7 @@ class TestOVSDBWriter(base.BaseTestCase):
mock.patch.object(ovsdb_writer.LOG,
'debug')
) as (recv_data, proc_response, debug):
self.l2gw_ovsdb._get_reply(self.op_id)
self.l2gw_ovsdb._get_reply(self.op_id, mock.ANY)
self.assertTrue(recv_data.called)
self.assertTrue(proc_response.called)
@ -135,9 +135,9 @@ class TestOVSDBWriter(base.BaseTestCase):
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_reply') as mock_reply:
self.l2gw_ovsdb._send_and_receive('some_query',
self.op_id, True)
mock_send.assert_called_with('some_query')
mock_reply.assert_called_with(self.op_id)
self.op_id, mock.ANY, True)
mock_send.assert_called_with('some_query', addr=mock.ANY)
mock_reply.assert_called_with(self.op_id, mock.ANY)
def test_send_and_receive_with_rcv_required_false(self):
"""Test case to test _send_and_receive."""
@ -147,8 +147,8 @@ class TestOVSDBWriter(base.BaseTestCase):
with mock.patch.object(ovsdb_writer.OVSDBWriter,
'_get_reply') as mock_reply:
self.l2gw_ovsdb._send_and_receive('some_query',
self.op_id, False)
mock_send.assert_called_with('some_query')
self.op_id, mock.ANY, False)
mock_send.assert_called_with('some_query', addr=mock.ANY)
mock_reply.assert_not_called()
def test_delete_logical_switch(self):
@ -173,9 +173,10 @@ class TestOVSDBWriter(base.BaseTestCase):
with mock.patch.object(ovsdb_writer.LOG,
'debug'):
self.l2gw_ovsdb.delete_logical_switch(
'fake_logical_switch_uuid')
'fake_logical_switch_uuid', mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(query, self.op_id, True)
send_n_receive.assert_called_with(query, self.op_id,
mock.ANY, True)
def test_insert_ucast_macs_remote(self):
"""Test case to test insert ucast_macs_remote."""
@ -203,10 +204,11 @@ class TestOVSDBWriter(base.BaseTestCase):
mock_ucmr):
self.l2gw_ovsdb.insert_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock())
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, True)
self.op_id, mock.ANY, True)
self.assertTrue(get_ucast_mac_remote.called)
@ -248,7 +250,8 @@ class TestOVSDBWriter(base.BaseTestCase):
ls.name = 'ab-cd'
self.l2gw_ovsdb.insert_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock(),
mock.MagicMock())
mock.MagicMock(),
mock.ANY)
self.assertTrue(get_ucast_mac_remote.called)
self.assertTrue(get_physical_locator_dict.called)
self.assertTrue(get_logical_switch_dict.called)
@ -276,10 +279,11 @@ class TestOVSDBWriter(base.BaseTestCase):
mock_pl,
mock_ucmr):
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock())
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, True)
self.op_id, mock.ANY, True)
self.assertTrue(get_update_ucast_mac_remote.called)
@ -313,7 +317,8 @@ class TestOVSDBWriter(base.BaseTestCase):
locator = mock_pl.return_value
locator.uuid = None
self.l2gw_ovsdb.update_ucast_macs_remote(mock.MagicMock(),
mock.MagicMock())
mock.MagicMock(),
mock.ANY)
self.assertTrue(get_update_ucast_mac_remote.called)
self.assertTrue(get_physical_locator_dict.called)
@ -333,10 +338,11 @@ class TestOVSDBWriter(base.BaseTestCase):
send_n_receive,
mock_log):
self.l2gw_ovsdb.delete_ucast_macs_remote(mock.Mock(),
mock.MagicMock())
mock.MagicMock(),
mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, True)
self.op_id, mock.ANY, True)
def test_update_connection_to_gateway(self):
"""Test case to test update_connection_to_gateway."""
@ -357,10 +363,11 @@ class TestOVSDBWriter(base.BaseTestCase):
send_n_receive,
mock_log):
self.l2gw_ovsdb.update_connection_to_gateway(
mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock())
mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock(),
mock.ANY, mock.ANY)
get_rand.assert_called_with(128)
send_n_receive.assert_called_with(mock.ANY,
self.op_id, True)
self.op_id, mock.ANY, True)
self.assertTrue(get_bindings.called)
def test_get_bindings_to_update1(self):
@ -473,7 +480,7 @@ class TestOVSDBWriter(base.BaseTestCase):
ovsdb_conf = base_test.FakeConf()
l2gw_obj = ovsdb_writer.OVSDBWriter(
cfg.CONF.ovsdb, ovsdb_conf)
result = l2gw_obj._recv_data()
result = l2gw_obj._recv_data(mock.ANY)
self.assertEqual(jsonutils.dumps(fake_data), result)
def test_recv_data_with_empty_data(self):
@ -490,7 +497,7 @@ class TestOVSDBWriter(base.BaseTestCase):
ovsdb_conf = base_test.FakeConf()
l2gw_obj = ovsdb_writer.OVSDBWriter(
cfg.CONF.ovsdb, ovsdb_conf)
result = l2gw_obj._recv_data()
result = l2gw_obj._recv_data(mock.ANY)
self.assertEqual(None, result)
def test_recv_data_with_socket_error(self):
@ -507,7 +514,7 @@ class TestOVSDBWriter(base.BaseTestCase):
ovsdb_conf = base_test.FakeConf()
l2gw_obj = ovsdb_writer.OVSDBWriter(
cfg.CONF.ovsdb, ovsdb_conf)
result = l2gw_obj._recv_data()
result = l2gw_obj._recv_data(mock.ANY)
self.assertEqual(None, result)
fake_warn.assert_called_with(
_LW("Did not receive any reply from the OVSDB "

Write Preview
Loading…
Cancel
Save