1 ##############################################################################
\r
2 # Copyright (c) 2018 HUAWEI TECHNOLOGIES CO.,LTD and others.
\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
10 from src.step.test_step import TestStep
\r
16 class GeneralTestStep(TestStep):
\r
17 __step_type__ = "test"
\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
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
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
38 def _stepParse(self):
\r
39 self._args_temp = self._args
\r
42 # transform the service config
\r
43 envFilePath = os.path.join(
\r
44 self._getCurrentDir(), "..", "env",
\r
45 "service", self._serviceName + ".yaml")
\r
47 with open(envFilePath, 'r') as 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
56 # transform the args config
\r
57 self._contextTransform(self._args_temp)
\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
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
71 def _uriTransform(self, uri):
\r
72 return re.sub("\(\(.*?\)\)", self._uriResReplace, uri)
\r
74 def _uriResReplace(self, match):
\r
75 matchTrim = match.group()[2:-2]
\r
76 return self._args_temp[matchTrim]
\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