3 from collections import defaultdict
5 from hyper import HTTP20Connection
19 ADML_HOST = 'localhost'
21 ADML_ENDPOINT = ADML_HOST + ':' + ADML_PORT
25 CONTENT_LENGTH = 'content-length'
27 # Flow calculation throw admlf (ADML Flow) fixture:
28 # flow = admlf.getId()
30 # For sequenced tests, the id is just a monotonically increased number from 1.
31 # For parallel tests, this id is sequenced from 1 for every worker, then, globally,
32 # this is a handicap to manage flow ids (tests ids) for ADML FSM (finite state machine).
33 # We consider a base multiplier of 10000, so collisions would take place when workers
34 # reserves more than 10000 flows. With a 'worst-case' assumption of `5 flows per test case`,
35 # you should cover up to 5000 per worker. Anyway, feel free to increase this value,
36 # specially if you are thinking in use pytest and ADML Agent for system tests.
37 FLOW_BASE_MULTIPLIER = 10000
42 #def pytest_runtest_setup(item):
43 # print("pytest_runtest_setup")
44 #def pytest_runtest_logreport(report):
45 # print(f'Log Report:{report}')
46 #def pytest_sessionstart(session):
47 # print("pytest_session start")
48 #def pytest_sessionfinish(session):
49 # print("pytest_session finish")
51 ######################
52 # CLASSES & FIXTURES #
53 ######################
55 class Sequencer(object):
56 def __init__(self, request):
58 self.request = request
62 Returns the worker id, or 'master' if not parallel execution is done
65 if hasattr(self.request.config, 'slaveinput'):
66 wid = self.request.config.slaveinput['slaveid']
72 Returns the next identifier value (monotonically increased in every call)
80 # Workers are named: wd0, wd1, wd2, etc.
81 wid_number = int(re.findall(r'\d+', wid)[0])
83 return FLOW_BASE_MULTIPLIER * wid_number + self.sequence
87 @pytest.fixture(scope='session')
92 return Sequencer(request)
99 # CRITICAL ERROR WARNING INFO DEBUG NOSET
100 def setLevelInfo(): logging.getLogger().setLevel(logging.INFO)
101 def setLevelDebug(): logging.getLogger().setLevel(logging.DEBUG)
103 def error(message): logging.getLogger().error(message)
104 def warning(message): logging.getLogger().warning(message)
105 def info(message): logging.getLogger().info(message)
106 def debug(message): logging.getLogger().debug(message)
108 @pytest.fixture(scope='session')
112 MyLogger.logger = logging.getLogger('CT')
115 @pytest.fixture(scope='session')
118 message_bytes = message.encode('ascii')
119 base64_bytes = base64.b64encode(message_bytes)
120 return base64_bytes.decode('ascii')
124 @pytest.fixture(scope='session')
126 def decode(base64_message):
127 base64_bytes = base64_message.encode('ascii')
128 message_bytes = base64.b64decode(base64_bytes)
129 return message_bytes.decode('ascii')
132 @pytest.fixture(scope='session')
133 def saveB64Artifact(b64_decode):
135 Decode base64 string provided to disk
137 base64_message: base64 encoded string
138 file_basename: file basename where to write
139 isXML: decoded data corresponds to xml string. In this case, also json equivalente representation is written to disk
141 def save_b64_artifact(base64_message, file_basename, isXML = True):
143 targetFile = file_basename + ".txt"
144 data = b64_decode(base64_message)
147 targetFile = file_basename + ".xml"
149 _file = open(targetFile, "w")
150 n = _file.write(data)
154 targetFile = file_basename + ".json"
155 xml_dict = xmltodict.parse(data)
156 data = json.dumps(xml_dict, indent = 4, sort_keys=True)
158 _file = open(targetFile, "w")
159 n = _file.write(data)
162 return save_b64_artifact
165 # HTTP communication:
166 class RestClient(object):
167 """A client helper to perform rest operations: GET, POST.
170 endpoint: server endpoint to make the HTTP2.0 connection
173 def __init__(self, endpoint):
174 """Return a RestClient object for ADML endpoint."""
175 self._endpoint = endpoint
176 self._ip = self._endpoint.split(':')[0]
177 self._connection = HTTP20Connection(host=self._endpoint)
179 def _log_http(self, kind, method, url, body, headers):
180 length = len(body) if body else 0
182 '{} {}{} {} headers: {!s} data: {}:{!a}'.format(
183 method, self._endpoint, url, kind, headers, length, body))
185 def _log_request(self, method, url, body, headers):
186 self._log_http('REQUEST', method, url, body, headers)
188 def _log_response(self, method, url, response):
190 'RESPONSE:{}'.format(response["status"]), method, url,
191 response["body"], response["headers"])
193 #def log_event(self, level, log_msg):
194 # # Log caller function name and formated message
195 # MyLogger.logger.log(level, inspect.getouterframes(inspect.currentframe())[1].function + ': {!a}'.format(log_msg))
197 def parse(self, response):
198 response_body = response.read(decode_content=True).decode('utf-8')
199 if len(response_body) != 0:
200 response_body_dict = json.loads(response_body)
202 response_body_dict = ''
203 response_data = { "status":response.status, "body":response_body_dict, "headers":response.headers }
206 def request(self, requestMethod, requestUrl, requestBody=None, requestHeaders=None):
208 Returns response data dictionary with 'status', 'body' and 'headers'
210 requestBody = RestClient._pad_body_and_length(requestBody, requestHeaders)
211 self._log_request(requestMethod, requestUrl, requestBody, requestHeaders)
212 self._connection.request(method=requestMethod, url=requestUrl, body=requestBody, headers=requestHeaders)
213 response = self.parse(self._connection.get_response())
214 self._log_response(requestMethod, requestUrl, response)
217 def _pad_body_and_length(requestBody, requestHeaders):
218 """Pad the body and adjust content-length if needed.
219 When the length of the body is multiple of 1024 this function appends
220 one space to the body and increases by one the content-length.
222 This is a workaround for hyper issue 355 [0].
223 The issue has been fixed but it has not been released yet.
225 [0]: https://github.com/Lukasa/hyper/issues/355
228 >>> body, headers = ' '*1024, { 'content-length':'41' }
229 >>> body = RestClient._pad_body_and_length(body, headers)
230 >>> ( len(body), headers['content-length'] )
233 if requestBody and 0 == (len(requestBody) % 1024):
234 logging.warning( "RestClient.request:" +
235 " padding body because" +
236 " its length ({})".format(len(requestBody)) +
237 " is multiple of 1024")
239 content_length = CONTENT_LENGTH
240 if requestHeaders and content_length in requestHeaders:
241 length = int(requestHeaders[content_length])
242 requestHeaders[content_length] = str(length+1)
245 def get(self, requestUrl):
246 return self.request('GET', requestUrl)
248 def post(self, requestUrl, requestBody = None, requestHeaders={'content-type': 'application/json'}):
249 return self.request('POST', requestUrl, requestBody, requestHeaders)
251 def postDict(self, requestUrl, requestBody = None, requestHeaders={'content-type': 'application/json'}):
253 Accepts request body as python dictionary
255 requestBodyJson = None
256 if requestBody: requestBodyJson = json.dumps(requestBody, indent=None, separators=(',', ':'))
257 return self.request('POST', requestUrl, requestBodyJson, requestHeaders)
260 #def delete(self, requestUrl):
261 # return self.request('DELETE', requestUrl)
263 def __assert_received_expected(self, received, expected, what):
264 match = (received == expected)
265 log = "Received {what}: {received} | Expected {what}: {expected}".format(received=received, expected=expected, what=what)
266 if match: MyLogger.info(log)
267 else: MyLogger.error(log)
271 def check_response_status(self, received, expected, **kwargs):
273 received: status code received (got from response data parsed, field 'status')
274 expected: status code expected
276 self.__assert_received_expected(received, expected, "status code")
278 #def check_expected_cause(self, response, **kwargs):
280 # received: response data parsed where field 'body'->'cause' is analyzed
281 # kwargs: aditional regexp to match expected cause
283 # if "expected_cause" in kwargs:
284 # received_content = response["body"]
285 # received_cause = received_content.get("cause", "")
286 # regular_expr_cause = kwargs["expected_cause"]
287 # regular_expr_flag = kwargs.get("regular_expression_flag", 0)
288 # matchObj = re.match(regular_expr_cause, received_cause, regular_expr_flag)
289 # log = 'Test error cause: "{}"~=/{}/.'.format(received_cause, regular_expr_cause)
290 # if matchObj: MyLogger.info(log)
291 # else: MyLogger.error(log)
293 # assert matchObj is not None
295 def check_response_body(self, received, expected, inputJsonString = False):
297 received: body content received (got from response data parsed, field 'body')
298 expected: body content expected
299 inputJsonString: input parameters as json string (default are python dictionaries)
303 received = json.loads(received)
304 expected = json.loads(expected)
306 self.__assert_received_expected(received, expected, "body")
308 def check_response_headers(self, received, expected):
310 received: headers received (got from response data parsed, field 'headers')
311 expected: headers expected
313 self.__assert_received_expected(received, expected, "headers")
315 def assert_response__status_body_headers(self, response, status, bodyDict, headersDict = None):
317 response: Response parsed data
318 status: numeric status code
319 body: body dictionary to match with
320 headers: headers dictionary to match with (by default, not checked: internally length and content-type application/json is verified)
322 self.check_response_status(response["status"], status)
323 self.check_response_body(response["body"], bodyDict)
324 if headersDict: self.check_response_headers(response["headers"], headersDict)
328 self._connection.close()
331 # ADML Client simple fixture
332 @pytest.fixture(scope='session')
334 admlc = RestClient(ADML_ENDPOINT)
337 print("ADMLC Teardown")
339 @pytest.fixture(scope='session')
342 MyLogger.info("Gathering test suite resources ...")
343 for resource in glob.glob('resources/*'):
344 f = open(resource, "r")
345 name = os.path.basename(resource)
346 resourcesDict[name] = f.read()
349 def get_resources(key, **kwargs):
350 # Be careful with templates containing curly braces:
351 # https://stackoverflow.com/questions/5466451/how-can-i-print-literal-curly-brace-characters-in-python-string-and-also-use-fo
352 resource = resourcesDict[key]
355 args = defaultdict (str, kwargs)
356 resource = resource.format_map(args)
366 REQUEST_BODY_DIAMETER_HEX = '''
368 "diameterHex":"{diameterHex}"
371 REQUEST_BODY_NODE = {
377 (ADML_URI_PREFIX, '/decode', REQUEST_BODY_DIAMETER_HEX, ADML_ENDPOINT),
380 # Share RestClient connection for all the tests: session-scoped fixture
381 @pytest.fixture(scope="session", params=PARAMS)
382 def request_data(request):
383 admlc = RestClient(request.param[3])
384 def get_request_data(**kwargs):
385 args = defaultdict (str, kwargs)
386 uri_prefix = request.param[0]
387 request_uri_suffix=request.param[1]
388 formatted_uri=uri_prefix + request_uri_suffix.format_map(args)
389 request_body=request.param[2]
390 formatted_request_body=request_body.format_map(args)
391 return formatted_uri,formatted_request_body,admlc
393 yield get_request_data
395 print("RestClient Teardown")
397 # Fixture usage example: requestUrl,requestBody,admlc = request_data(diameterHex="<hex content>")