Restricted real mongo interaction.
Still work to do, but it's a good start.
Fixed a bug with old subnets not being properly deleted.
Change-Id: I5f260e09f0e11a477a47cb031d397a454465123f
Signed-off-by: Ilia Abashin <abashinos@gmail.com>
mongo*.conf
mongo_access.log
.DS_Store
-
-app/test/event_based_scan/config/test_config.py
router_doc['gw_port_id'] = router['gw_port_id']
# add gateway port documents.
- port_doc = EventSubnetAdd().add_port_document(env, router_doc['gw_port_id'], project_name=project)
+ port_doc = EventSubnetAdd().add_port_document(env,
+ router_doc['gw_port_id'],
+ project_name=project)
mac_address = port_doc['mac_address'] if port_doc else None
subnet_id = interface['subnet_id']
router_id = encode_router_id(host_id, interface['id'])
- network_document = self.inv.get_by_field(env, "network", "subnet_ids", subnet_id, get_single=True)
+ network_document = self.inv.get_by_field(env, "network", "subnet_ids",
+ subnet_id, get_single=True)
if not network_document:
self.log.info("network document not found, aborting interface adding")
return EventResult(result=False, retry=True)
fetcher = ApiFetchRegions()
fetcher.set_env(env)
fetcher.get(None)
- port_doc = EventSubnetAdd().add_port_document(env, port_id, network_name=network_name)
+ port_doc = EventSubnetAdd().add_port_document(env, port_id,
+ network_name=network_name)
mac_address = port_doc['mac_address'] if port_doc else None
# Check DHCP enable, if true, scan network.
if subnet['enable_dhcp'] is True:
# update network
- # TODO: #AskCheng - why is this necessary?
if len(ApiAccess.regions) == 0:
fetcher = ApiFetchRegions()
fetcher.set_env(env)
if subnet['name'] == subnets[key]['name']:
subnets[key] = subnet
else:
- # TODO: #AskCheng shouldn't we remove the old one?
+ del subnets[key]
subnets[subnet['name']] = subnet
self.inv.set(network_document)
return EventResult(result=True,
related_object=subnet['id'],
display_context=network_id)
+ else:
+ self.log.info(
+ 'subnet not in network, aborting subnet update')
+ return EventResult(result=False, retry=False)
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_INSTANCE_ADD = {
'publisher_id': 'compute.node-251.cisco.com', '_context_resource_uuid': None,
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_INSTANCE_DELETE = {
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_INSTANCE_UPDATE = {
'id_path': '/'+ENV_CONFIG+'/'+ENV_CONFIG+'-regions/RegionOne/RegionOne-availability_zones/calipso-zone' +
'/node-223.cisco.com/node-223.cisco.com-instances/27a87908-bc1b-45cc-9238-09ad1ae686a7',
'show_in_tree': True}
+
+
+UPDATED_INSTANCE_FIELDS = {
+ 'name': 'test8',
+ 'object_name': 'test8',
+ 'name_path': '/'+ENV_CONFIG+'/Regions/RegionOne/Availability Zones' +
+ '/calipso-zone/node-223.cisco.com/Instances/test8'}
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_INTERFACE_ADD = {
'_context_timestamp': '2016-10-26 21:52:18.893134', '_context_project_name': 'calipso-project',
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_INTERFACE_DELETE = {
'message_id': 'da190e5f-127d-4e85-a813-bbdbbb35a2d0', '_context_tenant': '75c0eb79ff4a42b0ae4973c8375ddf40',
'tenant_id': '75c0eb79ff4a42b0ae4973c8375ddf40', 'mtu': 1400, 'subnets': [], 'status': 'ACTIVE',
'provider:segmentation_id': 8, 'port_security_enabled': True, 'name': 'calipso-network-add',
'admin_state_up': True}}}
+
+NETWORK_DOCUMENT = {'provider:physical_network': None,
+ 'router:external': False,
+ 'shared': False,
+ 'id': 'a8226605-40d0-4111-93bd-11ffa5b2d1d7',
+ 'provider:network_type': 'vxlan',
+ 'tenant_id': '75c0eb79ff4a42b0ae4973c8375ddf40',
+ 'mtu': 1400, 'subnets': {}, 'status': 'ACTIVE',
+ 'provider:segmentation_id': 8,
+ 'port_security_enabled': True,
+ 'name': 'calipso-network-add',
+ 'admin_state_up': True, 'environment': 'test-env',
+ 'type': 'network',
+ 'id_path': '/test-env/test-env-projects/75c0eb79ff4a42b0ae4973c8375ddf40/75c0eb79ff4a42b0ae4973c8375ddf40-networks/a8226605-40d0-4111-93bd-11ffa5b2d1d7',
+ 'cidrs': [], 'subnet_ids': [],
+ 'last_scanned': '2016-09-30 17:45:02.125633',
+ 'name_path': '/test-env/Projects/calipso-project/Networks/calipso-network-add',
+ 'network': 'a8226605-40d0-4111-93bd-11ffa5b2d1d7',
+ 'object_name': 'calipso-network-add',
+ 'parent_id': '75c0eb79ff4a42b0ae4973c8375ddf40-networks',
+ 'parent_text': 'Networks', 'parent_type': 'networks_folder',
+ 'project': 'calipso-project', 'show_in_tree': True}
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_NETWORK_DELETE = {
'_context_user': '13baa553aae44adca6615e711fd2f6d9', 'publisher_id': 'network.node-6.cisco.com'}
-EVENT_PAYLOAD_NETWORK = {
+NETWORK_DOCUMENT = {
"admin_state_up" : True,
"cidrs" : [
"172.16.9.0/24"
],
"environment" : ENV_CONFIG,
+ "_id": '583c0c69c5f6980fec665422',
"id" : '0bb0ba6c-6863-4121-ac89-93f81a9da2b0',
"id_path" : '/%s/%s-projects/' % (ENV_CONFIG, ENV_CONFIG) +'75c0eb79ff4a42b0ae4973c8375ddf40/75c0eb79ff4a42b' +
'0ae4973c8375ddf40-networks/0bb0ba6c-6863-4121-ac89-93f81a9da2b0' ,
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_NETWORK_UPDATE = {
'_context_user_id': '13baa553aae44adca6615e711fd2f6d9', '_context_user': '13baa553aae44adca6615e711fd2f6d9',
"subnets" : {},
"tenant_id" : "75c0eb79ff4a42b0ae4973c8375ddf40",
"type" : "network"
-}
\ No newline at end of file
+}
+
+UPDATED_NETWORK_FIELDS = {'name': '24',
+ 'name_path': '/{}/Projects/calipso-project/Networks/24'.format(ENV_CONFIG),
+ 'object_name': '24'}
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_PORT_INSTANCE_ADD = {
'_context_user_id': '73638a2687534f9794cd8057ba860637', 'payload': {
"show_in_tree": True,
"vnic_type": "instance_vnic"
}]
+
+PORTS_FOLDER = {'parent_id': '55550a69-24eb-47f5-a458-3aa086cc71c2',
+ 'create_object': True,
+ 'text': 'Ports',
+ 'show_in_tree': True,
+ 'id_path': 'test-env/test-env-projects/a83c8b0d2df24170a7c54f09f824230e/a83c8b0d2df24170a7c54f09f824230e-networks/55550a69-24eb-47f5-a458-3aa086cc71c2/55550a69-24eb-47f5-a458-3aa086cc71c2-ports/',
+ 'name_path': '/test-env/Projects/a83c8b0d2df24170a7c54f09f824230e/Networks/please_connect/Ports',
+ 'environment': ENV_CONFIG,
+ 'id': '55550a69-24eb-47f5-a458-3aa086cc71c2-ports',
+ 'name': 'Ports', 'parent_type': 'network',
+ 'type': 'ports_folder', 'object_name': 'Ports'}
+
+PORT_DOC = {'id': '1233445-75b6-4c05-9480-4bc648845c6f'}
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_PORT_DELETE = {
'_context_tenant': '75c0eb79ff4a42b0ae4973c8375ddf40', '_context_tenant_name': 'calipso-project',
"type": "port"
}
-VNIC_DOCS = [{
+VNIC_DOC = {
"IP Address": "172.16.10.2",
"IPv6 Address": "fe80::f816:3eff:fe96:5066/64",
"cidr": "172.16.10.0/25",
"environment": ENV_CONFIG,
"host": "node-251.cisco.com",
"id": "tapca33c645-5b",
+ '_id': '5970b9aa797ffad322bc9b84',
"id_path": "/" + ENV_CONFIG + "/" + ENV_CONFIG + "-regions/RegionOne/RegionOne-availability_zones/internal" +
"/node-251.cisco.com/node-251.cisco.com-vservices/node-251.cisco.com-vservices-dhcps/qdhcp-911fe57e-" +
"1ddd-4151-9dc7-6b578ab357b1/qdhcp-911fe57e-1ddd-4151-9dc7-6b578ab357b1-vnics/tapca33c645-5b",
"show_in_tree": True,
"type": "vnic",
"vnic_type": "vservice_vnic"
-}]
+}
INSTANCE_DOC = {
"environment": ENV_CONFIG,
- "id": "b2bda4bf-1259-4d60-99ab-85ab4d5014a8",
"type": "instance",
"uuid": "b2bda4bf-1259-4d60-99ab-85ab4d5014a8",
"network": [
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_PORT_UPDATE = {
'_context_timestamp': '2016-10-25 21:27:05.591848', '_context_user_name': 'admin',
###############################################################################
import datetime
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_ROUTER_ADD = {
'_context_show_deleted': False, '_context_domain': None,
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_ROUTER_DELETE = {
'_context_request_id': 'req-8b2dd9ba-5faa-4471-94c3-fb41781eef8d', '_unique_id': 'c7417f771ee74bb19036b06e685c93dc',
"environment": ENV_CONFIG,
"gw_port_id": None,
"host": "node-6.cisco.com",
+ "_id": "593fc4c6797ffad322bc5329",
"id": "node-6.cisco.com-qrouter-bde87a5a-7968-4f3b-952c-e87681a96078",
"id_path": "/" + ENV_CONFIG + "/" + ENV_CONFIG + "-regions/RegionOne/RegionOne-availability_zones/internal" +
"/node-6.cisco.com/node-6.cisco.com-vservices/node-6.cisco.com-vservices-routers/qrouter-bde87a5a" +
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_ROUTER_UPDATE = {
'_context_request_id': 'req-da45908c-0765-4f8a-9fac-79246901de41', '_unique_id': '80723cc09a4748c6b13214dcb867719e',
'tenant_id': '75c0eb79ff4a42b0ae4973c8375ddf40', 'status': 'ACTIVE'}},
'_context_request_id': 'req-d7e73189-4709-4234-8b4c-fb6b4dc2017b'}
-PORTS = {
+PORT = {
"admin_state_up": True,
"allowed_address_pairs": [
###############################################################################
import datetime
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
NETWORK_DOC = {'port_security_enabled': True, 'status': 'ACTIVE', 'subnet_ids': [], 'parent_type': 'networks_folder',
'parent_id': '75c0eb79ff4a42b0ae4973c8375ddf40-networks', 'parent_text': 'Networks', 'subnets': {},
'show_in_tree': True,
'id_path': '/' + ENV_CONFIG + '/' + ENV_CONFIG + '-regions/RegionOne',
'type': 'region'}}
+
+
+HOST_DOC = {
+ "environment": ENV_CONFIG,
+ "host": "node-6.cisco.com",
+ "host_type": [
+ "Controller",
+ "Network"
+ ],
+ "id": "node-6.cisco.com",
+ "id_path": "/" + ENV_CONFIG + "/" + ENV_CONFIG + "-regions/RegionOne/RegionOne-availability_zones" +
+ "/internal/node-6.cisco.com",
+ "name": "node-6.cisco.com",
+ "name_path": "/" + ENV_CONFIG + "/Regions/RegionOne/Availability Zones/internal/node-6.cisco.com",
+ "object_name": "node-6.cisco.com",
+ "parent_id": "internal",
+ "parent_type": "availability_zone",
+ "show_in_tree": True,
+ "type": "host",
+ "zone": "internal"
+}
+
+PORT_DOC = {
+ "admin_state_up": True,
+ "device_id": "c57216ca-c1c4-430d-a045-32851ca879e3",
+ "device_owner": "compute:nova",
+ "dns_assignment": [
+ {
+ "hostname": "host-172-16-10-1",
+ "ip_address": "172.16.10.1",
+ "fqdn": "host-172-16-10-1.openstacklocal."
+ }
+ ],
+ "dns_name": "",
+ "environment": ENV_CONFIG,
+ "extra_dhcp_opts": [
+
+ ],
+ "fixed_ips": [
+ {
+ "ip_address": "172.16.10.1",
+ "subnet_id": "6f6ef3b5-76c9-4f70-81e5-f3cc196db025"
+ }
+ ],
+ "id": "2233445-55b6-4c05-9480-4bc648845c6f",
+ "id_path": ENV_CONFIG + "/" + ENV_CONFIG + "-projects/75c0eb79ff4a42b0ae4973c8375ddf40/75c0eb79ff4a42b0ae4973c837" +
+ "5ddf40-networks/1bb0ba6c-6863-4121-ac89-93f81a9da2b0/1bb0ba6c-6863-4121-ac89-93f81a9da2b0-ports" +
+ "/2233445-55b6-4c05-9480-4bc648845c6f",
+ "last_scanned": 0,
+ "mac_address": "fa:16:3e:13:b2:aa",
+ "master_parent_id": "1bb0ba6c-6863-4121-ac89-93f81a9da2b0",
+ "master_parent_type": "network",
+ "name": "fa:16:3e:13:b2:aa",
+ "name_path": "/" + ENV_CONFIG + "/Projects/calipso-project/Networks/test_interface/Ports/" +
+ "2233445-55b6-4c05-9480-4bc648845c6f",
+ "network_id": "1bb0ba6c-6863-4121-ac89-93f81a9da2b0",
+ "object_name": "2233445-55b6-4c05-9480-4bc648845c6f",
+ "parent_id": "1bb0ba6c-6863-4121-ac89-93f81a9da2b0-ports",
+ "parent_text": "Ports",
+ "parent_type": "ports_folder",
+ "port_security_enabled": False,
+ "project": "calipso-project",
+ "security_groups": [
+
+ ],
+ "status": "DOWN",
+ "tenant_id": "75c0eb79ff4a42b0ae4973c8375ddf40",
+ "type": "port"
+}
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
EVENT_PAYLOAD_SUBNET_DELETE = {
'payload': {'subnet_id': '88442b4a-e62d-4d72-9d18-b8d6973eb3da'},
'_context_project_id': '75c0eb79ff4a42b0ae4973c8375ddf40'}
-EVENT_PAYLOAD_NETWORK = {
+NETWORK_DOC = {
"admin_state_up": True,
"cidrs": [
"172.16.10.0/25"
"tenant_id": "75c0eb79ff4a42b0ae4973c8375ddf40",
"type": "network"
}
+
+VNIC_DOC = {
+ "IP Address": "172.16.10.2",
+ "IPv6 Address": "fe80::f816:3eff:fe96:5066/64",
+ "cidr": "172.16.10.0/25",
+ "data": "Link encap:Ethernet HWaddr fa:16:3e:96:50:66\ninet addr:172.16.10.2 Bcast:172.16.10.127 " +
+ "Mask:255.255.255.128\ninet6 addr: fe80::f816:3eff:fe96:5066/64 Scope:Link\nUP BROADCAST RUNNING " +
+ "MULTICAST MTU:1450 Metric:1\nRX packets:17 errors:0 dropped:2 overruns:0 frame:0\nTX packets:8 " +
+ "errors:0 dropped:0 overruns:0 carrier:0\ncollisions:0 txqueuelen:0\nRX bytes:1593 " +
+ "(1.5 KB) TX bytes:648 (648.0 B)\n",
+ "environment": ENV_CONFIG,
+ "host": "node-6.cisco.com",
+ "id": "tapca33c645-5b",
+ '_id': '5970b9aa797ffad322bc9b84',
+ "id_path": "/" + ENV_CONFIG + "/" + ENV_CONFIG + "-regions/RegionOne/RegionOne-availability_zones/internal" +
+ "/node-6.cisco.com/node-6.cisco.com-vservices/node-6.cisco.com-vservices-dhcps/qdhcp-121c727b-6376-4a86-a5a8-793dfe7a8ef4/qdhcp-121c727b-6376-4a86-a5a8-793dfe7a8ef4-vnics/tapca33c645-5b",
+ "last_scanned": 0,
+ "mac_address": "fa:16:3e:13:b2:aa",
+ "name": "tapca33c645-5b",
+ "name_path": "/"+ENV_CONFIG+"/Regions/RegionOne/Availability Zones/internal/node-6.cisco.com/" +
+ "Vservices/DHCP servers/dhcp-test_interface/vNICs/tapca33c645-5b",
+ "netmask": "255.255.255.128",
+ "network": "121c727b-6376-4a86-a5a8-793dfe7a8ef4",
+ "object_name": "tapca33c645-5b",
+ "parent_id": "qdhcp-121c727b-6376-4a86-a5a8-793dfe7a8ef4-vnics",
+ "parent_text": "vNICs",
+ "parent_type": "vnics_folder",
+ "show_in_tree": True,
+ "type": "vnic",
+ "vnic_type": "vservice_vnic"
+}
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from test.event_based_scan.config.test_config import ENV_CONFIG
+from test.event_based_scan.test_data.test_config import ENV_CONFIG
NETWORK_DOC = {
'port_security_enabled': True, 'status': 'ACTIVE',
'_context_auth_token': 'gAAAAABYDp0ZacwkUNIRvtiS-3qjLQFZKbkOtTmvuoKX9yM8yCIvl-eZmMC_SPjwPAMJcd8qckE77lLpQSx0l'+
'WB67mT5jQA-tmp8bcz26kXXr8KlGCicxxjkYTYkJQhC9w8BbGc36CpbRBzIKlOrPtPXUYZrUmPgInQqCNA-eD'+
'eMyJ-AiA1zmNSZK3R43YIJtnDYieLQvX2P'}
+
+
+HOST_DOC = {
+ "environment": ENV_CONFIG,
+ "host": "node-6.cisco.com",
+ "host_type": [
+ "Controller",
+ "Network"
+ ],
+ "id": "node-6.cisco.com",
+ "id_path": "/" + ENV_CONFIG + "/" + ENV_CONFIG + "-regions/RegionOne/RegionOne-availability_zones" +
+ "/internal/node-6.cisco.com",
+ "name": "node-6.cisco.com",
+ "name_path": "/" + ENV_CONFIG + "/Regions/RegionOne/Availability Zones/internal/node-6.cisco.com",
+ "object_name": "node-6.cisco.com",
+ "parent_id": "internal",
+ "parent_type": "availability_zone",
+ "show_in_tree": True,
+ "type": "host",
+ "zone": "internal"
+}
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
+# local config info for test.
+ENV_CONFIG = 'test-env'
+
+COLLECTION_CONFIG = 'test-collection'
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-import re
import unittest
+from unittest.mock import patch, Mock
-from discover.configuration import Configuration
-from test.event_based_scan.config.test_config \
- import MONGODB_CONFIG, ENV_CONFIG, COLLECTION_CONFIG
-from utils.inventory_mgr import InventoryMgr
+from test.event_based_scan.test_data.test_config \
+ import ENV_CONFIG, COLLECTION_CONFIG
from utils.logging.console_logger import ConsoleLogger
-from utils.mongo_access import MongoAccess
class TestEvent(unittest.TestCase):
def setUp(self):
self.log = ConsoleLogger()
- self.mongo_config = MONGODB_CONFIG
self.env = ENV_CONFIG
self.collection = COLLECTION_CONFIG
-
- MongoAccess.set_config_file(self.mongo_config)
- self.conf = Configuration()
- self.conf.use_env(self.env)
-
- self.inv = InventoryMgr()
- self.inv.set_collections(self.collection)
self.item_ids = []
- def set_item(self, document):
- self.inv.set(document)
- self.item_ids.append(document['id'])
+ self.inv_patcher = patch('discover.events.event_base.InventoryMgr')
+ self.inv_class = self.inv_patcher.start()
+ self.inv = self.inv_class.return_value
- def assert_empty_by_id(self, object_id):
- doc = self.inv.get_by_id(self.env, object_id)
- self.assertIsNone(doc)
+ self.log_patcher = patch('discover.fetcher.FullLogger')
+ self.log_patcher.start()
def tearDown(self):
- for item_id in self.item_ids:
- item = self.inv.get_by_id(self.env, item_id)
- # delete children
- if item:
- regexp = re.compile('^{}/'.format(item['id_path']))
- self.inv.delete('inventory', {'id_path': {'$regex': regexp}})
-
- # delete target item
- self.inv.delete('inventory', {'id': item_id})
- item = self.inv.get_by_id(self.env, item_id)
- self.assertIsNone(item)
+ self.inv_patcher.stop()
+ self.log_patcher.stop()
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
+from unittest.mock import patch, call
+
+import re
from bson import ObjectId
-from discover.clique_finder import CliqueFinder
from discover.events.event_base import EventBase
from test.event_based_scan.test_event import TestEvent
super().setUp()
self.values = {}
- def set_item_for_deletion(self, object_type, document):
-
- payload = self.values['payload']
- self.item_id = payload['{}_id'.format(object_type)]
- if object_type == 'router':
- host_id = self.values['publisher_id'].replace("network.", "", 1)
- self.item_id = "-".join([host_id, "qrouter", self.item_id])
-
- self.assertEqual(document['id'], self.item_id, msg="Document id and payload id are different")
-
- item = self.inv.get_by_id(self.env, self.item_id)
- if not item:
- self.log.info('{} document is not found, add document for deleting.'.format(object_type))
+ self.cf = patch("discover.events.event_delete_base.CliqueFinder")
+ self.clique_finder = self.cf.start().return_value
+ self.clique_finder.find_links_by_source.return_value = []
+ self.clique_finder.find_links_by_target.return_value = []
- # add network document for deleting.
- self.set_item(document)
- item = self.inv.get_by_id(self.env, self.item_id)
- self.assertIsNotNone(item)
+ def handle_delete(self, handler: EventBase, db_object: dict):
+ with patch("discover.events.event_delete_base.CliqueFinder") as cf:
+ self.inv.get_by_id.return_value = db_object
- def handle_delete(self, handler: EventBase):
+ event_result = handler.handle(self.env, self.values)
+ self.assertTrue(event_result.result)
- item = self.inv.get_by_id(self.env, self.item_id)
- db_id = ObjectId(item['_id'])
- clique_finder = CliqueFinder()
+ self.check_inv_calls(db_object)
- # delete item
- event_result = handler.handle(self.env, self.values)
- self.assertTrue(event_result.result)
+ def check_inv_calls(self, db_object):
+ db_id = ObjectId(db_object['_id'])
+ id_path_regex = re.compile('^{}/'.format(db_object['id_path']))
- # check instance delete result.
- item = self.inv.get_by_id(self.env, self.item_id)
- self.assertIsNone(item)
+ delete_clique = call('cliques', {'focal_point': db_id})
+ delete_source_links = call('links', {'source': db_id})
+ delete_target_links = call('links', {'target': db_id})
+ delete_object = call('inventory', {'_id': db_id})
+ delete_id_path = call('inventory', {'id_path':
+ {'$regex': id_path_regex}})
- # check links
- matched_links_source = clique_finder.find_links_by_source(db_id)
- matched_links_target = clique_finder.find_links_by_target(db_id)
- self.assertEqual(matched_links_source.count(), 0)
- self.assertEqual(matched_links_target.count(), 0)
+ self.inv.delete.assert_has_calls([delete_clique,
+ delete_source_links,
+ delete_target_links,
+ delete_object,
+ delete_id_path])
- # check children
- matched_children = self.inv.get_children(self.env, None, self.item_id)
- self.assertEqual(len(matched_children), 0)
+ def tearDown(self):
+ super().tearDown()
+ self.cf.stop()
\ No newline at end of file
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import patch
+from unittest.mock import patch, ANY, call
from discover.events.event_instance_add import EventInstanceAdd
from test.event_based_scan.test_data.event_payload_instance_add \
- import EVENT_PAYLOAD_INSTANCE_ADD, INSTANCES_ROOT, HOST, INSTANCE_DOCUMENT
+ import EVENT_PAYLOAD_INSTANCE_ADD, INSTANCES_ROOT, HOST
from test.event_based_scan.test_event import TestEvent
class TestInstanceAdd(TestEvent):
- def insert_instance(self):
- self.set_item(INSTANCE_DOCUMENT)
+ def get_by_id(self, env, object_id):
+ instance_root_id = '-'.join((self.payload['host'], 'instances'))
+ if object_id == instance_root_id:
+ return INSTANCES_ROOT
+ elif object_id:
+ return HOST
+ else:
+ return None
# Patch ScanHost entirely to negate its side effects and supply our own
- @patch("discover.events.event_instance_add.ScanHost")
- def test_handle_instance_add(self, scan_host_mock):
+ @patch("discover.events.event_instance_add.Scanner")
+ def test_handle_instance_add(self, scanner_mock):
self.values = EVENT_PAYLOAD_INSTANCE_ADD
- payload = self.values['payload']
- self.instance_id = payload['instance_id']
- host_id = payload['host']
+ self.payload = self.values['payload']
+ instance_id = self.payload['instance_id']
- # prepare instances root, in case it's not there
- self.set_item(INSTANCES_ROOT)
+ self.inv.get_by_id.side_effect = self.get_by_id
- # prepare host, in case it's not existed.
- self.set_item(HOST)
-
- # check instance document
- instance = self.inv.get_by_id(self.env, self.instance_id)
- if instance:
- self.log.info('instance document exists, delete it first.')
- self.inv.delete('inventory', {'id': self.instance_id})
-
- instance = self.inv.get_by_id(self.env, self.instance_id)
- self.assertIsNone(instance)
-
- # simulate instance insertion after host scan
- scan_host_mock.return_value.scan_links.side_effect = self.insert_instance
-
- # check the return of instance handler.
handler = EventInstanceAdd()
ret = handler.handle(self.env, self.values)
self.assertEqual(ret.result, True)
- # check host document
- host = self.inv.get_by_id(self.env, host_id)
- self.assertIsNotNone(host)
+ root_call = call(ANY, INSTANCES_ROOT,
+ limit_to_child_id=instance_id,
+ limit_to_child_type='instance')
+ host_call = call(ANY, HOST,
+ limit_to_child_type=ANY)
- # check instance document
- instance_document = self.inv.get_by_id(self.env, self.instance_id)
- self.assertIsNotNone(instance_document)
+ scanner = scanner_mock.return_value
+ scanner.scan.assert_has_calls([root_call, host_call])
def setUp(self):
super().setUp()
self.values = EVENT_PAYLOAD_INSTANCE_DELETE
- self.set_item_for_deletion(object_type="instance", document=INSTANCE_DOCUMENT)
def test_handle_instance_delete(self):
- self.handle_delete(handler=EventInstanceDelete())
+ self.handle_delete(handler=EventInstanceDelete(),
+ db_object=INSTANCE_DOCUMENT)
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
from discover.events.event_instance_update import EventInstanceUpdate
-from test.event_based_scan.test_data.event_payload_instance_update import EVENT_PAYLOAD_INSTANCE_UPDATE, INSTANCE_DOCUMENT
+from test.event_based_scan.test_data.event_payload_instance_update \
+ import EVENT_PAYLOAD_INSTANCE_UPDATE, INSTANCE_DOCUMENT, \
+ UPDATED_INSTANCE_FIELDS
from test.event_based_scan.test_event import TestEvent
self.values = EVENT_PAYLOAD_INSTANCE_UPDATE
payload = self.values['payload']
self.instance_id = payload['instance_id']
- self.item_ids.append(self.instance_id)
- new_name = payload['display_name']
- # preparing instance to be updated
- instance = self.inv.get_by_id(self.env, self.instance_id)
- if not instance:
- self.log.info("instance document is not found, add document for updating")
+ instance = INSTANCE_DOCUMENT
- # add instance document for updating
- self.set_item(INSTANCE_DOCUMENT)
- instance = self.inv.get_by_id(self.env, self.instance_id)
- self.assertIsNotNone(instance)
- self.assertEqual(instance['name'], INSTANCE_DOCUMENT['name'])
-
- name_path = instance['name_path']
- new_name_path = name_path[:name_path.rindex('/') + 1] + new_name
+ self.inv.get_by_id.return_value = instance
# update instance document
- EventInstanceUpdate().handle(self.env, self.values)
+ res = EventInstanceUpdate().handle(self.env, self.values)
- # get new document
- instance = self.inv.get_by_id(self.env, self.instance_id)
+ self.assertTrue(res.result)
+ self.assertTrue(self.inv.values_replace.called)
+ self.assertTrue(self.inv.set.called)
- # check update result.
- self.assertEqual(instance['name'], new_name)
- self.assertEqual(instance['name_path'], new_name_path)
+ # check that all changed fields are updated
+ call_args, _ = self.inv.set.call_args
+ self.assertTrue(all(item in call_args[0].items()
+ for item in UPDATED_INSTANCE_FIELDS.items()))
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+import copy
+from unittest.mock import MagicMock, patch, Mock
+from discover.events.event_base import EventResult
from discover.events.event_interface_add import EventInterfaceAdd
-from discover.fetchers.api.api_access import ApiAccess
-from discover.fetchers.api.api_fetch_port import ApiFetchPort
-from discover.fetchers.cli.cli_fetch_host_vservice import CliFetchHostVservice
-from discover.fetchers.cli.cli_fetch_vservice_vnics import CliFetchVserviceVnics
-from discover.find_links_for_vservice_vnics import FindLinksForVserviceVnics
-from test.event_based_scan.test_data.event_payload_interface_add import EVENT_PAYLOAD_INTERFACE_ADD, NETWORK_DOC, \
+from test.event_based_scan.test_data.event_payload_interface_add import \
+ EVENT_PAYLOAD_INTERFACE_ADD, NETWORK_DOC, \
EVENT_PAYLOAD_REGION, PORT_DOC, ROUTER_DOCUMENT, HOST, VNIC_DOCS
from test.event_based_scan.test_event import TestEvent
+from test.event_based_scan.util import TestRegions
from utils.util import encode_router_id
class TestInterfaceAdd(TestEvent):
- def test_handle_interface_add(self):
- self.values = EVENT_PAYLOAD_INTERFACE_ADD
- self.payload = self.values['payload']
- self.interface = self.payload['router_interface']
-
- self.port_id = self.interface['port_id']
- self.host_id = self.values["publisher_id"].replace("network.", "", 1)
- self.router_id = encode_router_id(self.host_id, self.interface['id'])
- self.set_item(NETWORK_DOC)
- ApiAccess.regions = EVENT_PAYLOAD_REGION
+ def get_by_id(self, env, object_id):
+ interface = self.values["payload"]["router_interface"]
+ host_id = self.values["publisher_id"].replace("network.", "", 1)
+ router_id = encode_router_id(host_id, interface['id'])
- # mock port data,
- original_api_get_port = ApiFetchPort.get
- ApiFetchPort.get = MagicMock(return_value=[PORT_DOC])
- self.item_ids.append(PORT_DOC['id'])
+ if object_id == host_id:
+ return HOST
+ elif object_id == router_id:
+ return ROUTER_DOCUMENT
+ elif object_id == ROUTER_DOCUMENT["gw_port_id"]:
+ return PORT_DOC
+ else:
+ return None
- # set router document
- self.set_item(ROUTER_DOCUMENT)
-
- # set host document
- self.set_item(HOST)
+ @patch("discover.events.event_interface_add.FindLinksForVserviceVnics")
+ @patch("discover.events.event_interface_add.Scanner")
+ @patch("discover.events.event_interface_add.CliFetchHostVservice")
+ @patch("discover.events.event_interface_add.EventPortAdd")
+ @patch("discover.events.event_interface_add.EventSubnetAdd")
+ def test_handle_interface_add(self, subnet_add_class_mock,
+ port_add_class_mock,
+ fetcher_class_mock,
+ scanner_class_mock,
+ find_links_class_mock):
+ self.values = EVENT_PAYLOAD_INTERFACE_ADD
- # mock add_links
- original_add_links = FindLinksForVserviceVnics.add_links
- FindLinksForVserviceVnics.add_links = MagicMock()
+ self.inv.get_by_field.return_value = NETWORK_DOC
+ self.inv.get_by_id.side_effect = self.get_by_id
- # mock get_vservice
- original_get_vservice = CliFetchHostVservice.get_vservice
- CliFetchHostVservice.get_vservice = MagicMock(return_value=ROUTER_DOCUMENT)
+ subnet_add_mock = subnet_add_class_mock.return_value
+ subnet_add_mock.add_port_document.return_value = PORT_DOC
- # mock handle_vservice
- original_handle_service = CliFetchVserviceVnics.handle_service
- CliFetchVserviceVnics.handle_service = MagicMock(return_value=VNIC_DOCS)
+ port_add_mock = port_add_class_mock.return_value
+ port_add_mock.add_vnic_document = \
+ Mock(return_value=EventResult(result=True))
- # handle the notification
- EventInterfaceAdd().handle(self.env, self.values)
+ fetcher_mock = fetcher_class_mock.return_value
+ fetcher_mock.get_vservice.return_value = ROUTER_DOCUMENT
+ fetcher_mock.handle_service.return_value = VNIC_DOCS
- # reset the method.
- ApiFetchPort.get = original_api_get_port
- FindLinksForVserviceVnics.add_links = original_add_links
- CliFetchHostVservice.get_vservice = original_get_vservice
- CliFetchVserviceVnics.handle_service = original_handle_service
+ scanner_mock = scanner_class_mock.return_value
+ find_links_mock = find_links_class_mock.return_value
- # check port and router document
- port_doc = self.inv.get_by_id(self.env, self.port_id)
- self.assertIsNotNone(port_doc)
+ with patch("discover.fetcher.FullLogger"):
+ with TestRegions(EVENT_PAYLOAD_REGION):
+ res = EventInterfaceAdd().handle(self.env, self.values)
- router_doc = self.inv.get_by_id(self.env, self.router_id)
- self.assertIn(NETWORK_DOC['id'], router_doc['network'])
+ self.assertTrue(res.result)
+ self.inv.set.assert_called_with(ROUTER_DOCUMENT)
+ self.assertTrue(port_add_mock.add_vnic_document.called)
+ self.assertTrue(scanner_mock.scan_cliques.called)
+ self.assertTrue(find_links_mock.add_links.called)
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
+from unittest.mock import patch
+
+from discover.events.event_base import EventResult
from discover.events.event_interface_delete import EventInterfaceDelete
-from discover.fetchers.api.api_access import ApiAccess
-from test.event_based_scan.test_data.event_payload_interface_delete import EVENT_PAYLOAD_INTERFACE_DELETE, NETWORK_DOC, \
- EVENT_PAYLOAD_REGION, PORT_DOC, ROUTER_DOCUMENT, HOST, VNIC_DOCS
+from test.event_based_scan.test_data.event_payload_interface_delete \
+ import EVENT_PAYLOAD_INTERFACE_DELETE, PORT_DOC, ROUTER_DOCUMENT
from test.event_based_scan.test_event import TestEvent
from utils.util import encode_router_id
class TestInterfaceDelete(TestEvent):
- def test_handle_interface_delete(self):
+
+ def get_by_id(self, env, object_id):
+ if object_id == self.port_id:
+ return PORT_DOC
+ elif object_id == self.router_id:
+ return ROUTER_DOCUMENT
+ else:
+ return None
+
+ @patch("discover.events.event_interface_delete.EventPortDelete")
+ def test_handle_interface_delete(self,
+ port_delete_class_mock):
self.values = EVENT_PAYLOAD_INTERFACE_DELETE
self.payload = self.values['payload']
self.interface = self.payload['router_interface']
-
self.port_id = self.interface['port_id']
self.host_id = self.values["publisher_id"].replace("network.", "", 1)
self.router_id = encode_router_id(self.host_id, self.interface['id'])
- # set document for instance deleting.
- self.set_item(NETWORK_DOC)
- self.set_item(PORT_DOC)
- self.set_item(ROUTER_DOCUMENT)
- self.set_item(HOST)
- self.set_item(VNIC_DOCS[0])
- ApiAccess.regions = EVENT_PAYLOAD_REGION
+ port_delete_mock = port_delete_class_mock.return_value
+ port_delete_mock.delete_port.return_value = EventResult(result=True)
+
+ self.inv.get_by_id.side_effect = self.get_by_id
- # delete interface
- EventInterfaceDelete().handle(self.env, self.values)
+ res = EventInterfaceDelete().handle(self.env, self.values)
- # assert data
- router_doc = self.inv.get_by_id(self.env, ROUTER_DOCUMENT['id'])
- self.assertNotIn(NETWORK_DOC['id'], router_doc['network'])
+ self.assertTrue(res.result)
+ self.assertTrue(port_delete_mock.delete_port.called)
+ self.inv.set.assert_called_with(ROUTER_DOCUMENT)
- self.assert_empty_by_id(PORT_DOC['id'])
- self.assert_empty_by_id(VNIC_DOCS[0]['id'])
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
from discover.events.event_network_add import EventNetworkAdd
-from test.event_based_scan.test_data.event_payload_network_add import EVENT_PAYLOAD_NETWORK_ADD
+from test.event_based_scan.test_data.event_payload_network_add \
+ import EVENT_PAYLOAD_NETWORK_ADD, NETWORK_DOCUMENT
from test.event_based_scan.test_event import TestEvent
def test_handle_network_add(self):
self.values = EVENT_PAYLOAD_NETWORK_ADD
- self.payload = self.values['payload']
- self.network = self.payload['network']
- self.network_id = self.network['id']
- self.item_ids.append(self.network_id)
- network_document = self.inv.get_by_id(self.env, self.network_id)
- if network_document:
- self.log.info('network document existed already, deleting it first.')
- self.inv.delete('inventory', {'id': self.network_id})
-
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIsNone(network_document)
-
- # build network document for adding network
- project_name = self.values['_context_project_name']
- project_id = self.values['_context_project_id']
- parent_id = project_id + '-networks'
- network_name = self.network['name']
+ self.inv.get_by_id.return_value = None
# add network document
- EventNetworkAdd().handle(self.env, self.values)
-
- # check network document
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIsNotNone(network_document)
- self.assertEqual(network_document["project"], project_name)
- self.assertEqual(network_document["parent_id"], parent_id)
- self.assertEqual(network_document["name"], network_name)
+ res = EventNetworkAdd().handle(self.env, self.values)
+ self.assertTrue(res.result)
+ self.inv.set.assert_called_with(NETWORK_DOCUMENT)
###############################################################################
from discover.events.event_network_delete import EventNetworkDelete
from test.event_based_scan.test_data.event_payload_network_delete import EVENT_PAYLOAD_NETWORK_DELETE, \
- EVENT_PAYLOAD_NETWORK
+ NETWORK_DOCUMENT
from test.event_based_scan.test_event_delete_base import TestEventDeleteBase
def setUp(self):
super().setUp()
self.values = EVENT_PAYLOAD_NETWORK_DELETE
- self.set_item_for_deletion(object_type="network", document=EVENT_PAYLOAD_NETWORK)
def test_handle_network_delete(self):
- self.handle_delete(handler=EventNetworkDelete())
+ self.handle_delete(handler=EventNetworkDelete(),
+ db_object=NETWORK_DOCUMENT)
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
from discover.events.event_network_update import EventNetworkUpdate
-from test.event_based_scan.test_data.event_payload_network_update import EVENT_PAYLOAD_NETWORK_UPDATE, \
- NETWORK_DOCUMENT
+from test.event_based_scan.test_data.event_payload_network_update import \
+ EVENT_PAYLOAD_NETWORK_UPDATE, \
+ NETWORK_DOCUMENT, UPDATED_NETWORK_FIELDS
from test.event_based_scan.test_event import TestEvent
def test_handle_network_update(self):
self.values = EVENT_PAYLOAD_NETWORK_UPDATE
self.payload = self.values['payload']
- self.network = self.payload['network']
- name = self.network['name']
- status = self.network['admin_state_up']
- self.network_id = self.network['id']
- self.item_ids.append(self.network_id)
- self.set_item(NETWORK_DOCUMENT)
+ network = NETWORK_DOCUMENT
+ self.inv.get_by_id.return_value = network
- EventNetworkUpdate().handle(self.env, self.values)
+ res = EventNetworkUpdate().handle(self.env, self.values)
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertEqual(network_document['name'], name)
- self.assertEqual(network_document['admin_state_up'], status)
+ self.assertTrue(res.result)
+ self.assertTrue(self.inv.values_replace.called)
+ self.assertTrue(self.inv.set.called)
+
+ # check that all changed fields are updated
+ call_args, _ = self.inv.set.call_args
+ # Assert that all updated fields have been added to db
+ self.assertTrue(all(item in call_args[0].items()
+ for item in UPDATED_NETWORK_FIELDS.items()))
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
from discover.events.event_port_add import EventPortAdd
from discover.fetchers.api.api_fetch_host_instances import ApiFetchHostInstances
from discover.find_links_for_instance_vnics import FindLinksForInstanceVnics
from discover.find_links_for_vedges import FindLinksForVedges
from discover.scanner import Scanner
-from test.event_based_scan.test_data.event_payload_port_add import EVENT_PAYLOAD_PORT_INSTANCE_ADD, NETWORK_DOC, \
- INSTANCE_DOC, INSTANCES_ROOT, VNIC_DOCS, INSTANCE_DOCS
+from test.event_based_scan.test_data.event_payload_port_add import \
+ EVENT_PAYLOAD_PORT_INSTANCE_ADD, NETWORK_DOC, \
+ INSTANCE_DOC, INSTANCES_ROOT, VNIC_DOCS, INSTANCE_DOCS, PORTS_FOLDER, \
+ PORT_DOC
from test.event_based_scan.test_event import TestEvent
class TestPortAdd(TestEvent):
- def test_handle_port_add(self):
+
+ def get_by_id(self, env, object_id):
+ if object_id == self.network_id:
+ return NETWORK_DOC
+ elif object_id == "{}-ports".format(self.network_id):
+ return PORTS_FOLDER
+ elif object_id == self.instance_id:
+ return INSTANCE_DOC
+ elif object_id == "{}-instances".format(self.host_id):
+ return INSTANCES_ROOT
+ elif [call for call in self.inv.set.call_args_list
+ if call[0][0].get('type') == 'port']:
+ self.port_set = True
+ return PORT_DOC
+ else:
+ return None
+
+ @patch("discover.events.event_port_add.Scanner")
+ @patch("discover.events.event_port_add.FindLinksForVedges")
+ @patch("discover.events.event_port_add.FindLinksForInstanceVnics")
+ @patch("discover.events.event_port_add.CliFetchInstanceVnics")
+ @patch("discover.events.event_port_add.ApiFetchHostInstances")
+ def test_handle_port_add(self,
+ api_fetch_instances_class_mock,
+ cli_fetch_instances_class_mock,
+ fl_for_vnics_class_mock,
+ fl_for_vedges_class_mock,
+ scanner_class_mock):
self.values = EVENT_PAYLOAD_PORT_INSTANCE_ADD
self.payload = self.values['payload']
self.port = self.payload['port']
- self.port_id = self.port['id']
- self.item_ids.append(self.port_id)
-
- # prepare data for test
- self.set_item(NETWORK_DOC)
- self.set_item(INSTANCE_DOC)
- self.set_item(INSTANCES_ROOT)
- self.item_ids.append(VNIC_DOCS[0]['id'])
-
- # mock methods
- original_get_instance = ApiFetchHostInstances.get
- ApiFetchHostInstances.get = MagicMock(return_value=INSTANCE_DOCS)
-
- original_get_vnic = CliFetchInstanceVnics.get
- CliFetchInstanceVnics.get = MagicMock(return_value=VNIC_DOCS)
-
- original_find_link_instance = FindLinksForInstanceVnics.add_links
- original_find_link_vedge = FindLinksForVedges.add_links
- original_scan = Scanner.scan_cliques
+ self.host_id = self.port['binding:host_id']
+ self.instance_id = INSTANCE_DOC['id']
+ self.network_id = NETWORK_DOC['id']
- FindLinksForInstanceVnics.add_links = MagicMock(return_value=None)
- FindLinksForVedges.add_links = MagicMock(return_value=None)
- Scanner.scan_cliques = MagicMock(return_value=None)
+ self.inv.get_by_id.side_effect = self.get_by_id
- # add network document
- EventPortAdd().handle(self.env, self.values)
+ api_fetch_instances_mock = api_fetch_instances_class_mock.return_value
+ api_fetch_instances_mock.get.return_value = INSTANCE_DOCS
- # check network document
- port_document = self.inv.get_by_id(self.env, self.port_id)
- self.assertIsNotNone(port_document)
- self.assertEqual(port_document["name"], self.port['name'])
+ cli_fetch_instances_mock = cli_fetch_instances_class_mock.return_value
+ cli_fetch_instances_mock.get.return_value = VNIC_DOCS
- instance = self.inv.get_by_id(self.env, INSTANCE_DOC['id'])
- self.assertEqual(instance["network_info"][0]['devname'],
- INSTANCE_DOCS[0]["network_info"][0]['devname'])
- self.assertEqual(instance["network_info"],
- INSTANCE_DOCS[0]["network_info"])
- self.assertEqual(instance["network"], INSTANCE_DOCS[0]["network"])
+ scanner_mock = scanner_class_mock.return_value
+ fl_for_vnics_mock = fl_for_vnics_class_mock.return_value
+ fl_for_vedges_mock = fl_for_vedges_class_mock.return_value
- vnic = self.inv.get_by_field(self.env, 'vnic', 'mac_address',
- self.port['mac_address'])
- self.assertIsNotNone(vnic)
+ res = EventPortAdd().handle(self.env, self.values)
- FindLinksForVedges.add_links = original_find_link_vedge
- FindLinksForInstanceVnics.add_links = original_find_link_instance
- Scanner.scan_cliques = original_scan
- CliFetchInstanceVnics.get = original_get_vnic
- ApiFetchHostInstances.get = original_get_instance
+ self.assertTrue(res.result)
+ # Assert that port has been added to db
+ self.assertTrue(self.port_set)
+ self.assertTrue(fl_for_vnics_mock.add_links.called)
+ self.assertTrue(fl_for_vedges_mock.add_links.called)
+ self.assertTrue(scanner_mock.scan_cliques.called)
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+from unittest.mock import patch
from discover.events.event_port_delete import EventPortDelete
-from discover.fetchers.api.api_fetch_host_instances import ApiFetchHostInstances
-from test.event_based_scan.test_data.event_payload_port_delete import EVENT_PAYLOAD_PORT_DELETE, PORT_DOC, VNIC_DOCS, \
- INSTANCE_DOC, INSTANCE_DOCS
-from test.event_based_scan.test_event import TestEvent
+from test.event_based_scan.test_data.event_payload_port_delete import \
+ EVENT_PAYLOAD_PORT_DELETE, PORT_DOC, VNIC_DOC, INSTANCE_DOC, INSTANCE_DOCS
+from test.event_based_scan.test_event_delete_base import TestEventDeleteBase
+
+
+class TestPortDelete(TestEventDeleteBase):
+
+ def get_by_id(self, env, object_id):
+ if object_id == PORT_DOC['id']:
+ return PORT_DOC
+ elif object_id == VNIC_DOC['id']:
+ return VNIC_DOC
+ else:
+ return None
+
+ def get_by_field(self, environment, item_type, field_name, field_value,
+ get_single=False):
+ if item_type == "vnic":
+ return VNIC_DOC
+ elif item_type == "instance":
+ return INSTANCE_DOC
+ else:
+ return None
+
+ @patch("discover.events.event_port_delete.ApiFetchHostInstances")
+ def test_handle_port_delete(self,
+ api_fetch_instances_class_mock):
+ self.values = EVENT_PAYLOAD_PORT_DELETE
+ self.inv.get_by_id.side_effect = self.get_by_id
+ self.inv.get_by_field.side_effect = self.get_by_field
-class TestPortDelete(TestEvent):
- def test_handle_port_delete(self):
- self.values = EVENT_PAYLOAD_PORT_DELETE
- self.payload = self.values['payload']
- self.port_id = self.payload['port_id']
- self.item_ids.append(self.port_id)
-
- # set port data firstly.
- self.set_item(PORT_DOC)
- self.set_item(VNIC_DOCS[0])
- self.set_item(INSTANCE_DOC)
-
- # mock methods
- original_get_instance = ApiFetchHostInstances.get
- ApiFetchHostInstances.get = MagicMock(return_value=INSTANCE_DOCS)
- self.item_ids.append(INSTANCE_DOCS[0]['id'])
-
- # delete port
- EventPortDelete().handle(self.env, self.values)
-
- # assert data
- self.assert_empty_by_id(self.port_id)
- self.assert_empty_by_id(VNIC_DOCS[0]['id'])
- instance = self.inv.get_by_id(self.env, INSTANCE_DOC['id'])
- self.assertEqual(instance['mac_address'], None)
- self.assertEqual(instance['network'], [])
- self.assertEqual(instance['network_info'], [])
-
- ApiFetchHostInstances.get = original_get_instance
+ api_fetch_instances_mock = api_fetch_instances_class_mock.return_value
+ api_fetch_instances_mock.get.return_value = INSTANCE_DOCS
+
+ res = EventPortDelete().handle(self.env, self.values)
+
+ self.assertTrue(res.result)
+ self.check_inv_calls(VNIC_DOC)
+ self.inv.delete.assert_any_call('inventory', {'id': PORT_DOC['id']})
class TestPortUpdate(TestEvent):
def test_handle_port_update(self):
+ self.values = EVENT_PAYLOAD_PORT_UPDATE
+ self.port = self.values['payload']['port']
+
+ self.inv.get_by_id.return_value = PORT_DOCUMENT
+
+ res = EventPortUpdate().handle(self.env, self.values)
+
+ self.assertTrue(res.result)
+ self.assertTrue(self.inv.set.called)
+
+ updated_port = self.inv.set.call_args[0][0]
+ self.assertEqual(updated_port["name"],
+ self.port['name'])
+ self.assertEqual(updated_port['admin_state_up'],
+ self.port['admin_state_up'])
+ self.assertEqual(updated_port['binding:vnic_type'],
+ self.port['binding:vnic_type'])
+
+ def _test_handle_port_update(self):
self.values = EVENT_PAYLOAD_PORT_UPDATE
self.payload = self.values['payload']
self.port = self.payload['port']
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
from discover.events.event_port_add import EventPortAdd
from discover.events.event_router_add import EventRouterAdd
class TestRouterAdd(TestEvent):
- def test_handle_router_add(self):
+
+ def get_by_id(self, env, object_id):
+ if object_id == self.host_id:
+ return HOST_DOC
+ elif object_id == self.network_id:
+ return NETWORK_DOC
+ else:
+ return None
+
+ @patch("discover.events.event_router_add.EventPortAdd")
+ @patch("discover.events.event_router_add.EventSubnetAdd")
+ @patch("discover.events.event_router_add.Scanner")
+ @patch("discover.events.event_router_add.FindLinksForVserviceVnics")
+ @patch("discover.events.event_router_add.CliFetchHostVservice")
+ def test_handle_router_add(self,
+ cli_fetch_vservice_class_mock,
+ find_links_class_mock,
+ scanner_class_mock,
+ subnet_add_class_mock,
+ port_add_class_mock):
self.values = EVENT_PAYLOAD_ROUTER_ADD
self.payload = self.values['payload']
self.router = self.payload['router']
+ self.network_id = self.router['external_gateway_info']['network_id']
self.host_id = self.values["publisher_id"].replace("network.", "", 1)
self.router_id = encode_router_id(self.host_id, self.router['id'])
- self.set_item(HOST_DOC)
- self.host_id = HOST_DOC['id']
- gateway_info = self.router['external_gateway_info']
- if gateway_info:
- self.network_id = self.router['external_gateway_info']['network_id']
- self.inv.set(NETWORK_DOC)
-
- original_get_vservice = CliFetchHostVservice.get_vservice
- CliFetchHostVservice.get_vservice = MagicMock(return_value=ROUTER_DOCUMENT)
- self.gw_port_id = ROUTER_DOCUMENT['gw_port_id']
-
- original_add_port = EventSubnetAdd.add_port_document
- EventSubnetAdd.add_port_document = MagicMock()
-
- original_add_vnic = EventPortAdd.add_vnic_document
- EventPortAdd.add_vnic_document = MagicMock()
-
- handler = EventRouterAdd()
- handler.update_links_and_cliques = MagicMock()
+ self.inv.get_by_id.side_effect = self.get_by_id
- handler.handle(self.env, self.values)
+ cli_fetch_vservice_mock = cli_fetch_vservice_class_mock.return_value
+ cli_fetch_vservice_mock.get_vservice.return_value = ROUTER_DOCUMENT
- # reset the methods back
- CliFetchHostVservice.get_vservice = original_get_vservice
- EventSubnetAdd.add_port_document = original_add_port
- EventPortAdd.add_vnic_document = original_add_vnic
+ find_links_mock = find_links_class_mock.return_value
+ scanner_mock = scanner_class_mock.return_value
+ subnet_add_mock = subnet_add_class_mock.return_value
+ subnet_add_mock.add_port_document.return_value = True
+ port_add_mock = port_add_class_mock.return_value
+ port_add_mock.add_vnic_document.return_value = True
+ port_add_mock.add_vnic_folder.return_value = True
- # assert router document
- router_doc = self.inv.get_by_id(self.env, self.router_id)
- self.assertIsNotNone(router_doc, msg="router_doc not found.")
- self.assertEqual(ROUTER_DOCUMENT['name'], router_doc['name'])
- self.assertEqual(ROUTER_DOCUMENT['gw_port_id'], router_doc['gw_port_id'])
+ res = EventRouterAdd().handle(self.env, self.values)
- # assert children documents
- vnics_id = '-'.join(['qrouter', self.router['id'], 'vnics'])
- vnics_folder = self.inv.get_by_id(self.env, vnics_id)
- self.assertIsNotNone(vnics_folder, msg="Vnics folder not found.")
+ self.assertTrue(res.result)
- def tearDown(self):
- self.item_ids = [self.network_id, self.host_id, self.network_id+"-ports", self.gw_port_id,
- self.router_id+'-vnics', self.router_id]
- for item_id in self.item_ids:
- self.inv.delete('inventory', {'id': item_id})
- item = self.inv.get_by_id(self.env, item_id)
- self.assertIsNone(item)
+ # Assert that router has been added to db
+ router_insertions = [call_args for call_args
+ in self.inv.set.call_args_list
+ if call_args[0][0]['type'] == 'vservice']
+ self.assertTrue(router_insertions)
+ self.assertTrue(subnet_add_mock.add_ports_folder.called)
+ self.assertTrue(subnet_add_mock.add_port_document.called)
+ self.assertTrue(port_add_mock.add_vnics_folder.called)
+ self.assertTrue(port_add_mock.add_vnic_document.called)
+ find_links_mock.add_links\
+ .assert_called_with(search={"parent_id": self.router_id})
+ self.assertTrue(scanner_mock.scan_cliques.called)
- # delete vnics document
- self.inv.delete('inventory', {'parent_id': self.router_id+'-vnics'})
- item = self.inv.get_by_field(self.env, 'vnic', 'parent_id', self.router_id+'-vnics', get_single=True)
- self.assertIsNone(item)
def setUp(self):
super().setUp()
self.values = EVENT_PAYLOAD_ROUTER_DELETE
- self.set_item_for_deletion(object_type="router", document=ROUTER_DOCUMENT)
def test_handle_router_delete(self):
- self.handle_delete(handler=EventRouterDelete())
+ self.handle_delete(handler=EventRouterDelete(),
+ db_object=ROUTER_DOCUMENT)
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
from discover.events.event_router_update import EventRouterUpdate
-from discover.fetchers.api.api_fetch_port import ApiFetchPort
-from discover.fetchers.cli.cli_fetch_host_vservice import CliFetchHostVservice
from test.event_based_scan.test_data.event_payload_router_update import EVENT_PAYLOAD_ROUTER_UPDATE, ROUTER_DOCUMENT, \
- EVENT_PAYLOAD_ROUTER_SET_GATEWAY, EVENT_PAYLOAD_ROUTER_DEL_GATEWAY, ROUTER_VSERVICE, PORTS, NETWORK_DOC, HOST_DOC
+ EVENT_PAYLOAD_ROUTER_SET_GATEWAY, EVENT_PAYLOAD_ROUTER_DEL_GATEWAY, ROUTER_VSERVICE, PORT, NETWORK_DOC, HOST_DOC
from test.event_based_scan.test_event import TestEvent
from utils.util import encode_router_id
class TestRouterUpdate(TestEvent):
- def test_handle_router_update(self):
- for values in [EVENT_PAYLOAD_ROUTER_UPDATE, EVENT_PAYLOAD_ROUTER_SET_GATEWAY, EVENT_PAYLOAD_ROUTER_DEL_GATEWAY]:
- self.values = values
- self.payload = self.values['payload']
- self.router = self.payload['router']
- host_id = self.values['publisher_id'].replace("network.", "", 1)
- self.router_id = encode_router_id(host_id, self.router['id'])
- self.item_ids.append(self.router_id)
-
- # add document for testing
- self.set_item(ROUTER_DOCUMENT)
- self.set_item(PORTS)
- self.set_item(NETWORK_DOC)
- self.set_item(HOST_DOC)
-
- # mock the router document.
- original_get_vservice = CliFetchHostVservice.get_vservice
- CliFetchHostVservice.get_vservice = MagicMock(return_value=ROUTER_VSERVICE)
- self.gw_port_id = ROUTER_DOCUMENT['gw_port_id']
-
- # mock
- original_get_port = ApiFetchPort.get
- ApiFetchPort.get = MagicMock(return_value=[PORTS])
-
- handler = EventRouterUpdate()
- handler.handle(self.env, self.values)
-
- # reset the methods back
- CliFetchHostVservice.get_vservice = original_get_vservice
- ApiFetchPort.get = original_get_port
- # assert router document
- router_doc = self.inv.get_by_id(self.env, self.router_id)
- self.assertIsNotNone(router_doc, msg="router_doc not found.")
- self.assertEqual(self.router['name'], router_doc['name'])
- self.assertEqual(self.router['admin_state_up'], router_doc['admin_state_up'])
-
- if self.router['external_gateway_info'] is None:
- self.assertEqual(router_doc['gw_port_id'], None)
- self.assertEqual(router_doc['network'], [])
- else:
- self.assertIn(self.router['external_gateway_info']['network_id'], router_doc['network'])
+
+ def get_by_id(self, env, object_id):
+ if object_id == self.router_id:
+ return ROUTER_DOCUMENT
+ elif object_id == self.gw_port_id:
+ return PORT
+ elif object_id == self.host_id:
+ return HOST_DOC
+ else:
+ return None
+
+ @patch("discover.events.event_router_update.Scanner")
+ def _do_test(self,
+ values,
+ scanner_class_mock):
+ self.values = values
+ self.payload = self.values['payload']
+ self.router = self.payload['router']
+ self.host_id = self.values['publisher_id'].replace("network.", "", 1)
+ self.router_id = encode_router_id(self.host_id, self.router['id'])
+ self.gw_port_id = ROUTER_DOCUMENT['gw_port_id']
+
+ scanner_mock = scanner_class_mock.return_value
+
+ self.inv.get_by_id.side_effect = self.get_by_id
+
+ res = EventRouterUpdate().handle(self.env, self.values)
+
+ self.assertTrue(res.result)
+ self.assertTrue(scanner_mock.scan_cliques.called)
+
+ @patch("discover.events.event_router_update.EventPortDelete")
+ def test_handle_router_update(self,
+ event_port_delete_class_mock):
+ event_port_delete_mock = event_port_delete_class_mock.return_value
+ self._do_test(EVENT_PAYLOAD_ROUTER_UPDATE)
+ event_port_delete_mock.delete_port\
+ .assert_called_with(self.env, self.gw_port_id)
+
+ @patch("discover.events.event_router_update.FindLinksForVserviceVnics")
+ @patch("discover.events.event_router_update.EventRouterAdd")
+ @patch("discover.events.event_router_update.CliFetchHostVservice")
+ def test_handle_router_set_gateway(self,
+ cli_fetch_vservice_class_mock,
+ event_router_add_class_mock,
+ find_links_class_mock):
+ cli_fetch_vservice_mock = cli_fetch_vservice_class_mock.return_value
+ cli_fetch_vservice_mock.get_vservice.return_value = ROUTER_VSERVICE
+ event_router_add_mock = event_router_add_class_mock.return_value
+ find_links_mock = find_links_class_mock.return_value
+ self._do_test(EVENT_PAYLOAD_ROUTER_SET_GATEWAY)
+ cli_fetch_vservice_mock.get_vservice.assert_called_with(self.host_id,
+ self.router_id)
+ self.assertTrue(event_router_add_mock.add_children_documents.called)
+ self.assertTrue(find_links_mock.add_links.called)
+
+ @patch("discover.events.event_router_update.EventPortDelete")
+ def test_handle_router_delete_gateway(self,
+ event_port_delete_class_mock):
+ event_port_delete_mock = event_port_delete_class_mock.return_value
+ self._do_test(EVENT_PAYLOAD_ROUTER_DEL_GATEWAY)
+ event_port_delete_mock.delete_port \
+ .assert_called_with(self.env, self.gw_port_id)
+
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
from discover.events.event_subnet_add import EventSubnetAdd
from discover.fetchers.api.api_access import ApiAccess
from discover.find_links_for_pnics import FindLinksForPnics
from discover.find_links_for_vservice_vnics import FindLinksForVserviceVnics
-from test.event_based_scan.test_data.event_payload_subnet_add import EVENT_PAYLOAD_SUBNET_ADD,\
- EVENT_PAYLOAD_REGION, NETWORK_DOC
+from test.event_based_scan.test_data.event_payload_subnet_add import \
+ EVENT_PAYLOAD_SUBNET_ADD, \
+ EVENT_PAYLOAD_REGION, NETWORK_DOC, HOST_DOC, PORT_DOC
from test.event_based_scan.test_event import TestEvent
class TestSubnetAdd(TestEvent):
- def test_handle_subnet_add(self):
+ def get_by_id(self, env, object_id):
+ if object_id == self.network_id:
+ return NETWORK_DOC
+ elif object_id == self.host_id:
+ return HOST_DOC
+ else:
+ return None
+
+ def get_port_id(self, network_id):
+ return self.port_id if network_id == self.network_id else None
+
+ def get_port_docs(self, port_id):
+ return [PORT_DOC] if port_id == self.port_id else []
+
+ @patch("discover.events.event_subnet_add.Scanner")
+ @patch("discover.events.event_subnet_add.FindLinksForVserviceVnics")
+ @patch("discover.events.event_subnet_add.FindLinksForPnics")
+ @patch("discover.events.event_subnet_add.EventPortAdd")
+ @patch("discover.events.event_subnet_add.DbFetchPort")
+ @patch("discover.events.event_subnet_add.ApiFetchPort")
+ def test_handle_subnet_add(self,
+ api_fetch_port_class_mock,
+ db_fetch_port_class_mock,
+ event_port_add_class_mock,
+ find_links_for_pnics_class_mock,
+ find_links_for_vnics_class_mock,
+ scanner_class_mock):
self.values = EVENT_PAYLOAD_SUBNET_ADD
self.payload = self.values['payload']
self.subnet = self.payload['subnet']
self.subnet_id = self.subnet['id']
self.network_id = self.subnet['network_id']
- self.item_ids.append(self.network_id)
+ self.host_id = self.values["publisher_id"].replace("network.", "", 1)
+ self.port_id = PORT_DOC['id']
- network_document = self.inv.get_by_id(self.env, self.network_id)
- if network_document:
- # check subnet in network first.
- self.assertNotIn(self.subnet['cidr'], network_document['cidrs'])
- else:
- self.log.info("network document is not found, add it first.")
- self.set_item(NETWORK_DOC)
- # check network document
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIsNotNone(network_document)
+ db_fetch_port_mock = db_fetch_port_class_mock.return_value
+ db_fetch_port_mock.get_id.side_effect = self.get_port_id
+
+ api_fetch_port_mock = api_fetch_port_class_mock.return_value
+ api_fetch_port_mock.get.side_effect = self.get_port_docs
+
+ event_port_add_mock = event_port_add_class_mock.return_value
+ find_links_for_pnics_mock = find_links_for_pnics_class_mock.return_value
+ find_links_for_vnics_mock = find_links_for_vnics_class_mock.return_value
+ scanner_mock = scanner_class_mock.return_value
+
+ self.inv.get_by_id.side_effect = self.get_by_id
- # check region data.
if not ApiAccess.regions:
ApiAccess.regions = EVENT_PAYLOAD_REGION
- # Mock function instead of get children data. They should be test in their unit test.
- # add subnet document for updating network
- handler = EventSubnetAdd()
- handler.add_children_documents = MagicMock()
-
- original_add_pnic_links = FindLinksForPnics.add_links
- FindLinksForPnics.add_links = MagicMock()
+ res = EventSubnetAdd().handle(self.env, self.values)
- original_add_vservice_links = FindLinksForVserviceVnics.add_links
- FindLinksForVserviceVnics.add_links = MagicMock()
+ if ApiAccess.regions == EVENT_PAYLOAD_REGION:
+ ApiAccess.regions = None
- handler.handle(self.env, self.values)
+ self.assertTrue(res.result)
- # reset the methods back
- FindLinksForPnics.add_links = original_add_pnic_links
- FindLinksForVserviceVnics.add_links = original_add_vservice_links
+ # Assert that subnet has been added to network
+ set_call = [call[0][0] for call in self.inv.set.call_args_list
+ if self.payload['subnet']['id']
+ in call[0][0].get('subnet_ids', [])]
+ self.assertTrue(set_call)
- # check network document
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIn(self.subnet['cidr'], network_document['cidrs'])
- self.assertIn(self.subnet['name'], network_document['subnets'])
+ self.assertTrue(event_port_add_mock.add_network_services_folder.called)
+ self.assertTrue(event_port_add_mock.add_dhcp_document.called)
+ self.assertTrue(event_port_add_mock.add_vnics_folder.called)
+ self.assertTrue(event_port_add_mock.add_vnic_document.called)
+ self.assertTrue(find_links_for_pnics_mock.add_links.called)
+ self.assertTrue(find_links_for_vnics_mock.add_links.called)
+ self.assertTrue(scanner_mock.scan_cliques.called)
- #tearDown method has been implemented in class testEvent.
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
+from discover.events.event_base import EventResult
+from discover.events.event_delete_base import EventDeleteBase
from discover.events.event_subnet_delete import EventSubnetDelete
from test.event_based_scan.test_event import TestEvent
-from test.event_based_scan.test_data.event_payload_subnet_delete import EVENT_PAYLOAD_SUBNET_DELETE, \
- EVENT_PAYLOAD_NETWORK
-
+from test.event_based_scan.test_data.event_payload_subnet_delete import \
+ EVENT_PAYLOAD_SUBNET_DELETE, \
+ NETWORK_DOC, VNIC_DOC
+from unittest.mock import patch
class TestSubnetDelete(TestEvent):
- def test_handle_subnet_delete(self):
+ def get_by_field(self, environment, item_type, field_name, field_value,
+ get_single=False):
+ if item_type == "network":
+ return NETWORK_DOC
+ elif item_type == "vnic":
+ return VNIC_DOC
+ else:
+ return None
+
+ @patch.object(EventDeleteBase, "delete_handler",
+ return_value=EventResult(result=True))
+ def test_handle_subnet_delete(self,
+ delete_handler_mock):
self.values = EVENT_PAYLOAD_SUBNET_DELETE
self.subnet_id = self.values['payload']['subnet_id']
- self.network_doc = EVENT_PAYLOAD_NETWORK
+ self.network_doc = NETWORK_DOC
self.network_id = self.network_doc['id']
- self.item_ids.append(self.network_id)
-
- self.subnet_name = None
- self.cidr = None
-
- for subnet in self.network_doc['subnets'].values():
- if subnet['id'] == self.subnet_id:
- self.subnet_name = subnet['name']
- self.cidr = subnet['cidr']
- break
-
- # add document for subnet deleting test.
- self.set_item(self.network_doc)
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIsNotNone(network_document, "add network document failed")
-
- # delete subnet
- EventSubnetDelete().handle(self.env, self.values)
-
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertNotIn(self.subnet_id, network_document['subnet_ids'])
- self.assertNotIn(self.cidr, network_document['cidrs'])
- self.assertNotIn(self.subnet_name, network_document['subnets'])
-
- # assert children documents
- vservice_dhcp_id = 'qdhcp-' + network_document['id']
- dhcp_doc = self.inv.get_by_id(self.env, vservice_dhcp_id)
- self.assertIsNone(dhcp_doc)
-
- vnic_parent_id = vservice_dhcp_id + '-vnics'
- vnic = self.inv.get_by_field(self.env, 'vnic', 'parent_id', vnic_parent_id, get_single=True)
- self.assertIsNone(vnic)
+ self.vnic_id = VNIC_DOC['id']
+ self.vnic_folder_id = 'qdhcp-{}'.format(self.network_id)
+
+ self.inv.get_by_field.side_effect = self.get_by_field
+
+ res = EventSubnetDelete().handle(self.env, self.values)
+
+ self.assertTrue(res.result)
+ delete_handler_mock.assert_called_with(self.env,
+ self.vnic_folder_id, "vservice")
+
+ updated_network = [call[0][0] for call in self.inv.set.call_args_list
+ if call[0][0]['type'] == 'network']
+ self.assertTrue(updated_network)
+ self.assertTrue(self.subnet_id
+ not in updated_network[0].get('subnet_ids'))
+ self.assertTrue(self.inv.delete.called)
+
from discover.fetchers.api.api_access import ApiAccess
from test.event_based_scan.test_data.event_payload_subnet_add import \
EVENT_PAYLOAD_REGION
-from test.event_based_scan.test_data.event_payload_subnet_update import EVENT_PAYLOAD_SUBNET_UPDATE, NETWORK_DOC
+from test.event_based_scan.test_data.event_payload_subnet_update import \
+ EVENT_PAYLOAD_SUBNET_UPDATE, NETWORK_DOC, HOST_DOC
from test.event_based_scan.test_event import TestEvent
class TestSubnetUpdate(TestEvent):
+ def get_by_id(self, env, object_id):
+ if object_id == self.network_id:
+ return NETWORK_DOC
+ elif object_id == self.host_id:
+ return HOST_DOC
+ else:
+ return None
+
def test_handle_subnet_add(self):
self.values = EVENT_PAYLOAD_SUBNET_UPDATE
self.payload = self.values['payload']
self.subnet = self.payload['subnet']
self.subnet_id = self.subnet['id']
self.network_id = self.subnet['network_id']
- self.item_ids.append(self.network_id)
-
- #add network document for subnet.
- self.set_item(NETWORK_DOC)
+ self.host_id = self.values["publisher_id"].replace("network.", "", 1)
+ old_subnet_name = list(NETWORK_DOC['subnets'].keys())[0]
+ new_subnet_name = self.subnet['name']
- # check network document
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIsNotNone(network_document)
+ self.inv.get_by_id.side_effect = self.get_by_id
- # check region data.
if not ApiAccess.regions:
ApiAccess.regions = EVENT_PAYLOAD_REGION
- handler = EventSubnetUpdate()
- handler.handle(self.env, self.values)
+ res = EventSubnetUpdate().handle(self.env, self.values)
+
+ self.assertTrue(res.result)
+ updated_network = [call[0][0] for call in self.inv.set.call_args_list
+ if call[0][0]['type'] == 'network']
+ self.assertTrue(updated_network)
+ self.assertFalse(updated_network[0]['subnets'].get(old_subnet_name))
+ self.assertTrue(updated_network[0]['subnets'].get(new_subnet_name))
+
+ if ApiAccess.regions == EVENT_PAYLOAD_REGION:
+ ApiAccess.regions = None
- # check network document
- network_document = self.inv.get_by_id(self.env, self.network_id)
- self.assertIn(self.subnet['name'], network_document['subnets'])
- self.assertEqual(self.subnet['gateway_ip'], network_document['subnets'][self.subnet['name']]['gateway_ip'])
+ # TODO: write tests for "enable_dhcp" change handling
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
-# local config info for test.
+import copy
+from discover.fetchers.api.api_access import ApiAccess
-MONGODB_CONFIG = 'your-mongo-config-path-here'
-ENV_CONFIG = 'your-env-name-here'
+class TestRegions:
+ def __init__(self, test_regions):
+ super().__init__()
+ self.original_regions = copy.deepcopy(ApiAccess.regions)
+ self.test_regions = test_regions
-COLLECTION_CONFIG = 'your-inventory-collection-name-here'
+ def __enter__(self):
+ ApiAccess.regions = self.test_regions
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ ApiAccess.regions = self.original_regions
\ No newline at end of file
Code Review / calipso.git / commitdiff