1 ##############################################################################
2 # Copyright (c) 2018 HUAWEI TECHNOLOGIES CO.,LTD and others.
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 ##############################################################################
10 from src.step.test_step import TestStep
16 class GeneralTestStep(TestStep):
17 __step_type__ = "test"
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)
25 def _contextTransform(self, argsDict):
26 for (k, v) in argsDict.items():
27 if isinstance(v, str):
28 if re.match('^\(\(context\..*\)\)', v):
31 contextData = self._context
33 contextData = contextData[layer]
34 argsDict[k] = contextData
35 elif isinstance(v, dict):
36 self._contextTransform(v)
39 self._args_temp = self._args
42 # transform the service config
43 envFilePath = os.path.join(
44 self._getCurrentDir(), "..", "env",
45 "service", self._serviceName + ".yaml")
47 with open(envFilePath, 'r') as 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:
56 # transform the args config
57 self._contextTransform(self._args_temp)
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)
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
71 def _uriTransform(self, uri):
72 return re.sub("\(\(.*?\)\)", self._uriResReplace, uri)
74 def _uriResReplace(self, match):
75 matchTrim = match.group()[2:-2]
76 return self._args_temp[matchTrim]
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)