test/tools/CSIT_Test/base/testmodule.py

   1 """
   2 CSIT test tools.
   3 Authors: Baohua Yang@IBM, Denghui Huang@IBM
   4 Updated: 2013-10-30
   5 """
   6
   7 from restlib import *  # noqa
   8
   9
  10 class TestModule(object):
  11     """
  12     Basic module class for test restful APIS.
  13     Support the standard Create, Read, Update, Delete (CRUD) actions.
  14     """
  15
  16     def __init__(self, restSubContext, user=DEFAULT_USER, password=DEFAULT_PWD, container=DEFAULT_CONTAINER,
  17                  contentType='json', prefix=DEFAULT_PREFIX):
  18         self.restSubContext = restSubContext
  19         self.container = container
  20         self.user = user
  21         self.password = password
  22         self.contentType = contentType
  23         self.prefix = prefix
  24
  25     def extract_properties(self, content, key, property):
  26         """
  27         Return all nodes.
  28         """
  29         if not isinstance(content, dict) or key not in content:
  30             return None
  31         else:
  32             return [e.get(property) for e in content[key]]
  33
  34     def get_entries(self, suffix=None, key=None):
  35         """
  36         Get the existed entries in the service.
  37         """
  38         if isinstance(suffix, list) and key:
  39             result = {}
  40             result[key] = []
  41             for s in suffix:
  42                 result[key].extend(self.get_entries(s).get(key))
  43             return result
  44         elif isinstance(suffix, str):
  45             return self.read(suffix)
  46         elif not suffix:
  47             return self.read()
  48         else:
  49             return None
  50
  51     def add_entry(self, suffix, name, body):
  52         """
  53         Add entry to the service.
  54         """
  55         self.update(suffix + '/' + name, body)
  56
  57     def remove_entry(self, suffix, name):
  58         """
  59         Remove entry from the service.
  60         """
  61         self.delete(suffix + '/' + name)
  62
  63     def test_add_remove_operations(self, suffix_entries, suffix_entry, name, body, key):
  64         result = []
  65         # Add an entry
  66         self.add_entry(suffix_entry, name, body)
  67         r = self.get_entries(suffix_entries, key)
  68         if r:
  69             v = r.get(key)
  70             result.append(body in v if v else False)
  71             # Remove the added entry
  72         if result == [True]:
  73             self.remove_entry(suffix_entry, name)
  74             r = self.get_entries(suffix_entries, key)
  75             v = r.get(key)
  76             result.append(body not in v if v else True)
  77         return result == [True, True]
  78
  79     def create(self, suffix, body=None):
  80         """
  81         POST to given suffix url.
  82         TODO: complete
  83         """
  84         url = self.prefix + self.restSubContext
  85         if self.container:
  86             url += '/' + self.container
  87         if suffix:
  88             url += '/' + suffix
  89         return do_post_request(url, self.contentType, body, self.user, self.password)
  90
  91     def read(self, suffix=None):
  92         """
  93         GET from given suffix url.
  94         """
  95         url = self.prefix + self.restSubContext
  96         if self.container:
  97             url += '/' + self.container
  98         if suffix:
  99             url += '/' + suffix
 100         return do_get_request_with_response_content(url, self.contentType, self.user, self.password)
 101
 102     def update(self, suffix, body=None):
 103         """
 104         PUT to given suffix url.
 105         """
 106         url = self.prefix + self.restSubContext
 107         if self.container:
 108             url += '/' + self.container
 109         if suffix:
 110             url += '/' + suffix
 111         return do_put_request(url, self.contentType, body, self.user, self.password)
 112
 113     def delete(self, suffix):
 114         """
 115         DELETE to given suffix url.
 116         TODO: complete
 117         """
 118         url = self.prefix + self.restSubContext
 119         if self.container:
 120             url += '/' + self.container
 121         if suffix:
 122             url += '/' + suffix
 123         return do_delete_request(url, self.user, self.password)