diff options
Diffstat (limited to 'app/test/event_based_scan/test_subnet_add.py')
| -rw-r--r-- | app/test/event_based_scan/test_subnet_add.py | 95 |
1 files changed, 61 insertions, 34 deletions
diff --git a/app/test/event_based_scan/test_subnet_add.py b/app/test/event_based_scan/test_subnet_add.py index a8794ef..ce934f3 100644 --- a/app/test/event_based_scan/test_subnet_add.py +++ b/app/test/event_based_scan/test_subnet_add.py @@ -7,62 +7,89 @@ # 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. |