← Back to branch summary

~salvatore-orlando/neutron/quantum-api

« back to all changes in this revision

Viewing changes to tests/functional/test_service.py

  • Committer: Salvatore Orlando
  • Date: 2011-06-24 13:52:17 UTC
  • mfrom: (6.1.14 quantum-trunk)
  • Revision ID: salvatore.orlando@eu.citrix.com-20110624135217-h6uz1zu3fxxpf3wt
Merge trunk
Resolving conflicts

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/functional/test_service.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2011 Citrix Systems
 
4
# Copyright 2011 Nicira Networks
 
5
# All Rights Reserved.
 
6
#
 
7
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
8
#    not use this file except in compliance with the License. You may obtain
 
9
#    a copy of the License at
 
10
#
 
11
#         http://www.apache.org/licenses/LICENSE-2.0
 
12
#
 
13
#    Unless required by applicable law or agreed to in writing, software
 
14
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
15
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
16
#    License for the specific language governing permissions and limitations
 
17
#    under the License.
 
18
 
 
19
import gettext
 
20
import simplejson
 
21
import sys
 
22
import unittest
 
23
 
 
24
gettext.install('quantum', unicode=1)
 
25
 
 
26
from miniclient import MiniClient
 
27
from quantum.common.wsgi import Serializer
 
28
 
 
29
HOST = '127.0.0.1'
 
30
PORT = 9696
 
31
USE_SSL = False
 
32
 
 
33
TENANT_ID = 'totore'
 
34
FORMAT = "json"
 
35
 
 
36
test_network1_data = \
 
37
    {'network': {'network-name': 'test1'}}
 
38
test_network2_data = \
 
39
    {'network': {'network-name': 'test2'}}
 
40
 
 
41
 
 
42
def print_response(res):
 
43
    content = res.read()
 
44
    print "Status: %s" % res.status
 
45
    print "Content: %s" % content
 
46
    return content
 
47
 
 
48
 
 
49
class QuantumTest(unittest.TestCase):
 
50
    def setUp(self):
 
51
        self.client = MiniClient(HOST, PORT, USE_SSL)
 
52
 
 
53
    def create_network(self, data):
 
54
        content_type = "application/" + FORMAT
 
55
        body = Serializer().serialize(data, content_type)
 
56
        res = self.client.do_request(TENANT_ID, 'POST', "/networks." + FORMAT,
 
57
          body=body)
 
58
        self.assertEqual(res.status, 200, "bad response: %s" % res.read())
 
59
 
 
60
    def test_listNetworks(self):
 
61
        self.create_network(test_network1_data)
 
62
        self.create_network(test_network2_data)
 
63
        res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT)
 
64
        self.assertEqual(res.status, 200, "bad response: %s" % res.read())
 
65
 
 
66
    def test_createNetwork(self):
 
67
        self.create_network(test_network1_data)
 
68
 
 
69
    def test_createPort(self):
 
70
        self.create_network(test_network1_data)
 
71
        res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT)
 
72
        resdict = simplejson.loads(res.read())
 
73
        for n in resdict["networks"]:
 
74
            net_id = n["id"]
 
75
 
 
76
            # Step 1 - List Ports for network (should not find any)
 
77
            res = self.client.do_request(TENANT_ID, 'GET',
 
78
              "/networks/%s/ports.%s" % (net_id, FORMAT))
 
79
            self.assertEqual(res.status, 200, "Bad response: %s" % res.read())
 
80
            output = res.read()
 
81
            self.assertTrue(len(output) == 0,
 
82
              "Found unexpected ports: %s" % output)
 
83
 
 
84
            # Step 2 - Create Port for network
 
85
            res = self.client.do_request(TENANT_ID, 'POST',
 
86
              "/networks/%s/ports.%s" % (net_id, FORMAT))
 
87
            self.assertEqual(res.status, 200, "Bad response: %s" % output)
 
88
 
 
89
            # Step 3 - List Ports for network (again); should find one
 
90
            res = self.client.do_request(TENANT_ID, 'GET',
 
91
              "/networks/%s/ports.%s" % (net_id, FORMAT))
 
92
            output = res.read()
 
93
            self.assertEqual(res.status, 200, "Bad response: %s" % output)
 
94
            resdict = simplejson.loads(output)
 
95
            ids = []
 
96
            for p in resdict["ports"]:
 
97
                ids.append(p["id"])
 
98
            self.assertTrue(len(ids) == 1,
 
99
              "Didn't find expected # of ports (1): %s" % ids)
 
100
 
 
101
    def test_renameNetwork(self):
 
102
        self.create_network(test_network1_data)
 
103
        res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT)
 
104
        resdict = simplejson.loads(res.read())
 
105
        net_id = resdict["networks"][0]["id"]
 
106
 
 
107
        data = test_network1_data.copy()
 
108
        data['network']['network-name'] = 'test_renamed'
 
109
        content_type = "application/" + FORMAT
 
110
        body = Serializer().serialize(data, content_type)
 
111
        res = self.client.do_request(TENANT_ID, 'PUT',
 
112
          "/networks/%s.%s" % (net_id, FORMAT), body=body)
 
113
        resdict = simplejson.loads(res.read())
 
114
        self.assertTrue(resdict["networks"]["network"]["id"] == net_id,
 
115
          "Network_rename: renamed network has a different uuid")
 
116
        self.assertTrue(
 
117
            resdict["networks"]["network"]["name"] == "test_renamed",
 
118
            "Network rename didn't take effect")
 
119
 
 
120
    def delete_networks(self):
 
121
        # Remove all the networks created on the tenant
 
122
        res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT)
 
123
        resdict = simplejson.loads(res.read())
 
124
        for n in resdict["networks"]:
 
125
            net_id = n["id"]
 
126
            res = self.client.do_request(TENANT_ID, 'DELETE',
 
127
              "/networks/" + net_id + "." + FORMAT)
 
128
            self.assertEqual(res.status, 202)
 
129
 
 
130
    def tearDown(self):
 
131
        self.delete_networks()
 
132
 
 
133
# Standard boilerplate to call the main() function.
 
134
if __name__ == '__main__':
 
135
    suite = unittest.TestLoader().loadTestsFromTestCase(QuantumTest)
 
136
    unittest.TextTestRunner(verbosity=2).run(suite)