How to use make_headers method in tempest

Best Python code snippet using tempest_python

controller.py

Source:controller.py Github

copy

Full Screen

...62 raise NameError("Invalid service code for request.")63 print(uri)64 params = params65 if method == "GET":66 resp = requests.get(uri, headers=self.make_headers(), params=params, verify=False, timeout=100)67 elif method == "POST":68 print(uri, payload, params)69 resp = requests.post(uri, params=params, data=payload, headers=self.make_headers(), verify=False,70 timeout=100)71 elif method == "PUT":72 resp = requests.put(uri, params=params, data=payload, verify=False, timeout=100)73 elif method == "DELETE":74 resp = requests.delete(uri, headers=self.make_headers(), params=params, verify=False, timeout=100)75 self.check_response(method, resp, self.make_headers(), payload, uri)76 return resp77 def login(self):78 uri = self.build_uri_sec("oauth2")79 # self.session.mount(uri, HTTPAdapter(max_retries=15))80 payload = json.dumps({"userId": self.username, "password": self.password})81 print("Sending Command: " + "\n" +82 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +83 "URI: " + str(uri) + "\n" +84 "Data: " + str(payload) + "\n" +85 "Headers: " + str(self.make_headers()))86 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +87 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +88 "URI: " + str(uri) + "\n" +89 "Data: " + str(payload) + "\n" +90 "Headers: " + str(self.make_headers()))91 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)92 self.check_response("POST", resp, "", payload, uri)93 token = resp94 self.access_token = token.json()['access_token']95 resp.close()96 if resp.status_code != 200:97 pytest.exit(str(resp.json()))98 # self.session.headers.update({'Authorization': self.access_token})99 return token100 def get_gw_endpoint(self):101 uri = self.build_uri_sec("systemEndpoints")102 print("Sending Command: " + "\n" +103 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +104 "URI: " + str(uri) + "\n" +105 "Headers: " + str(self.make_headers()))106 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +107 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +108 "URI: " + str(uri) + "\n" +109 "Headers: " + str(self.make_headers()))110 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)111 self.check_response("GET", resp, self.make_headers(), "", uri)112 services = resp.json()113 gw_host = ""114 fms_host = ""115 for service in services['endpoints']:116 if service['type'] == "owgw":117 gw_host = urlparse(service["uri"])118 if service['type'] == "owfms":119 fms_host = urlparse(service["uri"])120 if service['type'] == "owprov":121 prov_host = urlparse(service["uri"])122 return gw_host, fms_host, prov_host123 def logout(self):124 uri = self.build_uri_sec('oauth2/%s' % self.access_token)125 print("Sending Command: " + "\n" +126 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +127 "URI: " + str(uri) + "\n" +128 "Headers: " + str(self.make_headers()))129 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +130 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +131 "URI: " + str(uri) + "\n" +132 "Headers: " + str(self.make_headers()))133 resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100)134 self.check_response("DELETE", resp, self.make_headers(), "", uri)135 r = resp136 resp.close()137 return r138 def make_headers(self):139 headers = {'Authorization': 'Bearer %s' % self.access_token,140 "Connection": "keep-alive",141 "Content-Type": "application/json",142 "Keep-Alive": "timeout=10, max=1000"143 }144 return headers145 def check_response(self, cmd, response, headers, data_str, url):146 try:147 print("Command Response: " + "\n" +148 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +149 "Response Code: " + str(response.status_code) + "\n" +150 "Response Body: " + str(response.json()))151 allure.attach(name="Command Response: ", body="Command Response: " + "\n" +152 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +153 "Response Code: " + str(response.status_code) + "\n" +154 "Response Body: " + str(response.json()))155 except:156 pass157 return True158class Controller(ConfigureController):159 def __init__(self, controller_data=None):160 super().__init__(controller_data)161 def get_devices(self):162 uri = self.build_uri("devices")163 print("Sending Command: " + "\n" +164 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +165 "URI: " + str(uri) + "\n" +166 "Headers: " + str(self.make_headers()))167 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +168 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +169 "URI: " + str(uri) + "\n" +170 "Headers: " + str(self.make_headers()))171 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)172 self.check_response("GET", resp, self.make_headers(), "", uri)173 return resp174 def get_device_by_serial_number(self, serial_number):175 uri = self.build_uri("device/" + serial_number)176 print("Sending Command: " + "\n" +177 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +178 "URI: " + str(uri) + "\n" +179 "Headers: " + str(self.make_headers()))180 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +181 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +182 "URI: " + str(uri) + "\n" +183 "Headers: " + str(self.make_headers()))184 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)185 self.check_response("GET", resp, self.make_headers(), "", uri)186 return resp187 def get_sdk_version(self):188 uri = self.build_uri("system?command=info")189 print("Sending Command: " + "\n" +190 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +191 "URI: " + str(uri) + "\n" +192 "Headers: " + str(self.make_headers()))193 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +194 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +195 "URI: " + str(uri) + "\n" +196 "Headers: " + str(self.make_headers()))197 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)198 self.check_response("GET", resp, self.make_headers(), "", uri)199 version = resp.json()200 return version['version']201 def get_system_gw(self):202 uri = self.build_uri("system?command=info")203 print("Sending Command: " + "\n" +204 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +205 "URI: " + str(uri) + "\n" +206 "Headers: " + str(self.make_headers()))207 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +208 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +209 "URI: " + str(uri) + "\n" +210 "Headers: " + str(self.make_headers()))211 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)212 self.check_response("GET", resp, self.make_headers(), "", uri)213 return resp214 def get_system_fms(self):215 uri = self.build_url_fms("system?command=info")216 print("Sending Command: " + "\n" +217 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +218 "URI: " + str(uri) + "\n" +219 "Headers: " + str(self.make_headers()))220 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +221 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +222 "URI: " + str(uri) + "\n" +223 "Headers: " + str(self.make_headers()))224 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)225 self.check_response("GET", resp, self.make_headers(), "", uri)226 return resp227 def get_system_prov(self):228 uri = self.build_url_prov("system?command=info")229 print("Sending Command: " + "\n" +230 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +231 "URI: " + str(uri) + "\n" +232 "Headers: " + str(self.make_headers()))233 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +234 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +235 "URI: " + str(uri) + "\n" +236 "Headers: " + str(self.make_headers()))237 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)238 self.check_response("GET", resp, self.make_headers(), "", uri)239 return resp240 def get_device_uuid(self, serial_number):241 device_info = self.get_device_by_serial_number(serial_number=serial_number)242 device_info = device_info.json()243 return device_info["UUID"]244 def add_device_to_gw(self, serial_number, payload):245 uri = self.build_uri("device/" + serial_number)246 payload = json.dumps(payload)247 print("Sending Command: " + "\n" +248 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +249 "URI: " + str(uri) + "\n" +250 "Data: " + str(payload) + "\n" +251 "Headers: " + str(self.make_headers()))252 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +253 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +254 "URI: " + str(uri) + "\n" +255 "Data: " + str(payload) + "\n" +256 "Headers: " + str(self.make_headers()))257 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)258 self.check_response("POST", resp, self.make_headers(), payload, uri)259 return resp260 def delete_device_from_gw(self, device_name):261 uri = self.build_uri("device/" + device_name)262 print("Sending Command: " + "\n" +263 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +264 "URI: " + str(uri) + "\n" +265 "Headers: " + str(self.make_headers()))266 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +267 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +268 "URI: " + str(uri) + "\n" +269 "Headers: " + str(self.make_headers()))270 resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100)271 self.check_response("DELETE", resp, self.make_headers(), "", uri)272 return resp273 def get_commands(self):274 uri = self.build_uri("commands")275 print("Sending Command: " + "\n" +276 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +277 "URI: " + str(uri) + "\n" +278 "Headers: " + str(self.make_headers()))279 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +280 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +281 "URI: " + str(uri) + "\n" +282 "Headers: " + str(self.make_headers()))283 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)284 self.check_response("GET", resp, self.make_headers(), "", uri)285 return resp286 def get_device_logs(self, serial_number):287 uri = self.build_uri("device/" + serial_number + "/logs")288 print("Sending Command: " + "\n" +289 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +290 "URI: " + str(uri) + "\n" +291 "Headers: " + str(self.make_headers()))292 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +293 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +294 "URI: " + str(uri) + "\n" +295 "Headers: " + str(self.make_headers()))296 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)297 self.check_response("GET", resp, self.make_headers(), "", uri)298 return resp299 def get_device_health_checks(self, serial_number):300 uri = self.build_uri("device/" + serial_number + "/healthchecks")301 print("Sending Command: " + "\n" +302 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +303 "URI: " + str(uri) + "\n" +304 "Headers: " + str(self.make_headers()))305 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +306 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +307 "URI: " + str(uri) + "\n" +308 "Headers: " + str(self.make_headers()))309 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)310 self.check_response("GET", resp, self.make_headers(), "", uri)311 return resp312 def get_device_capabilities(self, serial_number):313 uri = self.build_uri("device/" + serial_number + "/capabilities")314 print("Sending Command: " + "\n" +315 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +316 "URI: " + str(uri) + "\n" +317 "Headers: " + str(self.make_headers()))318 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +319 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +320 "URI: " + str(uri) + "\n" +321 "Headers: " + str(self.make_headers()))322 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)323 self.check_response("GET", resp, self.make_headers(), "", uri)324 return resp325 def get_device_statistics(self, serial_number):326 uri = self.build_uri("device/" + serial_number + "/statistics")327 print("Sending Command: " + "\n" +328 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +329 "URI: " + str(uri) + "\n" +330 "Headers: " + str(self.make_headers()))331 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +332 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +333 "URI: " + str(uri) + "\n" +334 "Headers: " + str(self.make_headers()))335 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)336 self.check_response("GET", resp, self.make_headers(), "", uri)337 return resp338 def get_device_status(self, serial_number):339 uri = self.build_uri("device/" + serial_number + "/status")340 print("Sending Command: " + "\n" +341 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +342 "URI: " + str(uri) + "\n" +343 "Headers: " + str(self.make_headers()))344 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +345 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +346 "URI: " + str(uri) + "\n" +347 "Headers: " + str(self.make_headers()))348 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)349 self.check_response("GET", resp, self.make_headers(), "", uri)350 return resp351 def ap_reboot(self, serial_number, payload):352 uri = self.build_uri("device/" + serial_number + "/reboot")353 payload = json.dumps(payload)354 print("Sending Command: " + "\n" +355 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +356 "URI: " + str(uri) + "\n" +357 "Data: " + str(payload) + "\n" +358 "Headers: " + str(self.make_headers()))359 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +360 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +361 "URI: " + str(uri) + "\n" +362 "Data: " + str(payload) + "\n" +363 "Headers: " + str(self.make_headers()))364 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)365 self.check_response("POST", resp, self.make_headers(), payload, uri)366 return resp367 def ap_factory_reset(self, serial_number, payload):368 uri = self.build_uri("device/" + serial_number + "/factory")369 payload = json.dumps(payload)370 print("Sending Command: " + "\n" +371 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +372 "URI: " + str(uri) + "\n" +373 "Data: " + str(payload) + "\n" +374 "Headers: " + str(self.make_headers()))375 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +376 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +377 "URI: " + str(uri) + "\n" +378 "Data: " + str(payload) + "\n" +379 "Headers: " + str(self.make_headers()))380 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)381 self.check_response("POST", resp, self.make_headers(), payload, uri)382 return resp383 def ping_device(self, serial_number, payload):384 uri = self.build_uri("device/" + serial_number + "/ping")385 payload = json.dumps(payload)386 print("Sending Command: " + "\n" +387 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +388 "URI: " + str(uri) + "\n" +389 "Data: " + str(payload) + "\n" +390 "Headers: " + str(self.make_headers()))391 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +392 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +393 "URI: " + str(uri) + "\n" +394 "Data: " + str(payload) + "\n" +395 "Headers: " + str(self.make_headers()))396 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)397 self.check_response("POST", resp, self.make_headers(), payload, uri)398 return resp399 def led_blink_device(self, serial_number, payload):400 uri = self.build_uri("device/" + serial_number + "/leds")401 payload = json.dumps(payload)402 print("Sending Command: " + "\n" +403 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +404 "URI: " + str(uri) + "\n" +405 "Data: " + str(payload) + "\n" +406 "Headers: " + str(self.make_headers()))407 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +408 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +409 "URI: " + str(uri) + "\n" +410 "Data: " + str(payload) + "\n" +411 "Headers: " + str(self.make_headers()))412 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)413 self.check_response("POST", resp, self.make_headers(), payload, uri)414 return resp415 def trace_device(self, serial_number, payload):416 uri = self.build_uri("device/" + serial_number + "/trace")417 payload = json.dumps(payload)418 print("Sending Command: " + "\n" +419 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +420 "URI: " + str(uri) + "\n" +421 "Data: " + str(payload) + "\n" +422 "Headers: " + str(self.make_headers()))423 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +424 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +425 "URI: " + str(uri) + "\n" +426 "Data: " + str(payload) + "\n" +427 "Headers: " + str(self.make_headers()))428 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)429 self.check_response("POST", resp, self.make_headers(), payload, uri)430 return resp431 def wifi_scan_device(self, serial_number, payload):432 uri = self.build_uri("device/" + serial_number + "/wifiscan")433 payload = json.dumps(payload)434 print("Sending Command: " + "\n" +435 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +436 "URI: " + str(uri) + "\n" +437 "Data: " + str(payload) + "\n" +438 "Headers: " + str(self.make_headers()))439 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +440 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +441 "URI: " + str(uri) + "\n" +442 "Data: " + str(payload) + "\n" +443 "Headers: " + str(self.make_headers()))444 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)445 self.check_response("POST", resp, self.make_headers(), payload, uri)446 return resp447 def request_specific_msg_from_device(self, serial_number, payload):448 uri = self.build_uri("device/" + serial_number + "/request")449 payload = json.dumps(payload)450 print("Sending Command: " + "\n" +451 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +452 "URI: " + str(uri) + "\n" +453 "Data: " + str(payload) + "\n" +454 "Headers: " + str(self.make_headers()))455 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +456 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +457 "URI: " + str(uri) + "\n" +458 "Data: " + str(payload) + "\n" +459 "Headers: " + str(self.make_headers()))460 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)461 self.check_response("POST", resp, self.make_headers(), payload, uri)462 return resp463 def event_queue(self, serial_number, payload):464 uri = self.build_uri("device/" + serial_number + "/eventqueue")465 payload = json.dumps(payload)466 print("Sending Command: " + "\n" +467 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +468 "URI: " + str(uri) + "\n" +469 "Data: " + str(payload) + "\n" +470 "Headers: " + str(self.make_headers()))471 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +472 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +473 "URI: " + str(uri) + "\n" +474 "Data: " + str(payload) + "\n" +475 "Headers: " + str(self.make_headers()))476 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)477 self.check_response("POST", resp, self.make_headers(), payload, uri)478 return resp479 def telemetry(self, serial_number, payload):480 uri = self.build_uri("device/" + serial_number + "/telemetry")481 payload = json.dumps(payload)482 print("Sending Command: " + "\n" +483 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +484 "URI: " + str(uri) + "\n" +485 "Data: " + str(payload) + "\n" +486 "Headers: " + str(self.make_headers()))487 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +488 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +489 "URI: " + str(uri) + "\n" +490 "Data: " + str(payload) + "\n" +491 "Headers: " + str(self.make_headers()))492 resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)493 self.check_response("POST", resp, self.make_headers(), payload, uri)494 return resp495 def get_rtty_params(self, serial_number):496 uri = self.build_uri("device/" + serial_number + "/rtty")497 print("Sending Command: " + "\n" +498 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +499 "URI: " + str(uri) + "\n" +500 "Headers: " + str(self.make_headers()))501 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +502 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +503 "URI: " + str(uri) + "\n" +504 "Headers: " + str(self.make_headers()))505 resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100)506 self.check_response("GET", resp, self.make_headers(), "", uri)507 return resp508 def edit_device_on_gw(self, serial_number, payload):509 uri = self.build_uri("device/" + serial_number)510 payload = json.dumps(payload)511 print("Sending Command: " + "\n" +512 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +513 "URI: " + str(uri) + "\n" +514 "Data: " + str(payload) + "\n" +515 "Headers: " + str(self.make_headers()))516 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +517 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +518 "URI: " + str(uri) + "\n" +519 "Data: " + str(payload) + "\n" +520 "Headers: " + str(self.make_headers()))521 resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100)522 self.check_response("PUT", resp, self.make_headers(), payload, uri)523 return resp524class FMSUtils:525 def __init__(self, sdk_client=None, controller_data=None):526 if sdk_client is None:527 self.sdk_client = Controller(controller_data=controller_data)528 self.sdk_client = sdk_client529 def upgrade_firmware(self, serial="", url=""):530 payload = "{ \"serialNumber\" : " + "\"" + \531 serial + "\"" + " , \"uri\" : " \532 + "\"" + url \533 + "\"" + ", \"when\" : 0" \534 + " }"535 command = "device/" + serial + "/upgrade"536 print("Sending Command: " + "\n" +537 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +538 "URI: " + str(command) + "\n" +539 "Data: " + str(payload) + "\n" +540 "Headers: " + str(self.sdk_client.make_headers()))541 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +542 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +543 "URI: " + str(command) + "\n" +544 "Data: " + str(payload) + "\n" +545 "Headers: " + str(self.sdk_client.make_headers()))546 response = self.sdk_client.request(service="gw", command="device/" + serial + "/upgrade",547 method="POST", params="serialNumber=" + serial,548 payload="{ \"serialNumber\" : " + "\"" + serial + "\"" +549 " , \"uri\" : " + "\"" + url + "\"" +550 ", \"when\" : 0" + " }")551 def ap_model_lookup(self, model=""):552 devices = self.get_device_set()553 model_name = ""554 for device in devices['deviceTypes']:555 if str(device).__eq__(model):556 model_name = device557 return model_name558 def get_revisions(self):559 response = self.sdk_client.request(service="fms", command="firmwares", method="GET", params="revisionSet=true",560 payload="")561 if response.status_code == 200:562 return response.json()563 else:564 return {}565 def get_latest_fw(self, model=""):566 device_type = self.ap_model_lookup(model=model)567 response = self.sdk_client.request(service="fms", command="firmwares", method="GET",568 params="latestOnly=true&deviceType=" + device_type,569 payload="")570 if response.status_code == 200:571 return response.json()572 else:573 return {}574 def get_device_set(self):575 response = self.sdk_client.request(service="fms", command="firmwares", method="GET", params="deviceSet=true",576 payload="")577 if response.status_code == 200:578 return response.json()579 else:580 return {}581 def get_firmwares(self, limit="10000", model="", latestonly="", branch="", commit_id="", offset="3000"):582 deviceType = self.ap_model_lookup(model=model)583 params = "limit=" + limit + \584 "&deviceType=" + deviceType + \585 "&latestonly=" + latestonly + \586 "offset=" + offset587 command = "firmwares"588 response = self.sdk_client.request(service="fms", command=command, method="GET", params=params, payload="")589 allure.attach(name=command + params,590 body=str(response.status_code) + "\n" + str(response.json()),591 attachment_type=allure.attachment_type.JSON)592 if response.status_code == 200:593 data = response.json()594 newlist = sorted(data['firmwares'], key=itemgetter('created'))595 # for i in newlist:596 # print(i['uri'])597 # print(i['revision'])598 # print(newlist)599 return newlist600 # print(data)601 return "error"602class ProvUtils:603 def __init__(self, sdk_client=None, controller_data=None):604 if sdk_client is None:605 self.sdk_client = Controller(controller_data=controller_data)606 self.sdk_client = sdk_client607 def get_inventory(self):608 uri = self.sdk_client.build_url_prov("inventory")609 print("Sending Command: " + "\n" +610 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +611 "URI: " + str(uri) + "\n" +612 "Headers: " + str(self.sdk_client.make_headers()))613 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +614 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +615 "URI: " + str(uri) + "\n" +616 "Headers: " + str(self.sdk_client.make_headers()))617 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)618 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)619 return resp620 def get_inventory_by_device(self, device_name):621 uri = self.sdk_client.build_url_prov("inventory/" + device_name)622 print("Sending Command: " + "\n" +623 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +624 "URI: " + str(uri) + "\n" +625 "Headers: " + str(self.sdk_client.make_headers()))626 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +627 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +628 "URI: " + str(uri) + "\n" +629 "Headers: " + str(self.sdk_client.make_headers()))630 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)631 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)632 return resp633 def get_system_prov(self):634 uri = self.sdk_client.build_url_prov("system?command=info")635 print("Sending Command: " + "\n" +636 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +637 "URI: " + str(uri) + "\n" +638 "Headers: " + str(self.sdk_client.make_headers()))639 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +640 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +641 "URI: " + str(uri) + "\n" +642 "Headers: " + str(self.sdk_client.make_headers()))643 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)644 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)645 return resp646 def add_device_to_inventory(self, device_name, payload):647 uri = self.sdk_client.build_url_prov("inventory/" + device_name)648 payload = json.dumps(payload)649 print("Sending Command: " + "\n" +650 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +651 "URI: " + str(uri) + "\n" +652 "Data: " + str(payload) + "\n" +653 "Headers: " + str(self.sdk_client.make_headers()))654 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +655 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +656 "URI: " + str(uri) + "\n" +657 "Data: " + str(payload) + "\n" +658 "Headers: " + str(self.sdk_client.make_headers()))659 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)660 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)661 return resp662 def delete_device_from_inventory(self, device_name):663 uri = self.sdk_client.build_url_prov("inventory/" + device_name)664 print("Sending Command: " + "\n" +665 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +666 "URI: " + str(uri) + "\n" +667 "Headers: " + str(self.sdk_client.make_headers()))668 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +669 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +670 "URI: " + str(uri) + "\n" +671 "Headers: " + str(self.sdk_client.make_headers()))672 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)673 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)674 return resp675 def get_entity(self):676 uri = self.sdk_client.build_url_prov("entity")677 print("Sending Command: " + "\n" +678 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +679 "URI: " + str(uri) + "\n" +680 "Headers: " + str(self.sdk_client.make_headers()))681 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +682 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +683 "URI: " + str(uri) + "\n" +684 "Headers: " + str(self.sdk_client.make_headers()))685 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)686 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)687 return resp688 def get_entity_by_id(self, entity_id):689 uri = self.sdk_client.build_url_prov("entity/" + entity_id)690 print("Sending Command: " + "\n" +691 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +692 "URI: " + str(uri) + "\n" +693 "Headers: " + str(self.sdk_client.make_headers()))694 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +695 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +696 "URI: " + str(uri) + "\n" +697 "Headers: " + str(self.sdk_client.make_headers()))698 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)699 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)700 return resp701 def add_entity(self, payload):702 uri = self.sdk_client.build_url_prov("entity/1")703 payload = json.dumps(payload)704 print("Sending Command: " + "\n" +705 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +706 "URI: " + str(uri) + "\n" +707 "Data: " + str(payload) + "\n" +708 "Headers: " + str(self.sdk_client.make_headers()))709 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +710 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +711 "URI: " + str(uri) + "\n" +712 "Data: " + str(payload) + "\n" +713 "Headers: " + str(self.sdk_client.make_headers()))714 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)715 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)716 return resp717 def delete_entity(self, entity_id):718 uri = self.sdk_client.build_url_prov("entity/" + entity_id)719 print("Sending Command: " + "\n" +720 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +721 "URI: " + str(uri) + "\n" +722 "Headers: " + str(self.sdk_client.make_headers()))723 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +724 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +725 "URI: " + str(uri) + "\n" +726 "Headers: " + str(self.sdk_client.make_headers()))727 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)728 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)729 return resp730 def edit_device_from_inventory(self, device_name, payload):731 uri = self.sdk_client.build_url_prov("inventory/" + device_name)732 payload = json.dumps(payload)733 print("Sending Command: " + "\n" +734 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +735 "URI: " + str(uri) + "\n" +736 "Data: " + str(payload) + "\n" +737 "Headers: " + str(self.sdk_client.make_headers()))738 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +739 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +740 "URI: " + str(uri) + "\n" +741 "Data: " + str(payload) + "\n" +742 "Headers: " + str(self.sdk_client.make_headers()))743 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)744 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)745 return resp746 def edit_entity(self, payload, entity_id):747 uri = self.sdk_client.build_url_prov("entity/" + entity_id)748 payload = json.dumps(payload)749 print("Sending Command: " + "\n" +750 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +751 "URI: " + str(uri) + "\n" +752 "Data: " + str(payload) + "\n" +753 "Headers: " + str(self.sdk_client.make_headers()))754 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +755 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +756 "URI: " + str(uri) + "\n" +757 "Data: " + str(payload) + "\n" +758 "Headers: " + str(self.sdk_client.make_headers()))759 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)760 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)761 return resp762 def get_contact(self):763 uri = self.sdk_client.build_url_prov("contact")764 print("Sending Command: " + "\n" +765 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +766 "URI: " + str(uri) + "\n" +767 "Headers: " + str(self.sdk_client.make_headers()))768 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +769 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +770 "URI: " + str(uri) + "\n" +771 "Headers: " + str(self.sdk_client.make_headers()))772 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)773 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)774 return resp775 def get_contact_by_id(self, contact_id):776 uri = self.sdk_client.build_url_prov("contact/" + contact_id)777 print("Sending Command: " + "\n" +778 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +779 "URI: " + str(uri) + "\n" +780 "Headers: " + str(self.sdk_client.make_headers()))781 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +782 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +783 "URI: " + str(uri) + "\n" +784 "Headers: " + str(self.sdk_client.make_headers()))785 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)786 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)787 return resp788 def add_contact(self, payload):789 uri = self.sdk_client.build_url_prov("contact/1")790 payload = json.dumps(payload)791 print("Sending Command: " + "\n" +792 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +793 "URI: " + str(uri) + "\n" +794 "Data: " + str(payload) + "\n" +795 "Headers: " + str(self.sdk_client.make_headers()))796 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +797 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +798 "URI: " + str(uri) + "\n" +799 "Data: " + str(payload) + "\n" +800 "Headers: " + str(self.sdk_client.make_headers()))801 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)802 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)803 return resp804 def delete_contact(self, contact_id):805 uri = self.sdk_client.build_url_prov("contact/" + contact_id)806 print("Sending Command: " + "\n" +807 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +808 "URI: " + str(uri) + "\n" +809 "Headers: " + str(self.sdk_client.make_headers()))810 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +811 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +812 "URI: " + str(uri) + "\n" +813 "Headers: " + str(self.sdk_client.make_headers()))814 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)815 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)816 return resp817 def edit_contact(self, payload, contact_id):818 uri = self.sdk_client.build_url_prov("contact/" + contact_id)819 payload = json.dumps(payload)820 print("Sending Command: " + "\n" +821 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +822 "URI: " + str(uri) + "\n" +823 "Data: " + str(payload) + "\n" +824 "Headers: " + str(self.sdk_client.make_headers()))825 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +826 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +827 "URI: " + str(uri) + "\n" +828 "Data: " + str(payload) + "\n" +829 "Headers: " + str(self.sdk_client.make_headers()))830 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)831 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)832 return resp833 def get_location(self):834 uri = self.sdk_client.build_url_prov("location")835 print("Sending Command: " + "\n" +836 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +837 "URI: " + str(uri) + "\n" +838 "Headers: " + str(self.sdk_client.make_headers()))839 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +840 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +841 "URI: " + str(uri) + "\n" +842 "Headers: " + str(self.sdk_client.make_headers()))843 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)844 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)845 return resp846 def get_location_by_id(self, location_id):847 uri = self.sdk_client.build_url_prov("location/" + location_id)848 print("Sending Command: " + "\n" +849 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +850 "URI: " + str(uri) + "\n" +851 "Headers: " + str(self.sdk_client.make_headers()))852 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +853 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +854 "URI: " + str(uri) + "\n" +855 "Headers: " + str(self.sdk_client.make_headers()))856 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)857 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)858 return resp859 def add_location(self, payload):860 uri = self.sdk_client.build_url_prov("location/1")861 payload = json.dumps(payload)862 print("Sending Command: " + "\n" +863 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +864 "URI: " + str(uri) + "\n" +865 "Data: " + str(payload) + "\n" +866 "Headers: " + str(self.sdk_client.make_headers()))867 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +868 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +869 "URI: " + str(uri) + "\n" +870 "Data: " + str(payload) + "\n" +871 "Headers: " + str(self.sdk_client.make_headers()))872 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)873 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)874 return resp875 def delete_location(self, location_id):876 uri = self.sdk_client.build_url_prov("location/" + location_id)877 print("Sending Command: " + "\n" +878 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +879 "URI: " + str(uri) + "\n" +880 "Headers: " + str(self.sdk_client.make_headers()))881 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +882 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +883 "URI: " + str(uri) + "\n" +884 "Headers: " + str(self.sdk_client.make_headers()))885 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)886 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)887 return resp888 def edit_location(self, payload, location_id):889 uri = self.sdk_client.build_url_prov("location/" + location_id)890 payload = json.dumps(payload)891 print("Sending Command: " + "\n" +892 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +893 "URI: " + str(uri) + "\n" +894 "Data: " + str(payload) + "\n" +895 "Headers: " + str(self.sdk_client.make_headers()))896 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +897 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +898 "URI: " + str(uri) + "\n" +899 "Data: " + str(payload) + "\n" +900 "Headers: " + str(self.sdk_client.make_headers()))901 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)902 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)903 return resp904 def get_venue(self):905 uri = self.sdk_client.build_url_prov("venue")906 print("Sending Command: " + "\n" +907 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +908 "URI: " + str(uri) + "\n" +909 "Headers: " + str(self.sdk_client.make_headers()))910 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +911 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +912 "URI: " + str(uri) + "\n" +913 "Headers: " + str(self.sdk_client.make_headers()))914 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)915 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)916 return resp917 def get_venue_by_id(self, venue_id):918 uri = self.sdk_client.build_url_prov("venue/" + venue_id)919 print("Sending Command: " + "\n" +920 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +921 "URI: " + str(uri) + "\n" +922 "Headers: " + str(self.sdk_client.make_headers()))923 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +924 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +925 "URI: " + str(uri) + "\n" +926 "Headers: " + str(self.sdk_client.make_headers()))927 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)928 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)929 return resp930 def add_venue(self, payload):931 uri = self.sdk_client.build_url_prov("venue/0")932 payload = json.dumps(payload)933 print("Sending Command: " + "\n" +934 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +935 "URI: " + str(uri) + "\n" +936 "Data: " + str(payload) + "\n" +937 "Headers: " + str(self.sdk_client.make_headers()))938 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +939 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +940 "URI: " + str(uri) + "\n" +941 "Data: " + str(payload) + "\n" +942 "Headers: " + str(self.sdk_client.make_headers()))943 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)944 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)945 return resp946 def delete_venue(self, venue_id):947 uri = self.sdk_client.build_url_prov("venue/" + venue_id)948 print("Sending Command: " + "\n" +949 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +950 "URI: " + str(uri) + "\n" +951 "Headers: " + str(self.sdk_client.make_headers()))952 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +953 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +954 "URI: " + str(uri) + "\n" +955 "Headers: " + str(self.sdk_client.make_headers()))956 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)957 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)958 return resp959 def edit_venue(self, payload, venue_id):960 uri = self.sdk_client.build_url_prov("venue/" + venue_id)961 payload = json.dumps(payload)962 print("Sending Command: " + "\n" +963 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +964 "URI: " + str(uri) + "\n" +965 "Data: " + str(payload) + "\n" +966 "Headers: " + str(self.sdk_client.make_headers()))967 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +968 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +969 "URI: " + str(uri) + "\n" +970 "Data: " + str(payload) + "\n" +971 "Headers: " + str(self.sdk_client.make_headers()))972 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)973 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)974 return resp975 def get_map(self):976 uri = self.sdk_client.build_url_prov("map")977 print("Sending Command: " + "\n" +978 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +979 "URI: " + str(uri) + "\n" +980 "Headers: " + str(self.sdk_client.make_headers()))981 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +982 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +983 "URI: " + str(uri) + "\n" +984 "Headers: " + str(self.sdk_client.make_headers()))985 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)986 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)987 return resp988 def get_map_by_id(self, map_id):989 uri = self.sdk_client.build_url_prov("map/" + map_id)990 print("Sending Command: " + "\n" +991 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +992 "URI: " + str(uri) + "\n" +993 "Headers: " + str(self.sdk_client.make_headers()))994 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +995 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +996 "URI: " + str(uri) + "\n" +997 "Headers: " + str(self.sdk_client.make_headers()))998 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)999 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1000 return resp1001 def add_map(self, payload):1002 uri = self.sdk_client.build_url_prov("map/0")1003 payload = json.dumps(payload)1004 print("Sending Command: " + "\n" +1005 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1006 "URI: " + str(uri) + "\n" +1007 "Data: " + str(payload) + "\n" +1008 "Headers: " + str(self.sdk_client.make_headers()))1009 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1010 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1011 "URI: " + str(uri) + "\n" +1012 "Data: " + str(payload) + "\n" +1013 "Headers: " + str(self.sdk_client.make_headers()))1014 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1015 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)1016 return resp1017 def delete_map(self, map_id):1018 uri = self.sdk_client.build_url_prov("map/" + map_id)1019 print("Sending Command: " + "\n" +1020 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1021 "URI: " + str(uri) + "\n" +1022 "Headers: " + str(self.sdk_client.make_headers()))1023 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1024 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1025 "URI: " + str(uri) + "\n" +1026 "Headers: " + str(self.sdk_client.make_headers()))1027 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1028 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)1029 return resp1030 def edit_map(self, payload, map_id):1031 uri = self.sdk_client.build_url_prov("map/" + map_id)1032 payload = json.dumps(payload)1033 print("Sending Command: " + "\n" +1034 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1035 "URI: " + str(uri) + "\n" +1036 "Data: " + str(payload) + "\n" +1037 "Headers: " + str(self.sdk_client.make_headers()))1038 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1039 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1040 "URI: " + str(uri) + "\n" +1041 "Data: " + str(payload) + "\n" +1042 "Headers: " + str(self.sdk_client.make_headers()))1043 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1044 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)1045 return resp1046 def get_operator(self):1047 uri = self.sdk_client.build_url_prov("operator")1048 print("Sending Command: " + "\n" +1049 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1050 "URI: " + str(uri) + "\n" +1051 "Headers: " + str(self.sdk_client.make_headers()))1052 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1053 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1054 "URI: " + str(uri) + "\n" +1055 "Headers: " + str(self.sdk_client.make_headers()))1056 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1057 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1058 return resp1059 def get_operator_by_id(self, operator_id):1060 uri = self.sdk_client.build_url_prov("operator/" + operator_id)1061 print("Sending Command: " + "\n" +1062 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1063 "URI: " + str(uri) + "\n" +1064 "Headers: " + str(self.sdk_client.make_headers()))1065 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1066 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1067 "URI: " + str(uri) + "\n" +1068 "Headers: " + str(self.sdk_client.make_headers()))1069 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1070 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1071 return resp1072 def add_operator(self, payload):1073 uri = self.sdk_client.build_url_prov("operator/1")1074 payload = json.dumps(payload)1075 print("Sending Command: " + "\n" +1076 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1077 "URI: " + str(uri) + "\n" +1078 "Data: " + str(payload) + "\n" +1079 "Headers: " + str(self.sdk_client.make_headers()))1080 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1081 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1082 "URI: " + str(uri) + "\n" +1083 "Data: " + str(payload) + "\n" +1084 "Headers: " + str(self.sdk_client.make_headers()))1085 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1086 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)1087 return resp1088 def delete_operator(self, operator_id):1089 uri = self.sdk_client.build_url_prov("operator/" + operator_id)1090 print("Sending Command: " + "\n" +1091 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1092 "URI: " + str(uri) + "\n" +1093 "Headers: " + str(self.sdk_client.make_headers()))1094 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1095 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1096 "URI: " + str(uri) + "\n" +1097 "Headers: " + str(self.sdk_client.make_headers()))1098 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1099 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)1100 return resp1101 def edit_operator(self, payload, operator_id):1102 uri = self.sdk_client.build_url_prov("operator/" + operator_id)1103 payload = json.dumps(payload)1104 print("Sending Command: " + "\n" +1105 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1106 "URI: " + str(uri) + "\n" +1107 "Data: " + str(payload) + "\n" +1108 "Headers: " + str(self.sdk_client.make_headers()))1109 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1110 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1111 "URI: " + str(uri) + "\n" +1112 "Data: " + str(payload) + "\n" +1113 "Headers: " + str(self.sdk_client.make_headers()))1114 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1115 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)1116 return resp1117 def get_service_class_by_operator_id(self, operator_id):1118 uri = self.sdk_client.build_url_prov("serviceClass?operatorId=" + operator_id)1119 print("Sending Command: " + "\n" +1120 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1121 "URI: " + str(uri) + "\n" +1122 "Headers: " + str(self.sdk_client.make_headers()))1123 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1124 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1125 "URI: " + str(uri) + "\n" +1126 "Headers: " + str(self.sdk_client.make_headers()))1127 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1128 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1129 return resp1130 def get_service_class_by_id(self, service_class_id):1131 uri = self.sdk_client.build_url_prov("serviceClass/" + service_class_id)1132 print("Sending Command: " + "\n" +1133 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1134 "URI: " + str(uri) + "\n" +1135 "Headers: " + str(self.sdk_client.make_headers()))1136 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1137 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1138 "URI: " + str(uri) + "\n" +1139 "Headers: " + str(self.sdk_client.make_headers()))1140 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1141 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1142 return resp1143 def add_service_class(self, payload):1144 uri = self.sdk_client.build_url_prov("serviceClass/1")1145 payload = json.dumps(payload)1146 print("Sending Command: " + "\n" +1147 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1148 "URI: " + str(uri) + "\n" +1149 "Data: " + str(payload) + "\n" +1150 "Headers: " + str(self.sdk_client.make_headers()))1151 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1152 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1153 "URI: " + str(uri) + "\n" +1154 "Data: " + str(payload) + "\n" +1155 "Headers: " + str(self.sdk_client.make_headers()))1156 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1157 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)1158 return resp1159 def delete_service_class(self, service_class_id):1160 uri = self.sdk_client.build_url_prov("serviceClass/" + service_class_id)1161 print("Sending Command: " + "\n" +1162 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1163 "URI: " + str(uri) + "\n" +1164 "Headers: " + str(self.sdk_client.make_headers()))1165 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1166 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1167 "URI: " + str(uri) + "\n" +1168 "Headers: " + str(self.sdk_client.make_headers()))1169 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1170 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)1171 return resp1172 def edit_service_class(self, payload, service_class_id):1173 uri = self.sdk_client.build_url_prov("serviceClass/" + service_class_id)1174 payload = json.dumps(payload)1175 print("Sending Command: " + "\n" +1176 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1177 "URI: " + str(uri) + "\n" +1178 "Data: " + str(payload) + "\n" +1179 "Headers: " + str(self.sdk_client.make_headers()))1180 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1181 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1182 "URI: " + str(uri) + "\n" +1183 "Data: " + str(payload) + "\n" +1184 "Headers: " + str(self.sdk_client.make_headers()))1185 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1186 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)1187 return resp1188 def get_configuration(self):1189 uri = self.sdk_client.build_url_prov("configuration")1190 print("Sending Command: " + "\n" +1191 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1192 "URI: " + str(uri) + "\n" +1193 "Headers: " + str(self.sdk_client.make_headers()))1194 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1195 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1196 "URI: " + str(uri) + "\n" +1197 "Headers: " + str(self.sdk_client.make_headers()))1198 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1199 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1200 return resp1201 def get_configuration_by_id(self, configuration_id):1202 uri = self.sdk_client.build_url_prov("configuration/" + configuration_id)1203 print("Sending Command: " + "\n" +1204 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1205 "URI: " + str(uri) + "\n" +1206 "Headers: " + str(self.sdk_client.make_headers()))1207 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1208 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1209 "URI: " + str(uri) + "\n" +1210 "Headers: " + str(self.sdk_client.make_headers()))1211 resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1212 self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)1213 return resp1214 def add_configuration(self, payload):1215 uri = self.sdk_client.build_url_prov("configuration/1")1216 payload = json.dumps(payload)1217 print("Sending Command: " + "\n" +1218 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1219 "URI: " + str(uri) + "\n" +1220 "Data: " + str(payload) + "\n" +1221 "Headers: " + str(self.sdk_client.make_headers()))1222 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1223 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1224 "URI: " + str(uri) + "\n" +1225 "Data: " + str(payload) + "\n" +1226 "Headers: " + str(self.sdk_client.make_headers()))1227 resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1228 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri)1229 return resp1230 def delete_configuration(self, configuration_id):1231 uri = self.sdk_client.build_url_prov("configuration/" + configuration_id)1232 print("Sending Command: " + "\n" +1233 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1234 "URI: " + str(uri) + "\n" +1235 "Headers: " + str(self.sdk_client.make_headers()))1236 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1237 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1238 "URI: " + str(uri) + "\n" +1239 "Headers: " + str(self.sdk_client.make_headers()))1240 resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1241 self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri)1242 return resp1243 def edit_configuration(self, payload, configuration_id):1244 uri = self.sdk_client.build_url_prov("configuration/" + configuration_id)1245 payload = json.dumps(payload)1246 print("Sending Command: " + "\n" +1247 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1248 "URI: " + str(uri) + "\n" +1249 "Data: " + str(payload) + "\n" +1250 "Headers: " + str(self.sdk_client.make_headers()))1251 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1252 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1253 "URI: " + str(uri) + "\n" +1254 "Data: " + str(payload) + "\n" +1255 "Headers: " + str(self.sdk_client.make_headers()))1256 resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100)1257 self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri)1258 return resp1259class UProfileUtility:1260 def __init__(self, sdk_client=None, controller_data=None):1261 if sdk_client is None:1262 self.sdk_client = Controller(controller_data=controller_data)1263 self.sdk_client = sdk_client1264 self.base_profile_config = {1265 "uuid": 1,1266 "radios": [],1267 "interfaces": [{1268 "name": "WAN",1269 "role": "upstream",1270 "services": ["ssh", "lldp", "dhcp-snooping"],1271 "ethernet": [1272 {1273 "select-ports": [1274 "WAN*"1275 ]1276 }1277 ],1278 "ipv4": {1279 "addressing": "dynamic"1280 }1281 },1282 {1283 "name": "LAN",1284 "role": "downstream",1285 "services": ["ssh", "lldp", "dhcp-snooping"],1286 "ethernet": [1287 {1288 "select-ports": [1289 "LAN*"1290 ]1291 }1292 ],1293 "ipv4": {1294 "addressing": "static",1295 "subnet": "192.168.1.1/16",1296 "dhcp": {1297 "lease-first": 10,1298 "lease-count": 10000,1299 "lease-time": "6h"1300 }1301 },1302 }],1303 "metrics": {1304 "statistics": {1305 "interval": 60,1306 "types": ["ssids", "lldp", "clients"]1307 },1308 "health": {1309 "interval": 1201310 },1311 "wifi-frames": {1312 "filters": ["probe",1313 "auth",1314 "assoc",1315 "disassoc",1316 "deauth",1317 "local-deauth",1318 "inactive-deauth",1319 "key-mismatch",1320 "beacon-report",1321 "radar-detected"]1322 },1323 "dhcp-snooping": {1324 "filters": ["ack", "discover", "offer", "request", "solicit", "reply", "renew"]1325 }1326 },1327 "services": {1328 "lldp": {1329 "describe": "TIP OpenWiFi",1330 "location": "QA"1331 },1332 "ssh": {1333 "port": 221334 }1335 }1336 }1337 self.vlan_section = {1338 "name": "WAN100",1339 "role": "upstream",1340 "vlan": {1341 "id": 1001342 },1343 "ethernet": [1344 {1345 "select-ports": [1346 "WAN*"1347 ]1348 }1349 ],1350 "ipv4": {1351 "addressing": "dynamic"1352 }1353 }1354 self.mode = None1355 def set_mesh_services(self):1356 self.base_profile_config["interfaces"][1]["ipv4"]["subnet"] = "192.168.97.1/24"1357 self.base_profile_config["interfaces"][1]["ipv4"]["dhcp"]["lease-count"] = 1001358 del self.base_profile_config['metrics']['wifi-frames']1359 del self.base_profile_config['metrics']['dhcp-snooping']1360 var = {1361 "filters": ["probe",1362 "auth"]1363 }1364 self.base_profile_config["metrics"]['wifi-frames'] = var1365 del self.base_profile_config['services']1366 var2 = {1367 "lldp": {1368 "describe": "uCentral",1369 "location": "universe"1370 },1371 "ssh": {1372 "port": 221373 }1374 }1375 self.base_profile_config['services'] = var21376 def set_express_wifi(self, open_flow=None):1377 if self.mode == "NAT":1378 self.base_profile_config["interfaces"][1]["services"] = ["ssh", "lldp", "open-flow"]1379 self.base_profile_config["interfaces"][1]["ipv4"]["subnet"] = "192.168.97.1/24"1380 self.base_profile_config["interfaces"][1]["ipv4"]["dhcp"]["lease-count"] = 1001381 self.base_profile_config['services']["open-flow"] = open_flow1382 self.base_profile_config['services']['lldp']['describe'] = "OpenWiFi - expressWiFi"1383 self.base_profile_config['services']['lldp']['location'] = "Hotspot"1384 def set_captive_portal(self):1385 if self.mode == "NAT":1386 max_client = {1387 "max-clients": 321388 }1389 # sourceFile = open('captive_config.py', 'w')1390 self.base_profile_config["interfaces"][1]["name"] = "captive"1391 self.base_profile_config["interfaces"][1]["ipv4"]["subnet"] = "192.168.2.1/24"1392 self.base_profile_config["interfaces"][1]["captive"] = max_client1393 del self.base_profile_config["interfaces"][1]["ethernet"]1394 del self.base_profile_config["interfaces"][1]["services"]1395 del self.base_profile_config["metrics"]["wifi-frames"]1396 del self.base_profile_config["metrics"]["dhcp-snooping"]1397 # print(self.base_profile_config)1398 # print(self.base_profile_config, file=sourceFile)1399 # sourceFile.close()1400 def encryption_lookup(self, encryption="psk"):1401 encryption_mapping = {1402 "none": "open",1403 "psk": "wpa",1404 "psk2": "wpa2",1405 "sae": "wpa3",1406 "psk-mixed": "wpa|wpa2",1407 "sae-mixed": "wpa3",1408 "wpa": 'wap',1409 "wpa2": "eap",1410 "wpa3": "eap",1411 "wpa-mixed": "eap",1412 "wpa3-mixed": "sae"1413 }1414 if encryption in encryption_mapping.keys():1415 return encryption_mapping[encryption]1416 else:1417 return False1418 def get_ssid_info(self):1419 ssid_info = []1420 for interfaces in self.base_profile_config["interfaces"]:1421 if "ssids" in interfaces.keys():1422 for ssid_data in interfaces["ssids"]:1423 for band in ssid_data["wifi-bands"]:1424 temp = [ssid_data["name"]]1425 if ssid_data["encryption"]["proto"] == "none" or "radius" in ssid_data.keys():1426 temp.append(self.encryption_lookup(encryption=ssid_data["encryption"]["proto"]))1427 temp.append('[BLANK]')1428 else:1429 temp.append(self.encryption_lookup(encryption=ssid_data["encryption"]["proto"]))1430 temp.append(ssid_data["encryption"]["key"])1431 temp.append(band)1432 ssid_info.append(temp)1433 return ssid_info1434 def set_radio_config(self, radio_config={}):1435 base_radio_config_2g = {1436 "band": "2G",1437 "country": "CA",1438 "channel-mode": "HE",1439 "channel": "auto"1440 }1441 base_radio_config_5g = {1442 "band": "5G",1443 "country": "CA",1444 "allow-dfs": True,1445 "channel-mode": "HE",1446 "channel": "auto"1447 }1448 for band in radio_config:1449 if band == "2G":1450 for keys in radio_config[band]:1451 base_radio_config_2g[keys] = radio_config[band][keys]1452 if band == "5G":1453 for keys in radio_config[band]:1454 base_radio_config_5g[keys] = radio_config[band][keys]1455 # if band == "6G":1456 # for keys in radio_config[band]:1457 # base_radio_config_6g[keys] = radio_config[band][keys]1458 self.base_profile_config["radios"].append(base_radio_config_2g)1459 self.base_profile_config["radios"].append(base_radio_config_5g)1460 print(self.base_profile_config)1461 self.vlan_section["ssids"] = []1462 self.vlan_ids = []1463 def set_mode(self, mode, mesh=False):1464 self.mode = mode1465 if mode == "NAT":1466 if mesh:1467 self.base_profile_config['interfaces'][0]['tunnel'] = {1468 "proto": "mesh"1469 }1470 self.base_profile_config['interfaces'][1]['ssids'] = []1471 elif mode == "BRIDGE":1472 if mesh:1473 self.base_profile_config['interfaces'][0]['tunnel'] = {1474 "proto": "mesh"1475 }1476 self.base_profile_config['interfaces'][0]['ssids'] = []1477 elif mode == "VLAN":1478 if mesh:1479 self.base_profile_config['interfaces'][0]['tunnel'] = {1480 "proto": "mesh"1481 }1482 del self.base_profile_config['interfaces'][1]1483 self.base_profile_config['interfaces'][0]['ssids'] = []1484 self.base_profile_config['interfaces'] = []1485 wan_section_vlan = {1486 "name": "WAN",1487 "role": "upstream",1488 "services": ["lldp", "ssh", "dhcp-snooping"],1489 "ethernet": [1490 {1491 "select-ports": [1492 "WAN*"1493 ]1494 }1495 ],1496 "ipv4": {1497 "addressing": "dynamic"1498 }1499 }1500 self.base_profile_config['interfaces'].append(wan_section_vlan)1501 else:1502 print("Invalid Mode")1503 return 01504 def add_ssid(self, ssid_data, radius=False, radius_auth_data={}, radius_accounting_data={}):1505 print("ssid data : ", ssid_data)1506 ssid_info = {'name': ssid_data["ssid_name"], "bss-mode": "ap", "wifi-bands": [], "services": ["wifi-frames"]}1507 for options in ssid_data:1508 if options == "multi-psk":1509 ssid_info[options] = ssid_data[options]1510 print("hi", ssid_info)1511 if options == "rate-limit":1512 ssid_info[options] = ssid_data[options]1513 for i in ssid_data["appliedRadios"]:1514 ssid_info["wifi-bands"].append(i)1515 ssid_info['encryption'] = {}1516 ssid_info['encryption']['proto'] = ssid_data["security"]1517 try:1518 ssid_info['encryption']['key'] = ssid_data["security_key"]1519 except Exception as e:1520 pass1521 ssid_info['encryption']['ieee80211w'] = "optional"1522 if radius:1523 ssid_info["radius"] = {}1524 ssid_info["radius"]["authentication"] = {1525 "host": radius_auth_data["ip"],1526 "port": radius_auth_data["port"],1527 "secret": radius_auth_data["secret"]1528 }1529 ssid_info["radius"]["accounting"] = {1530 "host": radius_accounting_data["ip"],1531 "port": radius_accounting_data["port"],1532 "secret": radius_accounting_data["secret"]1533 }1534 if self.mode == "NAT":1535 self.base_profile_config['interfaces'][1]['ssids'].append(ssid_info)1536 elif self.mode == "BRIDGE":1537 self.base_profile_config['interfaces'][0]['ssids'].append(ssid_info)1538 elif self.mode == "VLAN":1539 vid = ssid_data["vlan"]1540 self.vlan_section = {1541 "name": "WAN100",1542 "role": "upstream",1543 "services": ["lldp", "dhcp-snooping"],1544 "vlan": {1545 "id": 1001546 },1547 "ethernet": [1548 {1549 "select-ports": [1550 "WAN*"1551 ]1552 }1553 ],1554 "ipv4": {1555 "addressing": "dynamic"1556 }1557 }1558 vlan_section = self.vlan_section1559 if vid in self.vlan_ids:1560 print("sss", self.vlan_ids)1561 for i in self.base_profile_config['interfaces']:1562 if i["name"] == "WANv%s" % (vid):1563 i["ssids"].append(ssid_info)1564 else:1565 print(self.vlan_ids)1566 self.vlan_ids.append(vid)1567 vlan_section['name'] = "WANv%s" % (vid)1568 vlan_section['vlan']['id'] = int(vid)1569 vlan_section["ssids"] = []1570 vlan_section["ssids"].append(ssid_info)1571 self.base_profile_config['interfaces'].append(vlan_section)1572 print(vlan_section)1573 vsection = 01574 else:1575 print("invalid mode")1576 pytest.exit("invalid Operating Mode")1577 def push_config(self, serial_number):1578 payload = {"configuration": self.base_profile_config, "serialNumber": serial_number, "UUID": 1}1579 uri = self.sdk_client.build_uri("device/" + serial_number + "/configure")1580 basic_cfg_str = json.dumps(payload)1581 print("Sending Command: " + "\n" +1582 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1583 "URI: " + str(uri) + "\n" +1584 "Data: " + str(payload) + "\n" +1585 "Headers: " + str(self.sdk_client.make_headers()))1586 allure.attach(name="Sending Command:", body="Sending Command: " + "\n" +1587 "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +1588 "URI: " + str(uri) + "\n" +1589 "Data: " + str(payload) + "\n" +1590 "Headers: " + str(self.sdk_client.make_headers()))1591 resp = requests.post(uri, data=basic_cfg_str, headers=self.sdk_client.make_headers(),1592 verify=False, timeout=100)1593 self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), basic_cfg_str, uri)1594 resp.close()1595 return resp1596if __name__ == '__main__':1597 controller = {1598 'url': 'https://sec-qa01.cicd.lab.wlan.tip.build:16001', # API base url for the controller1599 'username': "tip@ucentral.com",1600 'password': 'OpenWifi%123',1601 }1602 obj = Controller(controller_data=controller)1603 # po = ProvUtils(sdk_client=obj)1604 # print(po.get_inventory())1605 # up = UProfileUtility(sdk_client=obj, controller_data=controller)1606 # up.set_mode(mode="BRIDGE")1607 # up.set_radio_config()...

Full Screen

Full Screen

test_view_decorators.py

Source:test_view_decorators.py Github

copy

Full Screen

...42 fresh_access_token = create_access_token('username', fresh=True)43 refresh_token = create_refresh_token('username')44 # Access and fresh access should be able to access this45 for token in (access_token, fresh_access_token):46 response = test_client.get(url, headers=make_headers(token))47 assert response.status_code == 20048 assert response.get_json() == {'foo': 'bar'}49 # Test accessing jwt_required with no jwt in the request50 response = test_client.get(url, headers=None)51 assert response.status_code == 40152 assert response.get_json() == {'msg': 'Missing Authorization Header'}53 # Test refresh token access to jwt_required54 response = test_client.get(url, headers=make_headers(refresh_token))55 assert response.status_code == 42256 assert response.get_json() == {'msg': 'Only access tokens are allowed'}57def test_fresh_jwt_required(app):58 jwtM = get_jwt_manager(app)59 url = '/fresh_protected'60 test_client = app.test_client()61 with app.test_request_context():62 access_token = create_access_token('username')63 fresh_access_token = create_access_token('username', fresh=True)64 refresh_token = create_refresh_token('username')65 fresh_timed_access_token = create_access_token(66 identity='username',67 fresh=timedelta(minutes=5)68 )69 stale_timed_access_token = create_access_token(70 identity='username',71 fresh=timedelta(minutes=-1)72 )73 response = test_client.get(url, headers=make_headers(fresh_access_token))74 assert response.status_code == 20075 assert response.get_json() == {'foo': 'bar'}76 response = test_client.get(url, headers=make_headers(access_token))77 assert response.status_code == 40178 assert response.get_json() == {'msg': 'Fresh token required'}79 response = test_client.get(url, headers=make_headers(fresh_timed_access_token))80 assert response.status_code == 20081 assert response.get_json() == {'foo': 'bar'}82 response = test_client.get(url, headers=make_headers(stale_timed_access_token))83 assert response.status_code == 40184 assert response.get_json() == {'msg': 'Fresh token required'}85 response = test_client.get(url, headers=None)86 assert response.status_code == 40187 assert response.get_json() == {'msg': 'Missing Authorization Header'}88 response = test_client.get(url, headers=make_headers(refresh_token))89 assert response.status_code == 42290 assert response.get_json() == {'msg': 'Only access tokens are allowed'}91 # Test with custom response92 @jwtM.needs_fresh_token_loader93 def custom_response():94 return jsonify(msg='foobar'), 20195 response = test_client.get(url, headers=make_headers(access_token))96 assert response.status_code == 20197 assert response.get_json() == {'msg': 'foobar'}98def test_refresh_jwt_required(app):99 url = '/refresh_protected'100 test_client = app.test_client()101 with app.test_request_context():102 access_token = create_access_token('username')103 fresh_access_token = create_access_token('username', fresh=True)104 refresh_token = create_refresh_token('username')105 response = test_client.get(url, headers=make_headers(fresh_access_token))106 assert response.status_code == 422107 assert response.get_json() == {'msg': 'Only refresh tokens are allowed'}108 response = test_client.get(url, headers=make_headers(access_token))109 assert response.status_code == 422110 assert response.get_json() == {'msg': 'Only refresh tokens are allowed'}111 response = test_client.get(url, headers=None)112 assert response.status_code == 401113 assert response.get_json() == {'msg': 'Missing Authorization Header'}114 response = test_client.get(url, headers=make_headers(refresh_token))115 assert response.status_code == 200116 assert response.get_json() == {'foo': 'bar'}117@pytest.mark.parametrize("delta_func", [timedelta, relativedelta])118def test_jwt_optional(app, delta_func):119 url = '/optional_protected'120 test_client = app.test_client()121 with app.test_request_context():122 access_token = create_access_token('username')123 fresh_access_token = create_access_token('username', fresh=True)124 refresh_token = create_refresh_token('username')125 expired_token = create_access_token(126 identity='username',127 expires_delta=delta_func(minutes=-1)128 )129 response = test_client.get(url, headers=make_headers(fresh_access_token))130 assert response.status_code == 200131 assert response.get_json() == {'foo': 'baz'}132 response = test_client.get(url, headers=make_headers(access_token))133 assert response.status_code == 200134 assert response.get_json() == {'foo': 'baz'}135 response = test_client.get(url, headers=make_headers(refresh_token))136 assert response.status_code == 422137 assert response.get_json() == {'msg': 'Only access tokens are allowed'}138 response = test_client.get(url, headers=None)139 assert response.status_code == 200140 assert response.get_json() == {'foo': 'bar'}141 response = test_client.get(url, headers=make_headers(expired_token))142 assert response.status_code == 401143 assert response.get_json() == {'msg': 'Token has expired'}144def test_invalid_jwt(app):145 url = '/protected'146 jwtM = get_jwt_manager(app)147 test_client = app.test_client()148 invalid_token = "aaaaa.bbbbb.ccccc"149 # Test default response150 response = test_client.get(url, headers=make_headers(invalid_token))151 assert response.status_code == 422152 assert response.get_json() == {'msg': 'Invalid header padding'}153 # Test custom response154 @jwtM.invalid_token_loader155 def custom_response(err_str):156 return jsonify(msg='foobar'), 201157 response = test_client.get(url, headers=make_headers(invalid_token))158 assert response.status_code == 201159 assert response.get_json() == {'msg': 'foobar'}160def test_jwt_missing_claims(app):161 url = '/protected'162 test_client = app.test_client()163 token = encode_token(app, {'foo': 'bar'})164 response = test_client.get(url, headers=make_headers(token))165 assert response.status_code == 422166 assert response.get_json() == {'msg': 'Missing claim: identity'}167def test_jwt_invalid_audience(app):168 url = '/protected'169 test_client = app.test_client()170 # No audience claim expected or provided - OK171 access_token = encode_token(app, {'identity': 'me'})172 response = test_client.get(url, headers=make_headers(access_token))173 assert response.status_code == 200174 # Audience claim expected and not provided - not OK175 app.config['JWT_DECODE_AUDIENCE'] = 'my_audience'176 access_token = encode_token(app, {'identity': 'me'})177 response = test_client.get(url, headers=make_headers(access_token))178 assert response.status_code == 422179 assert response.get_json() == {'msg': 'Token is missing the "aud" claim'}180 # Audience claim still expected and wrong one provided - not OK181 access_token = encode_token(app, {'aud': 'different_audience', 'identity': 'me'})182 response = test_client.get(url, headers=make_headers(access_token))183 assert response.status_code == 422184 assert response.get_json() == {'msg': 'Invalid audience'}185def test_jwt_invalid_issuer(app):186 url = '/protected'187 test_client = app.test_client()188 # No issuer claim expected or provided - OK189 access_token = encode_token(app, {'identity': 'me'})190 response = test_client.get(url, headers=make_headers(access_token))191 assert response.status_code == 200192 # Issuer claim expected and not provided - not OK193 app.config['JWT_DECODE_ISSUER'] = 'my_issuer'194 access_token = encode_token(app, {'identity': 'me'})195 response = test_client.get(url, headers=make_headers(access_token))196 assert response.status_code == 422197 assert response.get_json() == {'msg': 'Token is missing the "iss" claim'}198 # Issuer claim still expected and wrong one provided - not OK199 access_token = encode_token(app, {'iss': 'different_issuer', 'identity': 'me'})200 response = test_client.get(url, headers=make_headers(access_token))201 assert response.status_code == 422202 assert response.get_json() == {'msg': 'Invalid issuer'}203def test_malformed_token(app):204 url = '/protected'205 test_client = app.test_client()206 access_token = 'foobarbaz'207 response = test_client.get(url, headers=make_headers(access_token))208 assert response.status_code == 422209 assert response.get_json() == {'msg': 'Not enough segments'}210@pytest.mark.parametrize("delta_func", [timedelta, relativedelta])211def test_expired_token(app, delta_func):212 url = '/protected'213 jwtM = get_jwt_manager(app)214 test_client = app.test_client()215 with app.test_request_context():216 token = create_access_token('username', expires_delta=delta_func(minutes=-1))217 # Test default response218 response = test_client.get(url, headers=make_headers(token))219 assert response.status_code == 401220 assert response.get_json() == {'msg': 'Token has expired'}221 # Test depreciated custom response222 @jwtM.expired_token_loader223 def depreciated_custom_response():224 return jsonify(msg='foobar'), 201225 warnings.simplefilter("always")226 with warnings.catch_warnings(record=True) as w:227 response = test_client.get(url, headers=make_headers(token))228 assert response.status_code == 201229 assert response.get_json() == {'msg': 'foobar'}230 assert w[0].category == DeprecationWarning231 # Test new custom response232 @jwtM.expired_token_loader233 def custom_response(token):234 assert token['identity'] == 'username'235 assert token['type'] == 'access'236 return jsonify(msg='foobar'), 201237 warnings.simplefilter("always")238 with warnings.catch_warnings(record=True) as w:239 response = test_client.get(url, headers=make_headers(token))240 assert response.status_code == 201241 assert response.get_json() == {'msg': 'foobar'}242 assert len(w) == 0243def test_expired_token_via_decode_token(app):244 jwtM = get_jwt_manager(app)245 @jwtM.expired_token_loader246 def depreciated_custom_response(expired_token):247 assert expired_token['identity'] == 'username'248 return jsonify(msg='foobar'), 401249 @app.route('/test')250 def test_route():251 token = create_access_token('username', expires_delta=timedelta(minutes=-1))252 decode_token(token)253 return jsonify(msg='baz'), 200254 test_client = app.test_client()255 response = test_client.get('/test')256 assert response.get_json() == {'msg': 'foobar'}257 assert response.status_code == 401258def test_no_token(app):259 url = '/protected'260 jwtM = get_jwt_manager(app)261 test_client = app.test_client()262 # Test default response263 response = test_client.get(url, headers=None)264 assert response.status_code == 401265 assert response.get_json() == {'msg': 'Missing Authorization Header'}266 # Test custom response267 @jwtM.unauthorized_loader268 def custom_response(err_str):269 return jsonify(msg='foobar'), 201270 response = test_client.get(url, headers=None)271 assert response.status_code == 201272 assert response.get_json() == {'msg': 'foobar'}273def test_different_token_algorightm(app):274 url = '/protected'275 test_client = app.test_client()276 with app.test_request_context():277 token = create_access_token('username')278 app.config['JWT_ALGORITHM'] = 'HS512'279 response = test_client.get(url, headers=make_headers(token))280 assert response.status_code == 422...

