/
test_subnet_ip_count.py
166 lines (136 loc) · 5.89 KB
/
test_subnet_ip_count.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
#
# Copyright (c) 2013,2014 Juniper Networks, Inc. All rights reserved.
#
import gevent
import os
import sys
import socket
import errno
import uuid
import logging
import coverage
import testtools
from testtools.matchers import Equals, MismatchError, Not, Contains
from testtools import content, content_type, ExpectedException
import unittest
import re
import json
import copy
import inspect
import pycassa
import kombu
import requests
from vnc_api.vnc_api import *
import vnc_api.gen.vnc_api_test_gen
from vnc_api.gen.resource_test import *
import cfgm_common
from cfgm_common import vnc_cgitb
vnc_cgitb.enable(format='text')
sys.path.append('../common/tests')
from test_utils import *
import test_common
import test_case
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class TestSubnet(test_case.ApiServerTestCase):
def test_subnet_ip_count(self):
# Object needs to be cleanup after the test to get same results for
# repetative test. Until objects are not cleanedup, change subnet before
# testing it again.
domain_name = 'my-domain'
proj_name = 'my-proj'
subnet1 = '192.168.1.0'
prefix1 = 30
subnet2 = '10.10.1.0'
prefix2 = 29
vn_name = 'my-fe'
domain = Domain(domain_name)
self._vnc_lib.domain_create(domain)
print 'Created domain'
project = Project(proj_name, domain)
self._vnc_lib.project_create(project)
print 'Created Project'
ipam = NetworkIpam('default-network-ipam', project, IpamType("dhcp"))
self._vnc_lib.network_ipam_create(ipam)
print 'Created network ipam'
ipam = self._vnc_lib.network_ipam_read(fq_name=[domain_name, proj_name,
'default-network-ipam'])
print 'Read network ipam'
ipam_sn_1 = IpamSubnetType(subnet=SubnetType(subnet1, prefix1))
ipam_sn_2 = IpamSubnetType(subnet=SubnetType(subnet2, prefix2))
vn = VirtualNetwork(vn_name, project)
vn.add_network_ipam(ipam, VnSubnetsType([ipam_sn_1, ipam_sn_2]))
self._vnc_lib.virtual_network_create(vn)
print 'Created Virtual Network object ', vn.uuid
print 'Read no of instance ip for each subnet'
print '["192.168.1.0/30", "10.10.1.0/29"]'
subnet_list = ["192.168.1.0/30", "10.10.1.0/29"]
result = self._vnc_lib.virtual_network_subnet_ip_count(vn, subnet_list)
print 'Expected output: {"ip_count_list": [0, 0]}'
print 'Actual output:', result
net_obj = self._vnc_lib.virtual_network_read(id=vn.uuid)
ip_obj1 = InstanceIp(name=str(uuid.uuid4()))
ip_obj1.uuid = ip_obj1.name
print 'Created Instance IP object 1 ', ip_obj1.uuid
vm_inst_obj1 = VirtualMachine(str(uuid.uuid4()))
vm_inst_obj1.uuid = vm_inst_obj1.name
self._vnc_lib.virtual_machine_create(vm_inst_obj1)
id_perms = IdPermsType(enable=True)
port_obj1 = VirtualMachineInterface(
str(uuid.uuid4()), vm_inst_obj1, id_perms=id_perms)
port_obj1.uuid = port_obj1.name
port_obj1.set_virtual_network(vn)
ip_obj1.set_virtual_machine_interface(port_obj1)
ip_obj1.set_virtual_network(net_obj)
port_id1 = self._vnc_lib.virtual_machine_interface_create(port_obj1)
print 'Allocating an IP address for first VM'
ip_id1 = self._vnc_lib.instance_ip_create(ip_obj1)
ip_obj1 = self._vnc_lib.instance_ip_read(id=ip_id1)
ip_addr1 = ip_obj1.get_instance_ip_address()
print ' got IP Address for first instance', ip_addr1
net_obj = self._vnc_lib.virtual_network_read(id=vn.uuid)
result = self._vnc_lib.virtual_network_subnet_ip_count(vn, subnet_list)
print 'Expected output: {"ip_count_list": [1, 0]}'
print 'Actual output:', result
net_obj = self._vnc_lib.virtual_network_read(id=vn.uuid)
ip_obj2 = InstanceIp(name=str(uuid.uuid4()))
ip_obj2.uuid = ip_obj2.name
print 'Created Instance IP object 2', ip_obj2.uuid
vm_inst_obj2 = VirtualMachine(str(uuid.uuid4()))
vm_inst_obj2.uuid = vm_inst_obj2.name
self._vnc_lib.virtual_machine_create(vm_inst_obj2)
id_perms = IdPermsType(enable=True)
port_obj2 = VirtualMachineInterface(
str(uuid.uuid4()), vm_inst_obj2, id_perms=id_perms)
port_obj2.uuid = port_obj2.name
port_obj2.set_virtual_network(vn)
ip_obj2.set_virtual_machine_interface(port_obj2)
ip_obj2.set_virtual_network(net_obj)
port_id2 = self._vnc_lib.virtual_machine_interface_create(port_obj2)
print 'Allocating an IP address for Second VM'
ip_id2 = self._vnc_lib.instance_ip_create(ip_obj2)
ip_obj2 = self._vnc_lib.instance_ip_read(id=ip_id2)
ip_addr2 = ip_obj2.get_instance_ip_address()
print ' got IP Address for Second instance', ip_addr2
net_obj = self._vnc_lib.virtual_network_read(id=vn.uuid)
result = self._vnc_lib.virtual_network_subnet_ip_count(vn, subnet_list)
print 'Expected output: {"ip_count_list": [1, 1]}'
print 'Actual output:', result
print result
# cleanup
print 'Cleaning up'
self._vnc_lib.instance_ip_delete(id=ip_id1)
self._vnc_lib.instance_ip_delete(id=ip_id2)
self._vnc_lib.virtual_machine_interface_delete(id=port_obj1.uuid)
self._vnc_lib.virtual_machine_interface_delete(id=port_obj2.uuid)
self._vnc_lib.virtual_machine_delete(id=vm_inst_obj1.uuid)
self._vnc_lib.virtual_machine_delete(id=vm_inst_obj2.uuid)
self._vnc_lib.virtual_network_delete(id=vn.uuid)
self._vnc_lib.network_ipam_delete(id=ipam.uuid)
self._vnc_lib.project_delete(id=project.uuid)
self._vnc_lib.domain_delete(id=domain.uuid)
if __name__ == '__main__':
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
logger.addHandler(ch)
unittest.main()