mirror of
https://github.com/ZwareBear/awx.git
synced 2026-05-16 07:48:38 -05:00
Adds unit tests for network_ui
* Covers 100% of non-migration python lines of code
This commit is contained in:
Ben Thomasson
@@ -0,0 +1,240 @@
|
||||
|
||||
import mock
|
||||
import logging
|
||||
import json
|
||||
import imp
|
||||
from mock import patch
|
||||
patch('channels.auth.channel_session_user', lambda x: x).start()
|
||||
patch('channels.auth.channel_session_user_from_http', lambda x: x).start()
|
||||
|
||||
from awx.network_ui.consumers import parse_inventory_id, networking_events_dispatcher, send_snapshot # noqa
|
||||
from awx.network_ui.models import Topology, Device, Link, Interface, TopologyInventory, Client # noqa
|
||||
import awx # noqa
|
||||
import awx.network_ui # noqa
|
||||
import awx.network_ui.consumers # noqa
|
||||
imp.reload(awx.network_ui.consumers)
|
||||
|
||||
|
||||
def test_parse_inventory_id():
|
||||
assert parse_inventory_id({}) is None
|
||||
assert parse_inventory_id({'inventory_id': ['1']}) == 1
|
||||
assert parse_inventory_id({'inventory_id': ['0']}) is None
|
||||
assert parse_inventory_id({'inventory_id': ['X']}) is None
|
||||
assert parse_inventory_id({'inventory_id': []}) is None
|
||||
assert parse_inventory_id({'inventory_id': 'x'}) is None
|
||||
assert parse_inventory_id({'inventory_id': '12345'}) == 1
|
||||
assert parse_inventory_id({'inventory_id': 1}) is None
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message1():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle({})
|
||||
log_mock.assert_called_once_with(
|
||||
'Unsupported message %s: no topology', {})
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message2():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle({'topology': [0]})
|
||||
log_mock.assert_called_once_with(
|
||||
'Unsupported message %s: no client', {'topology': [0]})
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message3():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle({'topology': [1]})
|
||||
log_mock.assert_called_once_with(
|
||||
'Unsupported message %s: no client', {'topology': [1]})
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message4():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle({'topology': 1, 'client': 1})
|
||||
log_mock.assert_called_once_with('Unsupported message %s: no data', {
|
||||
'client': 1, 'topology': 1})
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message5():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
message = ['DeviceCreate']
|
||||
networking_events_dispatcher.handle(
|
||||
{'topology': 1, 'client': 1, 'text': json.dumps(message)})
|
||||
log_mock.assert_called_once_with('Unsupported message %s: no message type', {
|
||||
'text': '["DeviceCreate"]', 'client': 1, 'topology': 1})
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message6():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
message = ['DeviceCreate', []]
|
||||
networking_events_dispatcher.handle(
|
||||
{'topology': 1, 'client': 1, 'text': json.dumps(message)})
|
||||
log_mock.assert_has_calls([
|
||||
mock.call('Message has no sender'),
|
||||
mock.call('Unsupported message %s: no message type', {'text': '["DeviceCreate", []]', 'client': 1, 'topology': 1})])
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message7():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
message = ['DeviceCreate', {}]
|
||||
networking_events_dispatcher.handle(
|
||||
{'topology': 1, 'client': 1, 'text': json.dumps(message)})
|
||||
log_mock.assert_has_calls([
|
||||
mock.call('client_id mismatch expected: %s actual %s', 1, None),
|
||||
mock.call('Unsupported message %s: no message type', {'text': '["DeviceCreate", {}]', 'client': 1, 'topology': 1})])
|
||||
|
||||
|
||||
def test_network_events_handle_message_incomplete_message8():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
message = ['Unsupported', {'sender': 1}]
|
||||
networking_events_dispatcher.handle(
|
||||
{'topology': 1, 'client': 1, 'text': json.dumps(message)})
|
||||
log_mock.assert_called_once_with(
|
||||
'Unsupported message %s: no handler', u'Unsupported')
|
||||
|
||||
|
||||
def test_send_snapshot_empty():
|
||||
channel = mock.MagicMock()
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects'),\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects'),\
|
||||
mock.patch.object(Topology, 'objects'):
|
||||
send_snapshot(channel, 1)
|
||||
log_mock.assert_not_called()
|
||||
channel.send.assert_called_once_with(
|
||||
{'text': '["Snapshot", {"links": [], "devices": [], "sender": 0}]'})
|
||||
|
||||
|
||||
def test_send_snapshot_single():
|
||||
channel = mock.MagicMock()
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock:
|
||||
|
||||
interface_objects_mock.filter.return_value.values.return_value = [
|
||||
dict(cid=1, device_id=1, id=1, name="eth0")]
|
||||
device_objects_mock.filter.return_value.values.return_value = [
|
||||
dict(cid=1, id=1, device_type="host", name="host1", x=0, y=0,
|
||||
interface_id_seq=1, host_id=1)]
|
||||
send_snapshot(channel, 1)
|
||||
device_objects_mock.filter.assert_called_once_with(topology_id=1)
|
||||
device_objects_mock.filter.return_value.values.assert_called_once_with()
|
||||
interface_objects_mock.filter.assert_called_once_with(
|
||||
device__topology_id=1)
|
||||
interface_objects_mock.filter.return_value.values.assert_called_once_with()
|
||||
log_mock.assert_not_called()
|
||||
channel.send.assert_called_once_with(
|
||||
{'text': '''["Snapshot", {"links": [], "devices": [{"interface_id_seq": 1, \
|
||||
"name": "host1", "interfaces": [{"id": 1, "device_id": 1, "name": "eth0", "interface_id": 1}], \
|
||||
"device_type": "host", "host_id": 1, "y": 0, "x": 0, "id": 1, "device_id": 1}], "sender": 0}]'''})
|
||||
|
||||
|
||||
def test_ws_disconnect():
|
||||
message = mock.MagicMock()
|
||||
message.channel_session = dict(topology_id=1)
|
||||
message.reply_channel = 'foo'
|
||||
with mock.patch('channels.Group') as group_mock:
|
||||
awx.network_ui.consumers.ws_disconnect(message)
|
||||
group_mock.assert_called_once_with('topology-1')
|
||||
group_mock.return_value.discard.assert_called_once_with('foo')
|
||||
|
||||
|
||||
def test_ws_disconnect_no_topology():
|
||||
message = mock.MagicMock()
|
||||
with mock.patch('channels.Group') as group_mock:
|
||||
awx.network_ui.consumers.ws_disconnect(message)
|
||||
group_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_ws_message():
|
||||
message = mock.MagicMock()
|
||||
message.channel_session = dict(topology_id=1, client_id=1)
|
||||
message.__getitem__.return_value = json.dumps([])
|
||||
print (message['text'])
|
||||
with mock.patch('channels.Group') as group_mock:
|
||||
awx.network_ui.consumers.ws_message(message)
|
||||
group_mock.assert_called_once_with('topology-1')
|
||||
group_mock.return_value.send.assert_called_once_with({'text': '[]'})
|
||||
|
||||
|
||||
def test_ws_connect_unauthenticated():
|
||||
message = mock.MagicMock()
|
||||
message.user.is_authenticated.return_value = False
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch.object(logger, 'error') as log_mock:
|
||||
awx.network_ui.consumers.ws_connect(message)
|
||||
log_mock.assert_called_once_with('Request user is not authenticated to use websocket.')
|
||||
|
||||
|
||||
def test_ws_connect_new_topology():
|
||||
message = mock.MagicMock()
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch('awx.network_ui.consumers.Client') as client_mock,\
|
||||
mock.patch('awx.network_ui.consumers.Topology') as topology_mock,\
|
||||
mock.patch('channels.Group'),\
|
||||
mock.patch('awx.network_ui.consumers.send_snapshot') as send_snapshot_mock,\
|
||||
mock.patch.object(logger, 'warning'),\
|
||||
mock.patch.object(TopologyInventory, 'objects'),\
|
||||
mock.patch.object(TopologyInventory, 'save'),\
|
||||
mock.patch.object(Topology, 'save'),\
|
||||
mock.patch.object(Topology, 'objects'),\
|
||||
mock.patch.object(Device, 'objects'),\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects'):
|
||||
client_mock.return_value.pk = 777
|
||||
topology_mock.return_value = Topology(
|
||||
name="topology", scale=1.0, panX=0, panY=0, pk=999)
|
||||
awx.network_ui.consumers.ws_connect(message)
|
||||
message.reply_channel.send.assert_has_calls([
|
||||
mock.call({'text': '["id", 777]'}),
|
||||
mock.call({'text': '["topology_id", 999]'}),
|
||||
mock.call(
|
||||
{'text': '["Topology", {"scale": 1.0, "name": "topology", "device_id_seq": 0, "panY": 0, "panX": 0, "topology_id": 999, "link_id_seq": 0}]'}),
|
||||
])
|
||||
send_snapshot_mock.assert_called_once_with(message.reply_channel, 999)
|
||||
|
||||
|
||||
def test_ws_connect_existing_topology():
|
||||
message = mock.MagicMock()
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
with mock.patch('awx.network_ui.consumers.Client') as client_mock,\
|
||||
mock.patch('awx.network_ui.consumers.send_snapshot') as send_snapshot_mock,\
|
||||
mock.patch('channels.Group'),\
|
||||
mock.patch.object(logger, 'warning'),\
|
||||
mock.patch.object(TopologyInventory, 'objects') as topology_inventory_objects_mock,\
|
||||
mock.patch.object(TopologyInventory, 'save'),\
|
||||
mock.patch.object(Topology, 'save'),\
|
||||
mock.patch.object(Topology, 'objects') as topology_objects_mock,\
|
||||
mock.patch.object(Device, 'objects'),\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects'):
|
||||
topology_inventory_objects_mock.filter.return_value.values_list.return_value = [
|
||||
1]
|
||||
client_mock.return_value.pk = 888
|
||||
topology_objects_mock.get.return_value = Topology(pk=1001,
|
||||
id=1,
|
||||
name="topo",
|
||||
panX=0,
|
||||
panY=0,
|
||||
scale=1.0,
|
||||
link_id_seq=1,
|
||||
device_id_seq=1)
|
||||
awx.network_ui.consumers.ws_connect(message)
|
||||
message.reply_channel.send.assert_has_calls([
|
||||
mock.call({'text': '["id", 888]'}),
|
||||
mock.call({'text': '["topology_id", 1001]'}),
|
||||
mock.call(
|
||||
{'text': '["Topology", {"scale": 1.0, "name": "topo", "device_id_seq": 1, "panY": 0, "panX": 0, "topology_id": 1001, "link_id_seq": 1}]'}),
|
||||
])
|
||||
send_snapshot_mock.assert_called_once_with(message.reply_channel, 1001)
|
||||
@@ -0,0 +1,15 @@
|
||||
|
||||
|
||||
from awx.network_ui.models import Device, Topology, Interface
|
||||
|
||||
|
||||
def test_device():
|
||||
assert str(Device(name="foo")) == "foo"
|
||||
|
||||
|
||||
def test_topology():
|
||||
assert str(Topology(name="foo")) == "foo"
|
||||
|
||||
|
||||
def test_interface():
|
||||
assert str(Interface(name="foo")) == "foo"
|
||||
@@ -0,0 +1,451 @@
|
||||
import mock
|
||||
import json
|
||||
import logging
|
||||
|
||||
from awx.network_ui.consumers import networking_events_dispatcher
|
||||
from awx.network_ui.models import Topology, Device, Link, Interface
|
||||
|
||||
|
||||
def message(message):
|
||||
def wrapper(fn):
|
||||
fn.tests_message = message
|
||||
return fn
|
||||
return wrapper
|
||||
|
||||
|
||||
@message('DeviceMove')
|
||||
def test_network_events_handle_message_DeviceMove():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceMove', dict(
|
||||
msg_type='DeviceMove',
|
||||
sender=1,
|
||||
id=1,
|
||||
x=100,
|
||||
y=100,
|
||||
previous_x=0,
|
||||
previous_y=0
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid=1, topology_id=1)
|
||||
device_objects_mock.filter.return_value.update.assert_called_once_with(
|
||||
x=100, y=100)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('DeviceCreate')
|
||||
def test_network_events_handle_message_DeviceCreate():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceCreate', dict(msg_type='DeviceCreate',
|
||||
sender=1,
|
||||
id=1,
|
||||
x=0,
|
||||
y=0,
|
||||
name="test_created",
|
||||
type='host',
|
||||
host_id=None)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Topology.objects, 'filter') as topology_objects_mock,\
|
||||
mock.patch.object(Device.objects, 'get_or_create') as device_objects_mock:
|
||||
device_mock = mock.MagicMock()
|
||||
filter_mock = mock.MagicMock()
|
||||
device_objects_mock.return_value = [device_mock, True]
|
||||
topology_objects_mock.return_value = filter_mock
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.assert_called_once_with(
|
||||
cid=1,
|
||||
defaults={'name': u'test_created', 'cid': 1, 'device_type': u'host',
|
||||
'x': 0, 'y': 0, 'host_id': None},
|
||||
topology_id=1)
|
||||
device_mock.save.assert_called_once_with()
|
||||
topology_objects_mock.assert_called_once_with(
|
||||
device_id_seq__lt=1, pk=1)
|
||||
filter_mock.update.assert_called_once_with(device_id_seq=1)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('DeviceLabelEdit')
|
||||
def test_network_events_handle_message_DeviceLabelEdit():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceLabelEdit', dict(
|
||||
msg_type='DeviceLabelEdit',
|
||||
sender=1,
|
||||
id=1,
|
||||
name='test_changed',
|
||||
previous_name='test_created'
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device.objects, 'filter') as device_objects_filter_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_filter_mock.assert_called_once_with(
|
||||
cid=1, topology_id=1)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('DeviceSelected')
|
||||
def test_network_events_handle_message_DeviceSelected():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceSelected', dict(
|
||||
msg_type='DeviceSelected',
|
||||
sender=1,
|
||||
id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('DeviceUnSelected')
|
||||
def test_network_events_handle_message_DeviceUnSelected():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceUnSelected', dict(
|
||||
msg_type='DeviceUnSelected',
|
||||
sender=1,
|
||||
id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('DeviceDestroy')
|
||||
def test_network_events_handle_message_DeviceDestory():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['DeviceDestroy', dict(
|
||||
msg_type='DeviceDestroy',
|
||||
sender=1,
|
||||
id=1,
|
||||
previous_x=0,
|
||||
previous_y=0,
|
||||
previous_name="",
|
||||
previous_type="host",
|
||||
previous_host_id="1")]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid=1, topology_id=1)
|
||||
device_objects_mock.filter.return_value.delete.assert_called_once_with()
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('InterfaceCreate')
|
||||
def test_network_events_handle_message_InterfaceCreate():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['InterfaceCreate', dict(
|
||||
msg_type='InterfaceCreate',
|
||||
sender=1,
|
||||
device_id=1,
|
||||
id=1,
|
||||
name='eth0'
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock:
|
||||
device_objects_mock.get.return_value.pk = 99
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.get.assert_called_once_with(cid=1, topology_id=1)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid=1, interface_id_seq__lt=1, topology_id=1)
|
||||
interface_objects_mock.get_or_create.assert_called_once_with(
|
||||
cid=1, defaults={'name': u'eth0'}, device_id=99)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('InterfaceLabelEdit')
|
||||
def test_network_events_handle_message_InterfaceLabelEdit():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['InterfaceLabelEdit', dict(
|
||||
msg_type='InterfaceLabelEdit',
|
||||
sender=1,
|
||||
id=1,
|
||||
device_id=1,
|
||||
name='new name',
|
||||
previous_name='old name'
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
interface_objects_mock.filter.assert_called_once_with(
|
||||
cid=1, device__cid=1, device__topology_id=1)
|
||||
interface_objects_mock.filter.return_value.update.assert_called_once_with(
|
||||
name=u'new name')
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('LinkLabelEdit')
|
||||
def test_network_events_handle_message_LinkLabelEdit():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkLabelEdit', dict(
|
||||
msg_type='LinkLabelEdit',
|
||||
sender=1,
|
||||
id=1,
|
||||
name='new name',
|
||||
previous_name='old name'
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Link, 'objects') as link_objects_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
link_objects_mock.filter.assert_called_once_with(
|
||||
cid=1, from_device__topology_id=1)
|
||||
link_objects_mock.filter.return_value.update.assert_called_once_with(
|
||||
name=u'new name')
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('LinkCreate')
|
||||
def test_network_events_handle_message_LinkCreate():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkCreate', dict(
|
||||
msg_type='LinkCreate',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Link, 'objects') as link_objects_mock,\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock,\
|
||||
mock.patch.object(Topology, 'objects') as topology_objects_mock:
|
||||
values_list_mock = mock.MagicMock()
|
||||
values_list_mock.values_list.return_value = [(1,1), (2,2)]
|
||||
interface_objects_mock.get.return_value = mock.MagicMock()
|
||||
interface_objects_mock.get.return_value.pk = 7
|
||||
device_objects_mock.filter.return_value = values_list_mock
|
||||
topology_objects_mock.filter.return_value = mock.MagicMock()
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid__in=[1, 2], topology_id=1)
|
||||
values_list_mock.values_list.assert_called_once_with('cid', 'pk')
|
||||
link_objects_mock.get_or_create.assert_called_once_with(
|
||||
cid=1, from_device_id=1, from_interface_id=7, name=u'',
|
||||
to_device_id=2, to_interface_id=7)
|
||||
topology_objects_mock.filter.assert_called_once_with(
|
||||
link_id_seq__lt=1, pk=1)
|
||||
topology_objects_mock.filter.return_value.update.assert_called_once_with(
|
||||
link_id_seq=1)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('LinkCreate')
|
||||
def test_network_events_handle_message_LinkCreate_bad_device1():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkCreate', dict(
|
||||
msg_type='LinkCreate',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock,\
|
||||
mock.patch.object(Topology, 'objects') as topology_objects_mock:
|
||||
values_list_mock = mock.MagicMock()
|
||||
values_list_mock.values_list.return_value = [(9,1), (2,2)]
|
||||
interface_objects_mock.get.return_value = mock.MagicMock()
|
||||
interface_objects_mock.get.return_value.pk = 7
|
||||
device_objects_mock.filter.return_value = values_list_mock
|
||||
topology_objects_mock.filter.return_value = mock.MagicMock()
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid__in=[1, 2], topology_id=1)
|
||||
values_list_mock.values_list.assert_called_once_with('cid', 'pk')
|
||||
log_mock.assert_called_once_with('Device not found')
|
||||
|
||||
|
||||
@message('LinkCreate')
|
||||
def test_network_events_handle_message_LinkCreate_bad_device2():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkCreate', dict(
|
||||
msg_type='LinkCreate',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Link, 'objects'),\
|
||||
mock.patch.object(Interface, 'objects') as interface_objects_mock,\
|
||||
mock.patch.object(Topology, 'objects') as topology_objects_mock:
|
||||
values_list_mock = mock.MagicMock()
|
||||
values_list_mock.values_list.return_value = [(1,1), (9,2)]
|
||||
interface_objects_mock.get.return_value = mock.MagicMock()
|
||||
interface_objects_mock.get.return_value.pk = 7
|
||||
device_objects_mock.filter.return_value = values_list_mock
|
||||
topology_objects_mock.filter.return_value = mock.MagicMock()
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_objects_mock.filter.assert_called_once_with(
|
||||
cid__in=[1, 2], topology_id=1)
|
||||
values_list_mock.values_list.assert_called_once_with('cid', 'pk')
|
||||
log_mock.assert_called_once_with('Device not found')
|
||||
|
||||
|
||||
@message('LinkDestroy')
|
||||
def test_network_events_handle_message_LinkDestroy():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkDestroy', dict(
|
||||
msg_type='LinkDestroy',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device.objects, 'filter') as device_filter_mock,\
|
||||
mock.patch.object(Link.objects, 'filter') as link_filter_mock,\
|
||||
mock.patch.object(Interface.objects, 'get') as interface_get_mock:
|
||||
values_mock = mock.MagicMock()
|
||||
interface_get_mock.return_value = mock.MagicMock()
|
||||
interface_get_mock.return_value.pk = 7
|
||||
device_filter_mock.return_value = values_mock
|
||||
values_mock.values_list.return_value = [(1,1), (2,2)]
|
||||
networking_events_dispatcher.handle(message)
|
||||
device_filter_mock.assert_called_once_with(
|
||||
cid__in=[1, 2], topology_id=1)
|
||||
values_mock.values_list.assert_called_once_with('cid', 'pk')
|
||||
link_filter_mock.assert_called_once_with(
|
||||
cid=1, from_device_id=1, from_interface_id=7, to_device_id=2, to_interface_id=7)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('LinkDestroy')
|
||||
def test_network_events_handle_message_LinkDestroy_bad_device_map1():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkDestroy', dict(
|
||||
msg_type='LinkDestroy',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device.objects, 'filter') as device_filter_mock,\
|
||||
mock.patch.object(Link.objects, 'filter'),\
|
||||
mock.patch.object(Interface.objects, 'get') as interface_get_mock:
|
||||
values_mock = mock.MagicMock()
|
||||
interface_get_mock.return_value = mock.MagicMock()
|
||||
interface_get_mock.return_value.pk = 7
|
||||
device_filter_mock.return_value = values_mock
|
||||
values_mock.values_list.return_value = [(9,1), (2,2)]
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_called_once_with('Device not found')
|
||||
|
||||
|
||||
@message('LinkDestroy')
|
||||
def test_network_events_handle_message_LinkDestroy_bad_device_map2():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkDestroy', dict(
|
||||
msg_type='LinkDestroy',
|
||||
id=1,
|
||||
sender=1,
|
||||
name="",
|
||||
from_device_id=1,
|
||||
to_device_id=2,
|
||||
from_interface_id=1,
|
||||
to_interface_id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock,\
|
||||
mock.patch.object(Device.objects, 'filter') as device_filter_mock,\
|
||||
mock.patch.object(Link.objects, 'filter'),\
|
||||
mock.patch.object(Interface.objects, 'get') as interface_get_mock:
|
||||
values_mock = mock.MagicMock()
|
||||
interface_get_mock.return_value = mock.MagicMock()
|
||||
interface_get_mock.return_value.pk = 7
|
||||
device_filter_mock.return_value = values_mock
|
||||
values_mock.values_list.return_value = [(1,1), (9,2)]
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_called_once_with('Device not found')
|
||||
|
||||
|
||||
@message('LinkSelected')
|
||||
def test_network_events_handle_message_LinkSelected():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkSelected', dict(
|
||||
msg_type='LinkSelected',
|
||||
sender=1,
|
||||
id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('LinkUnSelected')
|
||||
def test_network_events_handle_message_LinkUnSelected():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['LinkUnSelected', dict(
|
||||
msg_type='LinkUnSelected',
|
||||
sender=1,
|
||||
id=1
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_not_called()
|
||||
|
||||
|
||||
@message('MultipleMessage')
|
||||
def test_network_events_handle_message_MultipleMessage_unsupported_message():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['MultipleMessage', dict(
|
||||
msg_type='MultipleMessage',
|
||||
sender=1,
|
||||
messages=[dict(msg_type="Unsupported")]
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_called_once_with(
|
||||
'Unsupported message %s', u'Unsupported')
|
||||
|
||||
|
||||
@message('MultipleMessage')
|
||||
def test_network_events_handle_message_MultipleMessage():
|
||||
logger = logging.getLogger('awx.network_ui.consumers')
|
||||
message_data = ['MultipleMessage', dict(
|
||||
msg_type='MultipleMessage',
|
||||
sender=1,
|
||||
messages=[dict(msg_type="DeviceSelected")]
|
||||
)]
|
||||
message = {'topology': 1, 'client': 1, 'text': json.dumps(message_data)}
|
||||
with mock.patch.object(logger, 'warning') as log_mock:
|
||||
networking_events_dispatcher.handle(message)
|
||||
log_mock.assert_not_called()
|
||||
@@ -0,0 +1,9 @@
|
||||
|
||||
import awx.network_ui.routing
|
||||
|
||||
|
||||
def test_routing():
|
||||
'''
|
||||
Tests that the number of routes in awx.network_ui.routing is 3.
|
||||
'''
|
||||
assert len(awx.network_ui.routing.channel_routing) == 3
|
||||
@@ -0,0 +1,65 @@
|
||||
|
||||
import mock
|
||||
|
||||
from awx.network_ui.views import topology_data, NetworkAnnotatedInterface, json_topology_data, yaml_topology_data
|
||||
from awx.network_ui.models import Topology, Device, Link, Interface
|
||||
|
||||
|
||||
|
||||
def test_topology_data():
|
||||
with mock.patch.object(Topology, 'objects'),\
|
||||
mock.patch.object(Device, 'objects') as device_objects_mock,\
|
||||
mock.patch.object(Link, 'objects') as link_objects_mock,\
|
||||
mock.patch.object(Interface, 'objects'),\
|
||||
mock.patch.object(NetworkAnnotatedInterface, 'filter'):
|
||||
device_objects_mock.filter.return_value.order_by.return_value = [
|
||||
Device(pk=1), Device(pk=2)]
|
||||
link_objects_mock.filter.return_value = [Link(from_device=Device(name='from', cid=1),
|
||||
to_device=Device(
|
||||
name='to', cid=2),
|
||||
from_interface=Interface(
|
||||
name="eth0", cid=1),
|
||||
to_interface=Interface(
|
||||
name="eth0", cid=1),
|
||||
name="",
|
||||
pk=1
|
||||
)]
|
||||
data = topology_data(1)
|
||||
assert len(data['devices']) == 2
|
||||
assert len(data['links']) == 1
|
||||
|
||||
|
||||
def test_json_topology_data():
|
||||
request = mock.MagicMock()
|
||||
request.GET = dict(topology_id=1)
|
||||
with mock.patch('awx.network_ui.views.topology_data') as topology_data_mock:
|
||||
topology_data_mock.return_value = dict()
|
||||
json_topology_data(request)
|
||||
topology_data_mock.assert_called_once_with(1)
|
||||
|
||||
|
||||
def test_yaml_topology_data():
|
||||
request = mock.MagicMock()
|
||||
request.GET = dict(topology_id=1)
|
||||
with mock.patch('awx.network_ui.views.topology_data') as topology_data_mock:
|
||||
topology_data_mock.return_value = dict()
|
||||
yaml_topology_data(request)
|
||||
topology_data_mock.assert_called_once_with(1)
|
||||
|
||||
|
||||
def test_json_topology_data_no_topology_id():
|
||||
request = mock.MagicMock()
|
||||
request.GET = dict()
|
||||
with mock.patch('awx.network_ui.views.topology_data') as topology_data_mock:
|
||||
topology_data_mock.return_value = dict()
|
||||
json_topology_data(request)
|
||||
topology_data_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_yaml_topology_data_no_topology_id():
|
||||
request = mock.MagicMock()
|
||||
request.GET = dict()
|
||||
with mock.patch('awx.network_ui.views.topology_data') as topology_data_mock:
|
||||
topology_data_mock.return_value = dict()
|
||||
yaml_topology_data(request)
|
||||
topology_data_mock.assert_not_called()
|
||||
Reference in New Issue
Block a user