← Back to branch summary

~markwash/network-service/test-cleanup

« back to all changes in this revision

Viewing changes to tests/unit/test_service.py

  • Committer: rajarammallya
  • Date: 2011-05-13 12:24:17 UTC
  • Revision ID: rajarammallya@gmail.com-20110513122417-e73n2iktbrvbhx6h
Rajaram/Santhosh|Renamed test_ip_blocks_controller to test_service

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unit/test_service.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2011 OpenStack LLC.
 
4
# All Rights Reserved.
 
5
#
 
6
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
7
#    not use this file except in compliance with the License. You may obtain
 
8
#    a copy of the License at
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
12
#    Unless required by applicable law or agreed to in writing, software
 
13
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
14
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
15
#    License for the specific language governing permissions and limitations
 
16
#    under the License.
 
17
 
 
18
import os
 
19
import unittest
 
20
 
 
21
from webtest import TestApp
 
22
from webob import Request, Response
 
23
from webob.exc import HTTPUnprocessableEntity
 
24
 
 
25
from melange.ipam.models import IpBlock
 
26
from melange.ipam.models import IpAddress
 
27
from melange.common import config
 
28
from melange.db import session
 
29
 
 
30
class TestIpBlockController(unittest.TestCase):
 
31
    def setUp(self):
 
32
        conf, melange_app = config.load_paste_app('melange',
 
33
                {"config_file":os.path.abspath("../../etc/melange.conf.test")}, None)
 
34
        self.app = TestApp(melange_app)
 
35
 
 
36
 
 
37
    def test_create(self):
 
38
        response = self.app.post("/ipam/ip_blocks",
 
39
                                 {'network_id':"300",'cidr':"10.1.1.0/2"})
 
40
 
 
41
        self.assertEqual(response.status,"200 OK")
 
42
        saved_block = IpBlock.find_by_network_id("300")
 
43
        self.assertEqual(saved_block.cidr, "10.1.1.0/2")
 
44
        self.assertEqual(response.json, {'id':saved_block.id,'network_id':"300",
 
45
                                         'cidr':"10.1.1.0/2"})
 
46
 
 
47
    def test_cannot_create_duplicate_public_cidr(self):
 
48
        self.app.post("/ipam/ip_blocks",
 
49
                      {"network_id":"300", 'cidr':"192.1.1.1/2", 'type':'public'})
 
50
        response = self.app.post("/ipam/ip_blocks",
 
51
                      {"network_id":"22200", 'cidr':"192.1.1.1/2", 'type':'public'},
 
52
                                 status="*")
 
53
 
 
54
        self.assertEqual(response.status, "400 Bad Request")
 
55
        self.assertTrue("[{'cidr': 'cidr for public ip is not unique'}]" in response.body)
 
56
    
 
57
    def test_create_with_bad_cidr(self):
 
58
        response = self.app.post("/ipam/ip_blocks",
 
59
                                 {'network_id':"300",'cidr':"10..."}, status="*")
 
60
        
 
61
        self.assertEqual(response.status, "400 Bad Request")
 
62
        self.assertTrue("[{'cidr': 'cidr is invalid'}]" in response.body)
 
63
        
 
64
 
 
65
    def test_show(self):
 
66
        block = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/2"})
 
67
        response = self.app.get("/ipam/ip_blocks/%s" %block.id)
 
68
 
 
69
        self.assertEqual(response.status,"200 OK")
 
70
        self.assertEqual(response.json, {'id': block.id,'network_id':"301",
 
71
                                         'cidr':"10.1.1.0/2"})
 
72
 
 
73
class TestIpAddressController(unittest.TestCase):
 
74
    def setUp(self):
 
75
        conf, melange_app = config.load_paste_app('melange',
 
76
                {"config_file":os.path.abspath("../../etc/melange.conf.test")}, None)
 
77
        self.app = TestApp(melange_app)
 
78
 
 
79
    def test_create(self):
 
80
        block = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
81
        response = self.app.post("/ipam/ip_blocks/%s/ip_addresses" % block.id)
 
82
        
 
83
        self.assertEqual(response.status,"200 OK")
 
84
        allocated_address = IpAddress.find_all_by_ip_block(block.id).first()
 
85
        self.assertEqual(allocated_address.address, "10.1.1.0")
 
86
        self.assertEqual(response.json, {'id':allocated_address.id,
 
87
                                         'address':allocated_address.address,
 
88
                                         'port_id':allocated_address.port_id})
 
89
 
 
90
    def test_create_when_no_more_addresses(self):
 
91
        block = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/32"})
 
92
        block.allocate_ip()
 
93
 
 
94
        response = self.app.post("/ipam/ip_blocks/%s/ip_addresses" % block.id,
 
95
                                 status="*")
 
96
        self.assertEqual(response.status,"422 Unprocessable Entity")
 
97
        self.assertTrue("ip block is full" in response.body)        
 
98
 
 
99
    def test_create_with_port(self):
 
100
        block = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
101
        response = self.app.post("/ipam/ip_blocks/%s/ip_addresses" % block.id,
 
102
                                 {"port_id":"1111"})
 
103
 
 
104
        allocated_address = IpAddress.find_all_by_ip_block(block.id).first()
 
105
        self.assertEqual(allocated_address.port_id, "1111")
 
106
 
 
107
    def test_show(self):
 
108
        block_1 = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
109
        block_2 = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
110
        ip = block_1.allocate_ip(port_id="3333")
 
111
        block_2.allocate_ip(port_id="9999")
 
112
 
 
113
        response = self.app.get("/ipam/ip_blocks/%s/ip_addresses/%s" %
 
114
                                (block_1.id, ip.address))
 
115
 
 
116
        self.assertEqual(response.status,"200 OK")
 
117
        self.assertEqual(response.json, {'id': ip.id,
 
118
                                         'address':ip.address,
 
119
                                         'port_id':"3333"})
 
120
 
 
121
    def test_delete_ip(self):
 
122
        block_1 = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
123
        block_2 = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
124
        ip = block_1.allocate_ip()
 
125
        block_2.allocate_ip()
 
126
        
 
127
        response = self.app.delete("/ipam/ip_blocks/%s/ip_addresses/%s" %
 
128
                                (block_1.id, ip.address))
 
129
 
 
130
        self.assertEqual(response.status, "200 OK")
 
131
        self.assertEqual(IpAddress.find(ip.id), None)
 
132
 
 
133
    def test_index(self):
 
134
        block = IpBlock.create({'network_id':"301",'cidr':"10.1.1.0/28"})
 
135
        address_1 = block.allocate_ip()
 
136
        address_2 = block.allocate_ip()
 
137
 
 
138
        response = self.app.get("/ipam/ip_blocks/%s/ip_addresses" % block.id)
 
139
 
 
140
        ip_addresses = response.json["ip_addresses"]
 
141
        self.assertEqual(response.status, "200 OK")
 
142
        self.assertEqual(len(ip_addresses), 2)
 
143
        self.assertEqual(ip_addresses[0]['address'], address_1.address)
 
144
        self.assertEqual(ip_addresses[1]['address'], address_2.address)