example/diameter/launcher/resources/rest_api/ct/conftest.py

   1 # Keep sorted
   2 import base64
   3 from collections import defaultdict
   4 import glob
   5 from hyper import HTTP20Connection
   6 #import inspect
   7 import json
   8 import logging
   9 import os
  10 import pytest
  11 import re
  12
  13 #############
  14 # CONSTANTS #
  15 #############
  16
  17 # Endpoint
  18 ADML_HOST = 'localhost'
  19 ADML_PORT = '8074'
  20 ADML_ENDPOINT = ADML_HOST + ':' + ADML_PORT
  21 ADML_URI_PREFIX = ''
  22
  23 # Headers
  24 CONTENT_LENGTH = 'content-length'
  25
  26 # Flow calculation throw admlf (ADML Flow) fixture:
  27 #   flow = admlf.getId()
  28 #
  29 # For sequenced tests, the id is just a monotonically increased number from 1.
  30 # For parallel tests, this id is sequenced from 1 for every worker, then, globally,
  31 #  this is a handicap to manage flow ids (tests ids) for ADML FSM (finite state machine).
  32 # We consider a base multiplier of 10000, so collisions would take place when workers
  33 #  reserves more than 10000 flows. With a 'worst-case' assumption of `5 flows per test case`,
  34 #  you should cover up to 5000 per worker. Anyway, feel free to increase this value,
  35 #  specially if you are thinking in use pytest and ADML Agent for system tests.
  36 FLOW_BASE_MULTIPLIER = 10000
  37
  38 #########
  39 # HOOKS #
  40 #########
  41 #def pytest_runtest_setup(item):
  42 #    print("pytest_runtest_setup")
  43 #def pytest_runtest_logreport(report):
  44 #    print(f'Log Report:{report}')
  45 #def pytest_sessionstart(session):
  46 #    print("pytest_session start")
  47 #def pytest_sessionfinish(session):
  48 #    print("pytest_session finish")
  49
  50 ######################
  51 # CLASSES & FIXTURES #
  52 ######################
  53
  54 class Sequencer(object):
  55     def __init__(self, request):
  56         self.sequence = 0
  57         self.request = request
  58
  59     def __wid(self):
  60         """
  61         Returns the worker id, or 'master' if not parallel execution is done
  62         """
  63         wid = 'master'
  64         if hasattr(self.request.config, 'slaveinput'):
  65           wid = self.request.config.slaveinput['slaveid']
  66
  67         return wid
  68
  69     def getId(self):
  70         """
  71         Returns the next identifier value (monotonically increased in every call)
  72         """
  73         self.sequence += 1
  74
  75         wid = self.__wid()
  76         if wid == "master":
  77           return self.sequence
  78
  79         # Workers are named: wd0, wd1, wd2, etc.
  80         wid_number = int(re.findall(r'\d+', wid)[0])
  81
  82         return FLOW_BASE_MULTIPLIER * wid_number + self.sequence
  83
  84
  85
  86 @pytest.fixture(scope='session')
  87 def admlf(request):
  88   """
  89   ADML Flow
  90   """
  91   return Sequencer(request)
  92
  93
  94
  95 # Logging
  96 class MyLogger:
  97
  98   # CRITICAL ERROR WARNING INFO DEBUG NOSET
  99   def setLevelInfo(): logging.getLogger().setLevel(logging.INFO)
 100   def setLevelDebug(): logging.getLogger().setLevel(logging.DEBUG)
 101
 102   def error(message): logging.getLogger().error(message)
 103   def warning(message): logging.getLogger().warning(message)
 104   def info(message): logging.getLogger().info(message)
 105   def debug(message): logging.getLogger().debug(message)
 106
 107 @pytest.fixture(scope='session')
 108 def mylogger():
 109   return MyLogger
 110
 111 MyLogger.logger = logging.getLogger('CT')
 112
 113 # Base64 encoding:
 114 @pytest.fixture(scope='session')
 115 def b64_encode():
 116   def encode(message):
 117     message_bytes = message.encode('ascii')
 118     base64_bytes = base64.b64encode(message_bytes)
 119     return base64_bytes.decode('ascii')
 120   return encode
 121
 122 # Base64 decoding:
 123 @pytest.fixture(scope='session')
 124 def b64_decode():
 125   def decode(base64_message):
 126     base64_bytes = base64_message.encode('ascii')
 127     message_bytes = base64.b64decode(base64_bytes)
 128     return message_bytes.decode('ascii')
 129   return decode
 130
 131 # HTTP communication:
 132 class RestClient(object):
 133     """A client helper to perform rest operations: GET, POST.
 134
 135     Attributes:
 136         endpoint: server endpoint to make the HTTP2.0 connection
 137     """
 138
 139     def __init__(self, endpoint):
 140         """Return a RestClient object for ADML endpoint."""
 141         self._endpoint = endpoint
 142         self._ip = self._endpoint.split(':')[0]
 143         self._connection = HTTP20Connection(host=self._endpoint)
 144
 145     def _log_http(self, kind, method, url, body, headers):
 146         length = len(body) if body else 0
 147         MyLogger.info(
 148                 '{} {}{} {} headers: {!s} data: {}:{!a}'.format(
 149                 method, self._endpoint, url, kind, headers, length, body))
 150
 151     def _log_request(self, method, url, body, headers):
 152         self._log_http('REQUEST', method, url, body, headers)
 153
 154     def _log_response(self, method, url, response):
 155         self._log_http(
 156                 'RESPONSE:{}'.format(response["status"]), method, url,
 157                 response["body"], response["headers"])
 158
 159     #def log_event(self, level, log_msg):
 160     #    # Log caller function name and formated message
 161     #    MyLogger.logger.log(level, inspect.getouterframes(inspect.currentframe())[1].function + ': {!a}'.format(log_msg))
 162
 163     def parse(self, response):
 164         response_body = response.read(decode_content=True).decode('utf-8')
 165         if len(response_body) != 0:
 166           response_body_dict = json.loads(response_body)
 167         else:
 168           response_body_dict = ''
 169         response_data = { "status":response.status, "body":response_body_dict, "headers":response.headers }
 170         return response_data
 171
 172     def request(self, requestMethod, requestUrl, requestBody=None, requestHeaders=None):
 173       """
 174       Returns response data dictionary with 'status', 'body' and 'headers'
 175       """
 176       requestBody = RestClient._pad_body_and_length(requestBody, requestHeaders)
 177       self._log_request(requestMethod, requestUrl, requestBody, requestHeaders)
 178       self._connection.request(method=requestMethod, url=requestUrl, body=requestBody, headers=requestHeaders)
 179       response = self.parse(self._connection.get_response())
 180       self._log_response(requestMethod, requestUrl, response)
 181       return response
 182
 183     def _pad_body_and_length(requestBody, requestHeaders):
 184         """Pad the body and adjust content-length if needed.
 185         When the length of the body is multiple of 1024 this function appends
 186         one space to the body and increases by one the content-length.
 187
 188         This is a workaround for hyper issue 355 [0].
 189         The issue has been fixed but it has not been released yet.
 190
 191         [0]: https://github.com/Lukasa/hyper/issues/355
 192
 193         EXAMPLE
 194         >>> body, headers = ' '*1024, { 'content-length':'41' }
 195         >>> body = RestClient._pad_body_and_length(body, headers)
 196         >>> ( len(body), headers['content-length'] )
 197         (1025, '42')
 198         """
 199         if requestBody and 0 == (len(requestBody) % 1024):
 200             logging.warning( "RestClient.request:" +
 201                              " padding body because" +
 202                              " its length ({})".format(len(requestBody)) +
 203                              " is multiple of 1024")
 204             requestBody += " "
 205             content_length = CONTENT_LENGTH
 206             if requestHeaders and content_length in requestHeaders:
 207                 length = int(requestHeaders[content_length])
 208                 requestHeaders[content_length] = str(length+1)
 209         return requestBody
 210
 211     def get(self, requestUrl):
 212         return self.request('GET', requestUrl)
 213
 214     def post(self, requestUrl, requestBody = None, requestHeaders={'content-type': 'application/json'}):
 215         return self.request('POST', requestUrl, requestBody, requestHeaders)
 216
 217     def postDict(self, requestUrl, requestBody = None, requestHeaders={'content-type': 'application/json'}):
 218         """
 219            Accepts request body as python dictionary
 220         """
 221         requestBodyJson = None
 222         if requestBody: requestBodyJson = json.dumps(requestBody, indent=None, separators=(',', ':'))
 223         return self.request('POST', requestUrl, requestBodyJson, requestHeaders)
 224
 225
 226     #def delete(self, requestUrl):
 227     #  return self.request('DELETE', requestUrl)
 228
 229     def __assert_received_expected(self, received, expected, what):
 230         match = (received == expected)
 231         log = "Received {what}: {received} | Expected {what}: {expected}".format(received=received, expected=expected, what=what)
 232         if match: MyLogger.info(log)
 233         else: MyLogger.error(log)
 234
 235         assert match
 236
 237     def check_response_status(self, received, expected, **kwargs):
 238         """
 239         received: status code received (got from response data parsed, field 'status')
 240         expected: status code expected
 241         """
 242         self.__assert_received_expected(received, expected, "status code")
 243
 244     #def check_expected_cause(self, response, **kwargs):
 245     #    """
 246     #    received: response data parsed where field 'body'->'cause' is analyzed
 247     #    kwargs: aditional regexp to match expected cause
 248     #    """
 249     #    if "expected_cause" in kwargs:
 250     #        received_content = response["body"]
 251     #        received_cause = received_content.get("cause", "")
 252     #        regular_expr_cause = kwargs["expected_cause"]
 253     #        regular_expr_flag = kwargs.get("regular_expression_flag", 0)
 254     #        matchObj = re.match(regular_expr_cause, received_cause, regular_expr_flag)
 255     #        log = 'Test error cause: "{}"~=/{}/.'.format(received_cause, regular_expr_cause)
 256     #        if matchObj: MyLogger.info(log)
 257     #        else: MyLogger.error(log)
 258     #
 259     #        assert matchObj is not None
 260
 261     def check_response_body(self, received, expected, inputJsonString = False):
 262         """
 263         received: body content received (got from response data parsed, field 'body')
 264         expected: body content expected
 265         inputJsonString: input parameters as json string (default are python dictionaries)
 266         """
 267         if inputJsonString:
 268           # Decode json:
 269           received = json.loads(received)
 270           expected = json.loads(expected)
 271
 272         self.__assert_received_expected(received, expected, "body")
 273
 274     def check_response_headers(self, received, expected):
 275         """
 276         received: headers received (got from response data parsed, field 'headers')
 277         expected: headers expected
 278         """
 279         self.__assert_received_expected(received, expected, "headers")
 280
 281     def assert_response__status_body_headers(self, response, status, bodyDict, headersDict = None):
 282         """
 283         response: Response parsed data
 284         status: numeric status code
 285         body: body dictionary to match with
 286         headers: headers dictionary to match with (by default, not checked: internally length and content-type application/json is verified)
 287         """
 288         self.check_response_status(response["status"], status)
 289         self.check_response_body(response["body"], bodyDict)
 290         if headersDict: self.check_response_headers(response["headers"], headersDict)
 291
 292
 293     def close(self):
 294       self._connection.close()
 295
 296
 297 # ADML Client simple fixture
 298 @pytest.fixture(scope='session')
 299 def admlc():
 300   admlc = RestClient(ADML_ENDPOINT)
 301   yield admlc
 302   admlc.close()
 303   print("ADMLC Teardown")
 304
 305 @pytest.fixture(scope='session')
 306 def resources():
 307   resourcesDict={}
 308   MyLogger.info("Gathering test suite resources ...")
 309   for resource in glob.glob('resources/*'):
 310     f = open(resource, "r")
 311     name = os.path.basename(resource)
 312     resourcesDict[name] = f.read()
 313     f.close()
 314
 315   def get_resources(key, **kwargs):
 316     # Be careful with templates containing curly braces:
 317     # https://stackoverflow.com/questions/5466451/how-can-i-print-literal-curly-brace-characters-in-python-string-and-also-use-fo
 318     resource = resourcesDict[key]
 319
 320     if kwargs:
 321       args = defaultdict (str, kwargs)
 322       resource = resource.format_map(args)
 323
 324     return resource
 325
 326   yield get_resources
 327
 328 ################
 329 # Experimental #
 330 ################
 331
 332 REQUEST_BODY_DIAMETER_HEX = '''
 333 {{
 334    "diameterHex":"{diameterHex}"
 335 }}'''
 336
 337 REQUEST_BODY_NODE = {
 338     "name":"{name}"
 339 }
 340
 341
 342 PARAMS = [
 343   (ADML_URI_PREFIX, '/decode', REQUEST_BODY_DIAMETER_HEX, ADML_ENDPOINT),
 344 ]
 345
 346 # Share RestClient connection for all the tests: session-scoped fixture
 347 @pytest.fixture(scope="session", params=PARAMS)
 348 def request_data(request):
 349   admlc = RestClient(request.param[3])
 350   def get_request_data(**kwargs):
 351     args = defaultdict (str, kwargs)
 352     uri_prefix = request.param[0]
 353     request_uri_suffix=request.param[1]
 354     formatted_uri=uri_prefix + request_uri_suffix.format_map(args)
 355     request_body=request.param[2]
 356     formatted_request_body=request_body.format_map(args)
 357     return formatted_uri,formatted_request_body,admlc
 358
 359   yield get_request_data
 360   admlc.close()
 361   print("RestClient Teardown")
 362
 363 # Fixture usage example: requestUrl,requestBody,admlc = request_data(diameterHex="<hex content>")
 364