# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from RequestsLibrary import RequestsLibrary
-from robot.api import logger
+
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
-import hashlib
-from ONAPLibrary.Base64Keywords import Base64Keywords
from ONAPLibrary.RequestsHelper import RequestsHelper
@keyword
def run_get_request(self, endpoint, data_path, user, accept="application/json", auth=None):
"""Runs an SDC get request"""
- resp = self.get_request(endpoint, data_path, user, accept, auth)
+ resp = self.reqs.get_request("sdc", endpoint, data_path, user, accept, auth)
self.builtin.should_be_equal_as_strings(resp.status_code, "200")
return resp
@keyword
def run_post_request(self, endpoint, data_path, data, user, accept="application/json", auth=None):
"""Runs an SDC post request"""
- return self.post_request(endpoint, data_path, data, user, files=None, accept=accept, auth=auth)
+ return self.reqs.post_request("sdc", endpoint, data_path, data, user, files=None, accept=accept, auth=auth)
@keyword
def run_post_files_request(self, endpoint, data_path, files, user, accept="application/json", auth=None):
"""Runs an SDC post files request"""
- return self.post_request(endpoint, data_path, files, user, files=None, accept=accept,
- content_type="multipart/form-data", auth=auth)
+ return self.reqs.post_request("sdc", endpoint, data_path, files, user, files=None, accept=accept,
+ content_type="multipart/form-data", auth=auth)
@keyword
def run_put_request(self, endpoint, data_path, data, user, accept="application/json", auth=None):
"""Runs an SDC post request"""
- return self.put_request(endpoint, data_path, data, user, accept, auth)
-
- def get_request(self, endpoint, data_path, user, accept="application/json", auth=None):
- """Runs an SDC get request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdc", endpoint, auth=auth)
- headers = self.reqs.create_headers(sdc_user_id=user, accept=accept)
- resp = RequestsLibrary().get_request("sdc", data_path, headers=headers)
- logger.info("Received response from sdc " + resp.text)
- return resp
-
- def post_request(self, endpoint, data_path, data, user, files=None, accept="application/json",
- content_type="application/json", auth=None):
- """Runs an SDC post request"""
- logger.info("Creating session" + endpoint)
- md5 = hashlib.md5()
- md5.update(data)
- md5checksum = Base64Keywords().base64_encode(md5.hexdigest())
- RequestsLibrary().create_session("sdc", endpoint, auth=auth)
- headers = self.reqs.create_headers(user, accept=accept, content_type=content_type, md5=md5checksum)
- resp = RequestsLibrary().post_request("sdc", data_path, files=files, data=data, headers=headers)
-
- logger.info("Received response from sdc " + resp.text)
- return resp
-
- def put_request(self, endpoint, data_path, data, user, accept="application/json", auth=None):
- """Runs an SDC post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdc", endpoint, auth=auth)
- headers = self.reqs.create_headers(sdc_user_id=user, accept=accept)
- resp = RequestsLibrary().put_request("sdc", data_path, data=data, headers=headers)
- logger.info("Received response from sdc " + resp.text)
- return resp
+ return self.reqs.put_request("sdc", endpoint, data_path, data, user, accept, auth)
- def delete_request(self, endpoint, data_path, data, user, accept="application/json", auth=None):
- """Runs an SDC post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdc", endpoint, auth=auth)
- headers = self.reqs.create_headers(sdc_user_id=user, accept=accept)
- resp = RequestsLibrary().delete_request("sdc", data_path, data=data, headers=headers)
- logger.info("Received response from sdc " + resp.text)
- return resp
+ @keyword
+ def run_delete_request(self, endpoint, data_path, data, user, accept="application/json", auth=None):
+ """Runs an SDC delete request"""
+ return self.reqs.delete_request("sdc", endpoint, data_path, data, user, accept, auth)
@keyword
def run_get_request(self, endpoint, data_path, accept="application/json", auth=None):
"""Runs an SDNC get request"""
- resp = self.get_request(endpoint, data_path, accept, auth)
+ resp = self.reqs.get_request("sdnc", endpoint, data_path, accept, auth)
self.builtin.should_be_equal_as_strings(resp.status_code, "200")
return resp
@keyword
def run_post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
"""Runs an SDNC post request"""
- return self.post_request(endpoint, data_path, data, accept, auth)
+ return self.reqs.post_request("sdnc", endpoint, data_path, data, accept, auth)
@keyword
def run_put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
"""Runs an SDNC post request"""
- return self.put_request(endpoint, data_path, data, accept, auth)
-
- def get_request(self, endpoint, data_path, accept="application/json", auth=None):
- """Runs an SDNC get request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
- resp = RequestsLibrary().get_request("sdnc", data_path, headers=self.reqs.create_headers(accept))
- logger.info("Received response from sdnc " + resp.text)
- return resp
-
- def post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an sdnc post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().post_request("sdnc", data_path, data=data, headers=headers)
- logger.info("Received response from sdnc " + resp.text)
- return resp
-
- def put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an sdnc post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().put_request("sdnc", data_path, data=data, headers=headers)
- logger.info("Received response from sdnc " + resp.text)
- return resp
-
- def delete_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an sdnc post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().delete_request("sdnc", data_path, data=data, headers=headers)
- logger.info("Received response from sdnc " + resp.text)
- return resp
+ return self.reqs.put_request("sdnc", endpoint, data_path, data, accept, auth)
\ No newline at end of file
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from RequestsLibrary import RequestsLibrary
-from robot.api import logger
+
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
@keyword
def run_get_request(self, endpoint, data_path, accept="application/json", auth=None):
"""Runs an SO get request"""
- resp = self.get_request(endpoint, data_path, accept, auth)
+ resp = self.reqs.get_request("so", endpoint, data_path, accept, auth)
self.builtin.should_be_equal_as_strings(resp.status_code, "200")
return resp
@keyword
def run_post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
"""Runs an SO post request"""
- return self.post_request(endpoint, data_path, data, accept, auth)
+ return self.reqs.post_request("so", endpoint, data_path, data, accept, auth)
@keyword
def run_put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
"""Runs an SO post request"""
- return self.put_request(endpoint, data_path, data, accept, auth)
-
- def get_request(self, endpoint, data_path, accept="application/json", auth=None):
- """Runs an SO get request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("so", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().get_request("so", data_path, headers=headers)
- logger.info("Received response from so " + resp.text)
- return resp
-
- def post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an SO post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("so", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().post_request("so", data_path, data=data, headers=headers)
- logger.info("Received response from so " + resp.text)
- return resp
-
- def put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an SO post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("so", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().put_request("so", data_path, data=data, headers=headers)
- logger.info("Received response from so " + resp.text)
- return resp
-
- def delete_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an SO post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("so", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().delete_request("so", data_path, data=data, headers=headers)
- logger.info("Received response from so " + resp.text)
- return resp
+ return self.reqs.put_request("so", endpoint, data_path, data, accept, auth)
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from RequestsLibrary import RequestsLibrary
-from robot.api import logger
+
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
@keyword
def run_get_request(self, endpoint, data_path, accept="application/json", auth=None):
"""Runs an MUSIC get request"""
- resp = self.get_request(endpoint, data_path, accept, auth)
- return resp
-
- def get_request(self, endpoint, data_path, accept="application/json", auth=None):
- """Runs an MUSIC get request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("music", endpoint, auth=auth)
- resp = RequestsLibrary().get_request("music", data_path, headers=self.reqs.create_headers(accept=accept))
- logger.info("Received response from music " + resp.text)
+ resp = self.reqs.get_request("music", endpoint, data_path, accept, auth)
return resp
def run_health_check(self, endpoint, health_check_path):
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.BaseSDNCKeywords import BaseSDNCKeywords
-from ONAPLibrary.UUIDKeywords import UUIDKeywords
+
+from ONAPLibrary.RequestsHelper import RequestsHelper
from ONAPLibrary.TemplatingKeywords import TemplatingKeywords
from robot.api.deco import keyword
-from robot.libraries.BuiltIn import BuiltIn
-from robot.libraries.OperatingSystem import OperatingSystem
class PreloadSDNCKeywords(object):
def __init__(self):
super(PreloadSDNCKeywords, self).__init__()
- self.application_id = "robot-ete"
- self.builtin = BuiltIn()
- self.base_keywords = BaseSDNCKeywords()
- self.utilities = UUIDKeywords()
- self.os = OperatingSystem()
+ self.reqs = RequestsHelper()
self.templating = TemplatingKeywords()
@keyword
"""Runs an SDNC request to preload certain data."""
self.templating.create_environment("sdnc", templates_folder)
data = self.templating.apply_template("sdnc", template, preload_dictionary)
- return self.base_keywords.post_request(endpoint, data_path, data)
+ return self.reqs.post_request("sdnc", endpoint, data_path, data)
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.BaseSOKeywords import BaseSOKeywords
+
+from ONAPLibrary.RequestsHelper import RequestsHelper
from robot.api import logger
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
super(RequestSOKeywords, self).__init__()
self.application_id = "robot-ete"
self.builtin = BuiltIn()
- self.base_keywords = BaseSOKeywords()
+ self.reqs = RequestsHelper()
@keyword
def run_polling_get_request(self, endpoint, data_path, complete_states=None, fail_states=None, tries=20,
complete_states = ["COMPLETE"]
# do this until it is done
for i in range(tries):
- resp = self.base_keywords.get_request(endpoint, data_path, auth=auth)
+ resp = self.reqs.get_request("so", endpoint, data_path, auth=auth)
self.builtin.should_not_contain_any(resp.text, fail_states)
logger.info(resp.json()['request']['requestStatus']['requestState'])
if resp.json()['request']['requestStatus']['requestState'] in complete_states:
@keyword
def run_create_request(self, endpoint, data_path, data, auth=None):
"""Runs an SO create request and returns the request id and instance id."""
- response = self.base_keywords.post_request(endpoint, data_path, data, auth=auth)
+ response = self.reqs.post_request("so", endpoint, data_path, data, auth=auth)
logger.info("Creation request submitted to SO, got response")
req_id = response.get('requestReferences', {}).get('requestId', '')
# limitations under the License.
from ONAPLibrary.UUIDKeywords import UUIDKeywords
+from RequestsLibrary import RequestsLibrary
+from robot.api import logger
+import hashlib
+from ONAPLibrary.Base64Keywords import Base64Keywords
class RequestsHelper(object):
- """ non keywords methods related to Requests library """
+ """ non keyword methods related to Requests library """
def __init__(self):
super(RequestsHelper, self).__init__()
self.uuid = UUIDKeywords()
self.application_id = "robot-ete"
+ self.requests = RequestsLibrary()
def create_headers(self, sdc_user_id=None, accept="application/json", content_type="application/json", md5=None):
"""Create the headers that are used by onap"""
if not md5:
headers["Content-MD5"] = md5
return headers
+
+ def get_request(self, alias, endpoint, data_path, sdc_user=None, accept="application/json", auth=None):
+ """Runs a get request"""
+ logger.info("Creating session" + endpoint)
+ self.requests.create_session(alias, endpoint, auth=auth)
+ headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
+ resp = self.requests.get_request(alias, data_path, headers=headers)
+ logger.info("Received response from [" + alias + "]: " + resp.text)
+ return resp
+
+ def post_request(self, alias, endpoint, data_path, data, sdc_user=None, files=None, accept="application/json",
+ content_type="application/json", auth=None):
+ """Runs a post request"""
+ logger.info("Creating session" + endpoint)
+ md5 = hashlib.md5()
+ md5.update(data)
+ md5checksum = Base64Keywords().base64_encode(md5.hexdigest())
+ self.requests.create_session(alias, endpoint, auth=auth)
+ headers = self.create_headers(sdc_user_id=sdc_user, accept=accept, content_type=content_type, md5=md5checksum)
+ resp = self.requests.post_request(alias, data_path, files=files, data=data, headers=headers)
+ logger.info("Received response from [" + alias + "]: " + resp.text)
+ return resp
+
+ def put_request(self, alias, endpoint, data_path, data, sdc_user=None, accept="application/json", auth=None):
+ """Runs a put request"""
+ logger.info("Creating session" + endpoint)
+ self.requests.create_session(alias, endpoint, auth=auth)
+ headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
+ resp = self.requests.put_request(alias, data_path, data=data, headers=headers)
+ logger.info("Received response from [" + alias + "]: " + resp.text)
+ return resp
+
+ def delete_request(self, alias, endpoint, data_path, data=None, sdc_user=None, accept="application/json", auth=None):
+ """Runs a delete request"""
+ logger.info("Creating session" + endpoint)
+ self.requests.create_session(alias, endpoint, auth=auth)
+ headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
+ resp = self.requests.delete_request(alias, data_path, data=data, headers=headers)
+ logger.info("Received response from [" + alias + "]: " + resp.text)
+ return resp
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from RequestsLibrary import RequestsLibrary
+
from robot.api import logger
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
@keyword
def run_sniro_get_request(self, endpoint, data_path, accept="application/json", auth=None):
"""Runs OOF-SNIRO Get request"""
- resp = self.get_request(endpoint, data_path, accept, auth)
+ resp = self.reqs.get_request("oof-sniro", endpoint, data_path, accept, auth)
self.builtin.should_be_equal_as_strings(resp.status_code, "200")
return resp
@keyword
def reset_sniro(self, endpoint):
logger.debug('Clearing SNIRO data')
- resp = self.post_request(endpoint, '/reset', None)
+ resp = self.reqs.post_request("oof-sniro", endpoint, '/reset', None)
self.builtin.should_be_equal_as_strings(resp.status_code, "200", 'Clearing SNIRO date failed.')
@keyword
base64_sniro_data = self.base64.base64_encode(sniro_data)
replace_dict = {'base64_sniro_data': base64_sniro_data}
sniro_request = self.templating.apply_template("sniro", template_sniro_request, replace_dict)
- resp = self.post_request(endpoint, '/', sniro_request)
+ resp = self.reqs.post_request("oof-sniro", endpoint, '/', sniro_request)
self.builtin.should_be_equal_as_strings(resp.status_code, "200", 'SNIRO preloading failed.')
return True
-
- def post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
- """Runs an SNIRO post request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("so", endpoint, auth=auth)
- headers = self.reqs.create_headers(accept=accept)
- resp = RequestsLibrary().post_request("so", data_path, data=data, headers=headers)
- logger.info("Received response from so " + resp.text)
- return resp
-
- def get_request(self, endpoint, data_path, accept="application/json", auth=None):
- """Runs an SNIRO get request"""
- logger.info("Creating session" + endpoint)
- RequestsLibrary().create_session("sniro", endpoint, auth=auth)
- resp = RequestsLibrary().get_request("sniro", data_path, headers=self.reqs.create_headers(accept=accept))
- logger.info("Received response from OOF-SNIRO " + resp.text)
- return resp