mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 11:18:03 +00:00
179 lines
8.2 KiB
Python
179 lines
8.2 KiB
Python
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
|
|
# Class holds default settings for json requests -
|
|
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
|
|
import sys
|
|
|
|
if sys.version_info[0] != 3:
|
|
print("This script requires Python 3")
|
|
exit()
|
|
|
|
|
|
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
import json
|
|
from LANforge import LFUtils
|
|
|
|
|
|
class LFRequest:
|
|
Default_Base_URL = "http://localhost:8080"
|
|
No_Data = {'No Data':0}
|
|
requested_url = ""
|
|
post_data = No_Data
|
|
default_headers = {
|
|
'Accept': 'application/json'}
|
|
|
|
def __init__(self, url, uri=None):
|
|
if "http://" not in url or "https://" not in url:
|
|
print("No http:// or https:// found, prepending http://")
|
|
url = "http://" + url
|
|
if not url.endswith('/') and not uri.startswith('/'):
|
|
url += '/'
|
|
if uri is not None:
|
|
self.requested_url = url + uri
|
|
if self.requested_url.find('//'):
|
|
protopos = self.requested_url.find("://")
|
|
self.requested_url = self.requested_url[:protopos + 2] + self.requested_url[protopos + 2:].replace("//", "/")
|
|
|
|
|
|
|
|
# request first url on stack
|
|
def formPost(self, show_error=True):
|
|
responses = []
|
|
urlenc_data = ""
|
|
if ((self.post_data != None) and (self.post_data is not self.No_Data)):
|
|
urlenc_data = urllib.parse.urlencode(self.post_data).encode("utf-8")
|
|
#print("data looks like:" + str(urlenc_data))
|
|
request = urllib.request.Request(url=self.requested_url,
|
|
data=urlenc_data,
|
|
headers=self.default_headers)
|
|
else:
|
|
request = urllib.request.Request(url=self.requested_url, headers=self.default_headers)
|
|
print("No data for this formPost?")
|
|
|
|
request.headers['Content-type'] = 'application/x-www-form-urlencoded'
|
|
resp = ''
|
|
try:
|
|
resp = urllib.request.urlopen(request)
|
|
responses.append(resp)
|
|
return responses[0]
|
|
except urllib.error.HTTPError as error:
|
|
if (show_error):
|
|
print("----- formPost() HTTPError: --------------------------------------------")
|
|
print("%s: %s; URL: %s"%(error.code, error.reason, request.get_full_url()))
|
|
LFUtils.debug_printer.pprint(error.headers)
|
|
#print("Error: ", sys.exc_info()[0])
|
|
#print("Request URL:", request.get_full_url())
|
|
print("Request Content-type:", request.get_header('Content-type'))
|
|
print("Request Accept:", request.get_header('Accept'))
|
|
print("Request Data:")
|
|
LFUtils.debug_printer.pprint(request.data)
|
|
if (len(responses) > 0):
|
|
print("----- Response: --------------------------------------------------------")
|
|
LFUtils.debug_printer.pprint(responses[0].reason)
|
|
|
|
print("------------------------------------------------------------------------")
|
|
except urllib.error.URLError as uerror:
|
|
if (show_error):
|
|
print("----- formPost() URLError: ---------------------------------------------")
|
|
print("Reason: %s; URL: %s"%(uerror.reason, request.get_full_url()))
|
|
print("------------------------------------------------------------------------")
|
|
|
|
return None
|
|
|
|
def jsonPost(self, show_error=True):
|
|
responses = []
|
|
if ((self.post_data != None) and (self.post_data is not self.No_Data)):
|
|
request = urllib.request.Request(url=self.requested_url,
|
|
data=json.dumps(self.post_data).encode("utf-8"),
|
|
headers=self.default_headers)
|
|
else:
|
|
request = urllib.request.Request(url=self.requested_url, headers=self.default_headers)
|
|
print("No data for this jsonPost?")
|
|
|
|
request.headers['Content-type'] = 'application/json'
|
|
try:
|
|
resp = urllib.request.urlopen(request)
|
|
responses.append(resp)
|
|
return responses[0]
|
|
except urllib.error.HTTPError as error:
|
|
if show_error:
|
|
print("----- jsonPost() HTTPError: --------------------------------------------")
|
|
print("<%s> HTTP %s: %s"%(request.get_full_url(), error.code, error.reason, ))
|
|
|
|
print("Error: ", sys.exc_info()[0])
|
|
print("Request URL:", request.get_full_url())
|
|
print("Request Content-type:", request.get_header('Content-type'))
|
|
print("Request Accept:", request.get_header('Accept'))
|
|
print("Request Data:")
|
|
LFUtils.debug_printer.pprint(request.data)
|
|
|
|
if error.headers:
|
|
# the HTTPError is of type HTTPMessage a subclass of email.message
|
|
# print(type(error.keys()))
|
|
for headername in sorted(error.headers.keys()):
|
|
print ("Response %s: %s "%(headername, error.headers.get(headername)))
|
|
|
|
if len(responses) > 0:
|
|
print("----- Response: --------------------------------------------------------")
|
|
LFUtils.debug_printer.pprint(responses[0].reason)
|
|
print("------------------------------------------------------------------------")
|
|
except urllib.error.URLError as uerror:
|
|
if show_error:
|
|
print("----- jsonPost() URLError: ---------------------------------------------")
|
|
print("Reason: %s; URL: %s"%(uerror.reason, request.get_full_url()))
|
|
print("------------------------------------------------------------------------")
|
|
return None
|
|
|
|
def get(self, show_error=True):
|
|
myrequest = urllib.request.Request(url=self.requested_url, headers=self.default_headers)
|
|
myresponses = []
|
|
try:
|
|
myresponses.append(urllib.request.urlopen(myrequest))
|
|
return myresponses[0]
|
|
except urllib.error.HTTPError as error:
|
|
if show_error:
|
|
print("----- get() HTTPError: --------------------------------------------")
|
|
print("<%s> HTTP %s: %s"%(myrequest.get_full_url(), error.code, error.reason, ))
|
|
if error.code != 404:
|
|
print("Error: ", sys.exc_info()[0])
|
|
print("Request URL:", myrequest.get_full_url())
|
|
print("Request Content-type:", myrequest.get_header('Content-type'))
|
|
print("Request Accept:", myrequest.get_header('Accept'))
|
|
print("Request Data:")
|
|
LFUtils.debug_printer.pprint(myrequest.data)
|
|
|
|
if error.headers:
|
|
# the HTTPError is of type HTTPMessage a subclass of email.message
|
|
# print(type(error.keys()))
|
|
for headername in sorted(error.headers.keys()):
|
|
print ("Response %s: %s "%(headername, error.headers.get(headername)))
|
|
|
|
if len(myresponses) > 0:
|
|
print("----- Response: --------------------------------------------------------")
|
|
LFUtils.debug_printer.pprint(myresponses[0].reason)
|
|
print("------------------------------------------------------------------------")
|
|
except urllib.error.URLError as uerror:
|
|
if show_error:
|
|
print("----- get() URLError: ---------------------------------------------")
|
|
print("Reason: %s; URL: %s"%(uerror.reason, myrequest.get_full_url()))
|
|
print("------------------------------------------------------------------------")
|
|
return None
|
|
|
|
def getAsJson(self, show_error=True):
|
|
responses = []
|
|
responses.append(self.get(show_error))
|
|
if (len(responses) < 1):
|
|
return None
|
|
if (responses[0] == None):
|
|
if (show_error):
|
|
print("No response from "+self.requested_url)
|
|
return None
|
|
json_data = json.loads(responses[0].read().decode('utf-8'))
|
|
return json_data
|
|
|
|
def addPostData(self, data):
|
|
self.post_data = data
|
|
|
|
# ~LFRequest
|