Full Screen

Full Screen

api_lib.py

Source:api_lib.py Github

copy

Full Screen

...4 """5 def __init__(self, base_url='http://localhost:8000/api/'): 6 self.base_url = base_url7 self.token = None8 def make_headers(self):9 headers = {'Accept': 'application/json', 'Content-Type': 'application/json;charset=utf-8'}10 if self.token:11 headers['Authorization'] = 'Bearer ' + self.token12 return headers13 def login(self, login, password):14 url = f'{self.base_url}/token/'15 headers = self.make_headers()16 body = {'username': login, 'password': password}17 response, status_code = request.post(url, headers, body)18 try:19 self.token = response['access']20 return status_code, 'You are logged in'21 except:22 return status_code, response23 def get_publications(self):24 url = f'{self.base_url}/publication/'25 headers = self.make_headers()26 response = request.get(url, headers)27 return response28 def get_publication(self, id):29 url = f'{self.base_url}/publication/{id}/'30 headers = self.make_headers()31 response = request.get(url, headers)32 return response33 def create_publication(self, data):34 body = data[0]35 url = f'{self.base_url}/publication/'36 headers = self.make_headers()37 response = request.post(url, headers, body)38 return response39 def update_publication(self, data):40 body = data[0]41 id = body['id']42 url = f'{self.base_url}/publication/{id}/'43 headers = self.make_headers()44 response = request.put(url, headers, body)45 return response46 47 def update_category(self, data):48 body = data[0]49 id = body['id']50 url = f'{self.base_url}/publication/{id}/'51 headers = self.make_headers()52 response = request.put(url, headers, body)53 return response54 def get_categories(self):55 url = f'{self.base_url}/category/'56 headers = self.make_headers()57 response = request.get(url, headers)58 return response59 def get_category(self, id):60 url = f'{self.base_url}/category/{id}/'61 headers = self.make_headers()62 response = request.get(url, headers)63 return response64 def create_category(self, data):65 body = data[0]66 url = f'{self.base_url}/category/'67 headers = self.make_headers()68 response = request.post(url, headers, body)...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful