"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"
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 ->>>>>>>>>>>>>>>')
headers = {'Content-type': 'application/json'}
- print('->>>>>>> Create ports p1 - p4 ->>>>>>>>>>>>>>>')
+ print('->>>>>>> Create ports p1 - p6 ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
# Get network id
network_resp = requests.get(
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"]
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"]
projects / osm / vim-emu.git / blobdiff