test-scheduler/server/src/step/general_test_step.py

   1 ##############################################################################
   2 # Copyright (c) 2018 HUAWEI TECHNOLOGIES CO.,LTD and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 from src.step.test_step import TestStep
  11 import os
  12 import yaml
  13 import re
  14
  15
  16 class GeneralTestStep(TestStep):
  17     __step_type__ = "test"
  18
  19     def __init__(self, id, name, service, action, args, context):
  20         super(GeneralTestStep, self).__init__(
  21             self.__step_type__, id, name, service, action, args, context)
  22         self._stepParse()
  23         self.action()
  24
  25     def _contextTransform(self, argsDict):
  26         for (k, v) in argsDict.items():
  27             if isinstance(v, str):
  28                 if re.match('^\(\(context\..*\)\)', v):
  29                     v = v[10:-2]
  30                     layers = v.split(".")
  31                     contextData = self._context
  32                     for layer in layers:
  33                         contextData = contextData[layer]
  34                     argsDict[k] = contextData
  35             elif isinstance(v, dict):
  36                 self._contextTransform(v)
  37
  38     def _stepParse(self):
  39         self._args_temp = self._args
  40         self._args = {}
  41
  42         # transform the service config
  43         envFilePath = os.path.join(
  44             self._getCurrentDir(), "..", "env",
  45             "service", self._serviceName + ".yaml")
  46         requestParam = {}
  47         with open(envFilePath, 'r') as f:
  48             conf = yaml.load(f)
  49             conf = conf[self._serviceName]
  50         for apiItem in conf["apis"]:
  51             if apiItem['name'] == self._serviceInterface:
  52                 interfaceConf = apiItem
  53         if interfaceConf is None:
  54             return
  55
  56         # transform the args config
  57         self._contextTransform(self._args_temp)
  58
  59         interfaceUri = interfaceConf['baseuri'] +  \
  60             interfaceConf['template']['uri'][11:]
  61         interfaceUri = "http://%s:%s/%s" % (
  62             conf['ip'], conf['port'], interfaceUri)
  63         requestParam['uri'] = self._uriTransform(interfaceUri)
  64
  65         requestParam['method'] = interfaceConf['method']
  66         if requestParam["method"] == "POST":
  67             requestParam['body'] = interfaceConf['template']['body']
  68             self._paramTransform(requestParam['body'], self._args_temp)
  69         self._args['http_request'] = requestParam
  70
  71     def _uriTransform(self, uri):
  72         return re.sub("\(\(.*?\)\)", self._uriResReplace, uri)
  73
  74     def _uriResReplace(self, match):
  75         matchTrim = match.group()[2:-2]
  76         return self._args_temp[matchTrim]
  77
  78     def _paramTransform(self, argsTemplate, argsDict):
  79         for (k, v) in argsTemplate.items():
  80             if isinstance(v, str):
  81                 if re.match('^\(\(.*\)\)', v):
  82                     argsTemplate[k] = argsDict[v[2:-2]]
  83             elif isinstance(v, dict):
  84                 self._paramTransform(v, argsDict)
  85
  86     def start(self):
  87         pass