import os
import logging
from behave.runner import Context as BehaveContext
from .api import Api, Request
[docs]
class Context():
def __init__(self):
[docs]
self.output_dir: str = ""
[docs]
self.logger: logging.Logger = None
[docs]
self.apis: dict[str, Api] = {}
from ..cloud import Cloud
[docs]
self.cloud: Cloud = None
[docs]
self.request: Request = None
[docs]
self.response = None
[docs]
class CloudContextHandler():
def __init__(self, context: BehaveContext):
[docs]
self.context = context
Context.__init__(self.context)
[docs]
def configure_logger(self, output_dir: str = ''):
self.context.logger = logging.getLogger()
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
self.context.logger.addHandler(console_handler)
if output_dir and os.path.exists(output_dir):
file_handle = logging.FileHandler(os.path.join(output_dir, 'test_driver.log'), mode='w', encoding='utf-8')
file_handle.setFormatter(logging.Formatter(fmt='%(asctime)s %(message)s',
datefmt='%d/%m/%Y %I:%M:%S'))
file_handle.setLevel(logging.INFO)
self.context.logger.addHandler(file_handle)
self.context.logger.setLevel(logging.INFO)
[docs]
def initialize(self, config: dict, output_dir: str, api_file: str, users_file: str):
self.context.in_scenario = False
self.context.output_dir = output_dir
self.configure_logger(output_dir)
self.context.logger.info("Initializing test driver ...")
self.load_config(config)
self.load_apis(api_file)
from ..cloud import Cloud
self.context.cloud = Cloud(users_file, config, self.context)
self.context.cloud.initialise(self.context)
[docs]
def load_config(self, config: dict):
self.context.timeout = config.get('REQUEST_TIMEOUT', 10)
self.context.users_nb_to_create = config.get('USERS_NB_TO_CREATE')
self.context.light_version = config.get('LIGHT_VERSION', False)
self.context.patch_open_api = config.get('PATCH_OPEN_API', False)
[docs]
def load_apis(self, api_file):
with open(api_file, 'r') as api_csv:
headers = api_csv.readline().strip()
headers = {key: i for i, key in enumerate(headers.split(';'))}
for line in api_csv.readlines():
line = line.strip().split(';')
api_type = line[headers['type']]
api = Api(api_type,
line[headers['api_url']],
line[headers['swagger_file']])
has_been_load = False
for other_api in self.context.apis.values():
if api.swagger_file == other_api.swagger_file:
self.context.apis[api_type] = other_api
has_been_load = True
break
if not has_been_load:
api.load_swagger(self.context.patch_open_api, self.context.logger)
self.context.apis[api_type] = api