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