blob: a9acd8e4e429286b6cd95a24c39e629a12253b4d [file] [log] [blame]
Gilles Depatie2e683692019-02-22 16:06:52 -05001#
2# Copyright 2018 the original author or authors.
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17"""
18vOLT-HA DHCP Test Case module
19"""
20
21import os
22import testCaseUtils
23import logging
24import subprocess
25
26
27class DHCP(object):
28 """
29 This class implements voltha DHCP test case
30 """
31
32 CHECK_IP_FILENAME = 'voltha_check_ip.log'
33 DE_ASSIGN_IP_FILENAME = 'voltha_de-assign_ip.log'
34 ASSIGN_DHCP_IP_FILENAME = 'voltha_assign_dhcp_ip.log'
35 CHECK_ASSIGNED_IP_FILENAME = 'voltha_check_assigned_dhcp_ip.log'
36
37 def __init__(self):
38 self.dirs = dict()
39 self.dirs['log'] = None
40 self.dirs['root'] = None
41 self.dirs['voltha'] = None
42
Gilles Depatieed99efe2019-03-12 16:12:26 -040043 self.__rgName = testCaseUtils.discover_rg_pod_name()
Gilles Depatie2e683692019-02-22 16:06:52 -050044 self.__fields = None
45 self.__deviceId = None
46 self.__portNumber = None
47
48 def h_set_log_dirs(self, root_dir, voltha_dir, log_dir):
49 testCaseUtils.config_dirs(self, log_dir, root_dir, voltha_dir)
50
Gilles Depatie2e683692019-02-22 16:06:52 -050051 def discover_authorized_users(self):
52 testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
53 'voltha_onos_users.log', 'aaa-users')
54
55 def retrieve_authorized_users_device_id_and_port_number(self):
56 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'AUTHORIZED', 'voltha_onos_users.log')
57 assert statusLines, 'No Users Authorized'
58 self.__fields = testCaseUtils.parse_fields(statusLines, ',')
59 deviceField = self.__fields[2].strip()
60 deviceStr, equal, deviceId = deviceField.partition('=')
61 self.__deviceId = deviceId
62 portField = self.__fields[4].strip()
63 portNumStr, equal, portNum = portField.partition('=')
64 self.__portNumber = portNum
65
66 def add_subscriber_access(self):
67 testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
68 'voltha_add_subscriber_access.log', 'volt-add-subscriber-access %s %s'
69 % (self.__deviceId, self.__portNumber))
70
71 def should_now_have_two_dhcp_flows(self):
72 testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
Gilles Depatieea423712019-04-12 16:39:12 -040073 'voltha_onos_flows.log', 'flows -s')
Gilles Depatie2e683692019-02-22 16:06:52 -050074 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'IP_PROTO:17', 'voltha_onos_flows.log')
75 assert statusLines, 'No DHCP Detection flows'
76 lines = statusLines.splitlines()
Gilles Depatieea423712019-04-12 16:39:12 -040077 assert len(lines) == 2, 'Expected 2 DHCP Detection Flows but result was %s' % len(lines)
Gilles Depatie2e683692019-02-22 16:06:52 -050078 for line in lines:
79 self.__fields = testCaseUtils.parse_fields(line, ',')
Gilles Depatieea423712019-04-12 16:39:12 -040080 inPortStr = self.__fields[5].strip()
Gilles Depatie2e683692019-02-22 16:06:52 -050081 selector, delimiter, inPort = inPortStr.partition('=[')
82 assert (inPort == 'IN_PORT:2' or inPort == 'IN_PORT:128'), 'DHCP detection flows not associated with expected ports'
83
84 def add_dhcp_server_configuration_data_in_onos(self):
85 logging.info('Adding DHCP Configuration Data to Onos NetCfg')
86 logging.debug('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
87 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/dhcp_json'
88 % testCaseUtils.get_dir(self, 'voltha'))
89 os.system('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
90 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/dhcp_json'
91 % testCaseUtils.get_dir(self, 'voltha'))
92
93 def activate_dhcp_server_in_onos(self):
94 logging.info('Activating DHCP server on Onos')
95 testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
96 'voltha_dhcp_server_activate.log', 'app activate dhcp')
97 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Activated', 'voltha_dhcp_server_activate.log')
98 assert statusLines, 'DHCP server failed to be Activated'
99
100 def deactivate_dhcp_server_in_onos(self):
101 logging.info('Deactivating DHCP server on Onos')
102 testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
103 'voltha_dhcp_server_deactivate.log', 'app deactivate dhcp')
104 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Deactivated', 'voltha_dhcp_server_deactivate.log')
105 assert statusLines, 'DHCP server failed to be Deactivated'
106
107 def query_for_default_ip_on_rg(self):
108 logging.info('De-assigning default IP on RG')
109 process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.CHECK_IP_FILENAME), 'w')
110 ifconfigCheck1 = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-n', 'voltha', self.__rgName, '--', 'bash', '-c',
111 'ifconfig'],
112 stdout=subprocess.PIPE,
113 stderr=subprocess.PIPE)
114 ifconfigCheck2 = subprocess.Popen(['grep', '-e', 'eth0', '-A1'], stdin=ifconfigCheck1.stdout,
115 stdout=process_output,
116 stderr=process_output)
117
118 ifconfigCheck1.wait()
119 ifconfigCheck1.stdout.close()
120 ifconfigCheck2.wait()
121
122 process_output.close()
123
124 testCaseUtils.print_log_file(self, self.CHECK_IP_FILENAME)
125
126 def de_assign_default_ip_on_rg(self):
127
128 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'inet', self.CHECK_IP_FILENAME)
129 if statusLines:
130 process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.DE_ASSIGN_IP_FILENAME), 'w')
131 os.system('/usr/bin/kubectl exec -n voltha %s -- bash -c "ifconfig eth0 0.0.0.0"' % self.__rgName)
132 ifconfigDeassign1 = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-n', 'voltha', self.__rgName, '--', 'bash', '-c',
133 'ifconfig'],
134 stdout=subprocess.PIPE,
135 stderr=subprocess.PIPE)
136
137 ifconfigDeassign2 = subprocess.Popen(['grep', '-e', 'eth0', '-A1'], stdin=ifconfigDeassign1.stdout,
138 stdout=process_output,
139 stderr=process_output)
140 ifconfigDeassign1.wait()
141 ifconfigDeassign1.stdout.close()
142 ifconfigDeassign2.wait()
143
144 process_output.close()
145
146 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'inet', self.DE_ASSIGN_IP_FILENAME)
147 assert not statusLines, 'IP addr not de-assigned'
148
149 else:
150 logging.info('No default IP addr assigned to eth0')
151
152 def assign_dhcp_ip_addr_to_rg(self):
153 logging.info('Assigning IP addr on RG using DHCP')
154 process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.ASSIGN_DHCP_IP_FILENAME), 'w')
155 dhcpAssignIp1 = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-it', '-n', 'voltha', self.__rgName, '--',
156 'dhclient', '-v', 'eth0'],
157 stdout=process_output,
158 stderr=process_output)
159
160 dhcpAssignIp1.wait()
161 process_output.close()
162
163 testCaseUtils.print_log_file(self, self.ASSIGN_DHCP_IP_FILENAME)
164
Gilles Depatieed99efe2019-03-12 16:12:26 -0400165 procPidDhclient1 = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-n', 'voltha', self.__rgName, '--', 'ps', '-ef'],
166 stdout=subprocess.PIPE,
167 stderr=subprocess.PIPE)
168 procPidDhclient2 = subprocess.Popen(['grep', '-e', 'dhclient'], stdin=procPidDhclient1.stdout,
169 stdout=subprocess.PIPE,
170 stderr=subprocess.PIPE)
171 procPidDhclient3 = subprocess.Popen(['awk', "{print $2}"], stdin=procPidDhclient2.stdout,
172 stdout=subprocess.PIPE,
173 stderr=subprocess.PIPE)
174
175 procPidDhclient1.stdout.close()
176 procPidDhclient2.stdout.close()
177
178 out, err = procPidDhclient3.communicate()
179 dhclientPid = out.strip()
180 if dhclientPid:
181 procKillDhclient = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-n', 'voltha', self.__rgName, '--', 'kill', dhclientPid],
182 stdout=subprocess.PIPE,
183 stderr=subprocess.PIPE)
184
185 out, err = procKillDhclient.communicate()
186 assert not err, 'Killing dhclient returned %s' % err
187
Gilles Depatie2e683692019-02-22 16:06:52 -0500188 def should_have_dhcp_assigned_ip(self):
189 process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.CHECK_ASSIGNED_IP_FILENAME), 'w')
190 ifConfigCheck1 = subprocess.Popen(['/usr/bin/kubectl', 'exec', '-n', 'voltha', self.__rgName, '--', 'bash', '-c',
191 'ifconfig'],
192 stdout=subprocess.PIPE,
193 stderr=subprocess.PIPE)
194
195 ifConfigCheck2 = subprocess.Popen(['grep', '-e', 'eth0', '-A1'], stdin=ifConfigCheck1.stdout,
196 stdout=process_output,
197 stderr=process_output)
198 ifConfigCheck1.wait()
199 ifConfigCheck1.stdout.close()
200 ifConfigCheck2.wait()
201
202 process_output.close()
203
204 testCaseUtils.print_log_file(self, self.CHECK_ASSIGNED_IP_FILENAME)
205
206 statusLines = testCaseUtils.get_fields_from_grep_command(self, 'inet', self.CHECK_ASSIGNED_IP_FILENAME)
207 assert statusLines, 'DHCP IP addr not assigned'
208
209
210def set_firewall_rules():
211 logging.info('Setting Firewall rules for DHCP test')
212 os.system('sudo iptables -P FORWARD ACCEPT')
213
214
215def run_test(root_dir, voltha_dir, log_dir):
216
217 set_firewall_rules()
218 dhcp = DHCP()
219 dhcp.h_set_log_dirs(root_dir, voltha_dir, log_dir)
Gilles Depatie2e683692019-02-22 16:06:52 -0500220 dhcp.discover_authorized_users()
221 dhcp.retrieve_authorized_users_device_id_and_port_number()
222 dhcp.add_subscriber_access()
223 dhcp.should_now_have_two_dhcp_flows()
224 dhcp.deactivate_dhcp_server_in_onos()
225 dhcp.add_dhcp_server_configuration_data_in_onos()
226 dhcp.activate_dhcp_server_in_onos()
227 dhcp.query_for_default_ip_on_rg()
228 dhcp.de_assign_default_ip_on_rg()
229 dhcp.assign_dhcp_ip_addr_to_rg()
230 dhcp.should_have_dhcp_assigned_ip()