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

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