+++ /dev/null
-##############################################################################\r
-# Copyright (c) 2018 HUAWEI TECHNOLOGIES CO.,LTD and others.\r
-#\r
-# All rights reserved. This program and the accompanying materials\r
-# are made available under the terms of the Apache License, Version 2.0\r
-# which accompanies this distribution, and is available at\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-##############################################################################\r
-\r
-from src.step.test_step import TestStep\r
-import os\r
-import yaml\r
-import re\r
-\r
-\r
-class GeneralTestStep(TestStep):\r
- __step_type__ = "test"\r
-\r
- def __init__(self, id, name, service, action, args, context):\r
- super(GeneralTestStep, self).__init__(\r
- self.__step_type__, id, name, service, action, args, context)\r
- self._stepParse()\r
- self.action()\r
-\r
- def _contextTransform(self, argsDict):\r
- for (k, v) in argsDict.items():\r
- if isinstance(v, str):\r
- if re.match('^\(\(context\..*\)\)', v):\r
- v = v[10:-2]\r
- layers = v.split(".")\r
- contextData = self._context\r
- for layer in layers:\r
- contextData = contextData[layer]\r
- argsDict[k] = contextData\r
- elif isinstance(v, dict):\r
- self._contextTransform(v)\r
-\r
- def _stepParse(self):\r
- self._args_temp = self._args\r
- self._args = {}\r
-\r
- # transform the service config\r
- envFilePath = os.path.join(\r
- self._getCurrentDir(), "..", "env",\r
- "service", self._serviceName + ".yaml")\r
- requestParam = {}\r
- with open(envFilePath, 'r') as f:\r
- conf = yaml.load(f)\r
- conf = conf[self._serviceName]\r
- for apiItem in conf["apis"]:\r
- if apiItem['name'] == self._serviceInterface:\r
- interfaceConf = apiItem\r
- if interfaceConf is None:\r
- return\r
-\r
- # transform the args config\r
- self._contextTransform(self._args_temp)\r
-\r
- interfaceUri = interfaceConf['baseuri'] + \\r
- interfaceConf['template']['uri'][11:]\r
- interfaceUri = "http://%s:%s/%s" % (\r
- conf['ip'], conf['port'], interfaceUri)\r
- requestParam['uri'] = self._uriTransform(interfaceUri)\r
-\r
- requestParam['method'] = interfaceConf['method']\r
- if requestParam["method"] == "POST":\r
- requestParam['body'] = interfaceConf['template']['body']\r
- self._paramTransform(requestParam['body'], self._args_temp)\r
- self._args['http_request'] = requestParam\r
-\r
- def _uriTransform(self, uri):\r
- return re.sub("\(\(.*?\)\)", self._uriResReplace, uri)\r
-\r
- def _uriResReplace(self, match):\r
- matchTrim = match.group()[2:-2]\r
- return self._args_temp[matchTrim]\r
-\r
- def _paramTransform(self, argsTemplate, argsDict):\r
- for (k, v) in argsTemplate.items():\r
- if isinstance(v, str):\r
- if re.match('^\(\(.*\)\)', v):\r
- argsTemplate[k] = argsDict[v[2:-2]]\r
- elif isinstance(v, dict):\r
- self._paramTransform(v, argsDict)\r
-\r
- def start(self):\r
- pass\r