1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 from ONAPLibrary.UUIDKeywords import UUIDKeywords
16 from RequestsLibrary import RequestsLibrary
18 from ONAPLibrary.Base64Keywords import Base64Keywords
19 from ONAPLibrary.HTTPKeywords import HTTPKeywords
20 from ONAPLibrary.RequestsDecorators import log_wrapped
21 from ONAPLibrary.RequestsDecorators import default_keywords
24 class RequestsHelper(object):
25 """ non keyword methods related to Requests library """
28 super(RequestsHelper, self).__init__()
29 self.uuid = UUIDKeywords()
30 self.application_id = "robot-ete"
31 self.requests = RequestsLibrary()
32 self.http = HTTPKeywords()
36 def get_request(self, **kwargs):
37 """Runs a get request"""
38 return self.requests.get_request(alias=kwargs['alias'], uri=kwargs['data_path'],
39 headers=self._perform_setup(**kwargs))
43 def post_request(self, **kwargs):
44 """Runs a post request"""
45 kwargs['md5'] = self._format_md5(kwargs['data'])
46 return self.requests.post_request(alias=kwargs['alias'], uri=kwargs['data_path'], files=kwargs['files'],
47 data=kwargs['data'], headers=self._perform_setup(**kwargs))
51 def put_request(self, **kwargs):
52 """Runs a put request"""
53 return self.requests.put_request(alias=kwargs['alias'], uri=kwargs['data_path'], data=kwargs['data'],
54 headers=self._perform_setup(**kwargs))
58 def delete_request(self, **kwargs):
59 """Runs a delete request"""
60 return self.requests.delete_request(alias=kwargs['alias'], uri=kwargs['data_path'], data=kwargs['data'],
61 headers=self._perform_setup(**kwargs))
63 def _perform_setup(self, **kwargs):
64 self.http.disable_warnings()
65 self._create_session(alias=kwargs['alias'], endpoint=kwargs['endpoint'], auth=kwargs['auth'],
66 client_certs=kwargs['client_certs'])
67 return self._create_headers(sdc_user_id=kwargs['sdc_user'], accept=kwargs['accept'],
68 content_type=kwargs['content_type'], md5=kwargs.get("md5", None))
70 def _create_session(self, alias, endpoint, auth=None, client_certs=None):
71 if client_certs is not None:
72 self.requests.create_client_cert_session(alias, endpoint, client_certs=client_certs)
74 self.requests.create_session(alias, endpoint, auth=auth)
76 def _create_headers(self, sdc_user_id=None, accept="application/json", content_type="application/json", md5=None):
77 """Create the headers that are used by onap"""
78 uuid = self.uuid.generate_uuid4()
81 "Content-Type": content_type,
82 "X-TransactionId": self.application_id + "-" + uuid,
83 "X-FromAppId": self.application_id
85 if sdc_user_id is not None:
86 headers["USER_ID"] = sdc_user_id
88 headers["Content-MD5"] = md5
92 def _format_md5(md5_input):
93 if md5_input is not None:
96 return Base64Keywords().base64_encode(md5.hexdigest())