testing-scheduler/server/src/step/workload.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 import json\r
  11 import os\r
  12 from src.step.test_step import TestStep\r
  13 \r
  14 \r
  15 class WorkloadStep(TestStep):\r
  16     __step_type__ = 'workload'\r
  17 \r
  18     def __init__(self, id, name, service, action, args):\r
  19         super(WorkloadStep, self).__init__(\r
  20             self.__step_type__, id, name, service, action, args)\r
  21         self._argsParse()\r
  22         self._action()\r
  23 \r
  24     def _argsParse(self):\r
  25         if self._callType == "REST":\r
  26             currentDirPath = os.path.dirname(os.path.abspath(__file__))\r
  27             envDirPath = os.path.abspath(\r
  28                 os.path.join(currentDirPath, os.pardir, os.pardir, 'env'))\r
  29             envFilePath = os.path.join(\r
  30                 envDirPath, "%s.json" % self._service['name'])\r
  31             with open(envFilePath) as f:\r
  32                 propDict = json.load(f)\r
  33                 self._args['ip'] = propDict['ip']\r
  34                 self._args['port'] = propDict['port']\r
  35                 self._args['api'] = "%s/%s" % (\r
  36                     propDict['api_map']['workload'], self._args['command'])\r
  37                 exclude = {'ip', 'port', 'api', 'command', 'method'}\r
  38                 self._args['req_body'] = {\r
  39                     key: value for key, value in\r
  40                     self._args.items() if key not in exclude}\r
  41 \r
  42     def _start(self):\r
  43         print "workload start"\r
  44 \r
  45     def _stop(self):\r
  46         print "workload stop"\r