X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=src%2Femuvim%2Ftest%2Funittests%2Ftest_openstack.py;h=787d398db76d7125dcc1ae48cf5be4fe81415a78;hb=refs%2Fchanges%2F28%2F7228%2F2;hp=a9273fb47c764cdbf09b2b18c404144dbda58d64;hpb=72f09885db3935e77901c3dee132cb176b927c7b;p=osm%2Fvim-emu.git diff --git a/src/emuvim/test/unittests/test_openstack.py b/src/emuvim/test/unittests/test_openstack.py index a9273fb..787d398 100755 --- a/src/emuvim/test/unittests/test_openstack.py +++ b/src/emuvim/test/unittests/test_openstack.py @@ -568,6 +568,22 @@ class testRestApi(ApiBaseOpenStack): "ports"][0]["id"], listPortsId1) print(" ") + print('->>>>>>> test Neutron List Ports By Device ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + server_url = "http://0.0.0.0:18774/v2.1/id_bla/servers/firewall1:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest" + server_response = requests.get(server_url, headers=headers) + firewall1_server = json.loads(server_response.content)["server"] + device_id = firewall1_server["id"] + url = "http://0.0.0.0:19696/v2.0/ports?device_id=%s" % device_id + list_ports_by_device_id_response = requests.get(url, headers=headers) + self.assertEqual(list_ports_by_device_id_response.status_code, 200) + list_ports_by_device_id_ports = json.loads(list_ports_by_device_id_response.content)["ports"] + + self.assertTrue(any(list_ports_by_device_id_ports), "Expected at least one port for device") + for port in list_ports_by_device_id_ports: + self.assertTrue(port["name"].startswith("firewall1:"), "Expected all ports to belong to firewall1") + print(" ") + print('->>>>>>> test Neutron List Non-Existing Ports ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/ports?id=non-existing-port-id" @@ -610,18 +626,26 @@ class testRestApi(ApiBaseOpenStack): url, data=createportdata, headers=headers) self.assertEqual(createportresponse.status_code, 201) print(createportresponse.content) - self.assertEqual(json.loads(createportresponse.content)[ - "port"]["name"], "new_port") + createport = json.loads(createportresponse.content)["port"] + self.assertEqual(createport["name"], "new_port") print(" ") print('->>>>>>> test Neutron Create Port With Existing Name ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/ports" - createportwithexistingnamedata = '{"port": {"name": "new_port", "network_id": "%s"} }' % ( - json.loads(createnetworkresponse.content)["network"]["id"]) - createportwithexistingnameresponse = requests.post( + network_id = json.loads(createnetworkresponse.content)["network"]["id"] + createportwithexistingnamedata = '{"port": {"name": "duplicate_port_name", "network_id": "%s"} }' % network_id + createportwithexistingnameresponse1 = requests.post( + url, data=createportwithexistingnamedata, headers=headers) + createportwithexistingnameresponse2 = requests.post( url, data=createportwithexistingnamedata, headers=headers) - self.assertEqual(createportwithexistingnameresponse.status_code, 500) + createportwithexistingname1 = json.loads(createportwithexistingnameresponse1.content)["port"] + createportwithexistingname2 = json.loads(createportwithexistingnameresponse2.content)["port"] + self.assertEqual(createportwithexistingnameresponse1.status_code, 201) + self.assertEqual(createportwithexistingnameresponse2.status_code, 201) + self.assertEqual(createportwithexistingname1["name"], "duplicate_port_name") + self.assertEqual(createportwithexistingname2["name"], "duplicate_port_name") + self.assertNotEqual(createportwithexistingname1["id"], createportwithexistingname2["id"], "Duplicate port should have different id") print(" ") print('->>>>>>> test Neutron Create Port Without Name ->>>>>>>>>>>>>>>') @@ -852,7 +876,7 @@ class testRestApi(ApiBaseOpenStack): headers = {'Content-type': 'application/json'} - print('->>>>>>> Create ports p1 - p4 ->>>>>>>>>>>>>>>') + print('->>>>>>> Create ports p1 - p6 ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') # Get network id network_resp = requests.get( @@ -860,37 +884,64 @@ class testRestApi(ApiBaseOpenStack): self.assertEqual(network_resp.status_code, 200) network_id = json.loads(network_resp.content)["networks"][0]["id"] - url = "http://0.0.0.0:19696/v2.0/ports" - port_request = '{"port": {"name": "%s", "network_id": "%s"}}' - p1_resp = requests.post(url, data=port_request % - ("p1", network_id), headers=headers) - self.assertEqual(p1_resp.status_code, 201) - p2_resp = requests.post(url, data=port_request % - ("p2", network_id), headers=headers) - self.assertEqual(p2_resp.status_code, 201) - p3_resp = requests.post(url, data=port_request % - ("p3", network_id), headers=headers) - self.assertEqual(p3_resp.status_code, 201) - p4_resp = requests.post(url, data=port_request % - ("p4", network_id), headers=headers) - self.assertEqual(p4_resp.status_code, 201) - - p1_id = json.loads(p1_resp.content)["port"]["id"] - p2_id = json.loads(p2_resp.content)["port"]["id"] - p3_id = json.loads(p3_resp.content)["port"]["id"] - p4_id = json.loads(p4_resp.content)["port"]["id"] + port_responses = map(lambda name: requests.post("http://0.0.0.0:19696/v2.0/ports", + data='{"port": {"name": "%s", "network_id": "%s"}}' % + (name, network_id), + headers=headers), + ["p1", "p2", "p3", "p4", "p5", "p6"]) + + for port in port_responses: + self.assertEqual(port.status_code, 201) + + port_ids = map(lambda response: json.loads(response.content)["port"]["id"], port_responses) + + listflavorsresponse = requests.get("http://0.0.0.0:18774/v2.1/id_bla/flavors", headers=headers) + self.assertEqual(listflavorsresponse.status_code, 200) + flavors = json.loads(listflavorsresponse.content)["flavors"] + m1_tiny_flavor = filter(lambda flavor: flavor["name"] == "m1.tiny", flavors)[0] + + listimagesdetailsresponse = requests.get("http://0.0.0.0:18774/v2.1/id_bla/images/detail", headers=headers) + self.assertEqual(listimagesdetailsresponse.status_code, 200) + images = json.loads(listimagesdetailsresponse.content)["images"] + ubuntu_image = filter(lambda image: image["name"] == "ubuntu:trusty", images)[0] + + server_url = "http://0.0.0.0:18774/v2.1/id_bla/servers" + server_template = \ + '{"server": {' \ + '"name": "%s",' \ + '"networks": [{"port": "%s"}, {"port": "%s"}],' \ + '"flavorRef": "%s",' \ + '"imageRef": "%s"' \ + '}}' + server_responses = map(lambda spec: ( + requests.post(server_url, + data=server_template % ( + spec["name"], + spec["ingress"], + spec["egress"], + m1_tiny_flavor["id"], + ubuntu_image["id"] + ), + headers=headers) + ), [ + {"name": "s1", "ingress": "p1", "egress": "p2"}, + {"name": "s2", "ingress": "p3", "egress": "p4"}, + {"name": "s3", "ingress": "p5", "egress": "p6"}, + ]) + for response in server_responses: + self.assertEqual(response.status_code, 200) print('->>>>>>> test Neutron SFC Port Pair Create ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs" pp1_resp = requests.post(url, data='{"port_pair": {"name": "pp1", "ingress": "%s", "egress": "%s"}}' % ( - p1_id, p2_id), headers=headers) + port_ids[0], port_ids[1]), headers=headers) self.assertEqual(pp1_resp.status_code, 201) pp2_resp = requests.post(url, data='{"port_pair": {"name": "pp2", "ingress": "%s", "egress": "%s"}}' % ( - p3_id, p4_id), headers=headers) + port_ids[2], port_ids[3]), headers=headers) self.assertEqual(pp2_resp.status_code, 201) pp3_resp = requests.post(url, data='{"port_pair": {"name": "pp3", "ingress": "%s", "egress": "%s"}}' % ( - p3_id, p4_id), headers=headers) + port_ids[4], port_ids[5]), headers=headers) self.assertEqual(pp3_resp.status_code, 201) pp1_id = json.loads(pp1_resp.content)["port_pair"]["id"] @@ -917,9 +968,18 @@ class testRestApi(ApiBaseOpenStack): url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs" pp_list_resp = requests.get(url, headers=headers) self.assertEqual(pp_list_resp.status_code, 200) + pp_list = json.loads(pp_list_resp.content)["port_pairs"] # only pp1 and pp2 should be left - self.assertEqual( - len(json.loads(pp_list_resp.content)["port_pairs"]), 2) + self.assertEqual(len(pp_list), 2) + + print('->>>>>>> test Neutron SFC Port Pair List filtered by id ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs?id=%s" % pp_list[0]["id"] + pp_list_filtered_by_id_resp = requests.get(url, headers=headers) + pp_list_filtered_by_id = json.loads(pp_list_filtered_by_id_resp.content)["port_pairs"] + self.assertEqual(pp_list_filtered_by_id_resp.status_code, 200) + self.assertEqual(len(pp_list_filtered_by_id), 1) + self.assertEqual(pp_list_filtered_by_id[0], pp_list[0]) print('->>>>>>> test Neutron SFC Port Pair Show ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') @@ -982,10 +1042,10 @@ class testRestApi(ApiBaseOpenStack): print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers" fc1_resp = requests.post( - url, data='{"flow_classifier": {"name": "fc1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) + url, data='{"flow_classifier": {"name": "fc1", "logical_source_port": "p1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) self.assertEqual(fc1_resp.status_code, 201) fc2_resp = requests.post( - url, data='{"flow_classifier": {"name": "fc2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) + url, data='{"flow_classifier": {"name": "fc2", "logical_source_port": "p2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) self.assertEqual(fc2_resp.status_code, 201) fc1_id = json.loads(fc1_resp.content)["flow_classifier"]["id"]