Last active
December 16, 2018 15:32
-
-
Save VehpuS/5e4dc5106e7b9c7e7aa83ae05b334fc5 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests # pip install requests | |
from functools import reduce | |
import json | |
from pprint import pprint, pformat | |
def create_mlnx_os_session(switch_ip, user_name, password, https=True, verify=False): | |
''' | |
@summary: | |
Using the requests library create a session with a MLNX-OS running machine, that can serve both web and JSON requests | |
@param switch_ip: | |
The IP of the machine running MLNX-OS, as a string. | |
@param user_name: | |
The user name we want to log in to. | |
@param password: | |
The password for the user. | |
@param https: | |
Boolean. If True, connect via HTTPS. | |
@param verify: | |
Boolean. | |
If True, https certificate will be verified and the function will fail if verification failed. | |
If False, failed verification will only log a warning to screen. | |
Default is False. | |
@return: | |
requests.session object with the logged in session's cookie. | |
@raise AssertionError: | |
If the login failed for any reason. | |
''' | |
session = requests.session() | |
session.verify = verify | |
# URL:http://<switch-ip-address>/admin/launch?script=rh&template=login&action=login | |
http_prefix = "https://" if https else "http://" | |
url = http_prefix + switch_ip + "/admin/launch?script=rh&template=login&action=login" | |
# data:"f_user_id=<user name>&f_password=<user password>" | |
data = { | |
"f_user_id": user_name, | |
"f_password": password, | |
} | |
headers = { | |
"Content-Type": "application/x-www-form-urlencoded", | |
} | |
r = session.post(url, data=data, headers=headers) | |
assert r.ok and "not logged in" not in str(r.content), "Failed to login with url: " + r.url | |
session.cookies = requests.cookies.cookiejar_from_dict( | |
{cookie.name: cookie.value | |
for old_r in r.history | |
for cookie in old_r.cookies} | |
) | |
return session | |
def get_sync_json_data(commands, session, switch_ip, https=True): | |
''' | |
@summary: | |
Run a MLNX-OS CLI command via the JSON API synchronously, and get the results | |
@param commands: | |
A list of commands to run through the JSON API. | |
@param session: | |
requests.session object with the logged in session's cookie. | |
@param switch_ip: | |
The IP of the machine running MLNX-OS, as a string. | |
@param https: | |
Boolean. If True, connect via HTTPS. | |
@return; | |
A list of responses matching each command in commands. | |
@raise AssertionError: | |
If the JSON response was erroraneous in any form. | |
''' | |
http_prefix = "https://" if https else "http://" | |
url = http_prefix + switch_ip + "/admin/launch?script=json" | |
response = session.post(url, | |
json={ | |
"execution_type": "sync", | |
"commands": commands | |
}) | |
json_err = ("JSON request failed with the following message: " + | |
str(response.content)) | |
assert response.ok, json_err | |
data = json.loads(response.content) | |
assert "results" in data, json_err | |
err_data = [entry for entry in data["results"] | |
if "status" in entry and entry["status"] == "ERROR"] | |
assert err_data == [], "The following requests failed:\n{}".format(pformat(err_data)) | |
try: | |
data = [entry["data"] for entry in data["results"]] | |
except BaseException as err: | |
print("Unexpeted data format - missing the 'data' key: {}".format(data)) | |
raise | |
return data |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment