csit/libraries/IoTDM/ciotdm.py

   1 """This is the base library for criotdm. Both work for IoTDM project."""
   2
   3 import requests
   4
   5 op_provision = ":8181/restconf/operations/onem2m:onem2m-cse-provisioning"
   6 op_tree = ":8181/restconf/operational/onem2m:onem2m-resource-tree"
   7 op_cleanup = ":8181/restconf/operations/onem2m:onem2m-cleanup-store"
   8
   9 cse_payload = """
  10 {    "input": {
  11         "onem2m-primitive": [
  12            {
  13                 "name": "CSE_ID",
  14                 "value": "%s"
  15             },
  16             {
  17                 "name": "CSE_TYPE",
  18                 "value": "IN-CSE"
  19             }
  20         ]
  21     }
  22 }
  23 """
  24
  25 resourcepayload = """
  26 {
  27     %s
  28 }
  29 """
  30
  31 ae_payload = """
  32 {
  33     "m2m:ae":{%s}
  34 }
  35 """
  36
  37 con_payload = """
  38 {
  39     "m2m:cnt":{%s}
  40 }
  41 """
  42
  43 cin_payload = """
  44 {
  45    "m2m:cin":{%s}
  46 }
  47 """
  48
  49 sub_payload = """
  50 {
  51     "m2m:sub":{%s}
  52 }
  53 """
  54
  55 acp_payload = """
  56 {
  57     "m2m:acp":{%s}
  58 }
  59 """
  60
  61 nod_payload = """
  62 {
  63     "m2m:nod":{%s}
  64 }
  65 """
  66
  67 resources = {
  68     "m2m:ae",
  69     "m2m:cnt",
  70     "m2m:cin",
  71     "m2m:sub",
  72     "m2m:acp",
  73     "m2m:nod",
  74     "m2m:grp",
  75     "m2m:cb",
  76     "ch",
  77 }
  78
  79 payload_map = {
  80     1: acp_payload,
  81     2: ae_payload,
  82     3: con_payload,
  83     4: cin_payload,
  84     14: nod_payload,
  85     23: sub_payload,
  86 }
  87
  88
  89 def find_key(response, key, first=None):
  90     """Deserialize response, return value for key or None."""
  91     dic = response.json()
  92     key1 = list(dic.keys())
  93     if len(key1) != 1:
  94         raise ValueError("The response should be json object")
  95     if key1[0] not in resources:
  96         raise ValueError("The resource is not recognized")
  97     if first is True:
  98         return dic.get(key1[0], None)
  99     else:
 100         return dic.get(key1[0], None).get(key, None)
 101
 102
 103 def childResource(response):
 104     """Return child resources from the response."""
 105     return find_key(response, "ch")
 106
 107
 108 def childResourceFirst(response):
 109     """Return child resources from the response."""
 110     return find_key(response, "ch", True)
 111
 112
 113 def name(response):
 114     """Return the resource name in the response."""
 115     return find_key(response, "rn")
 116
 117
 118 def lastModifiedTime(response):
 119     """Return the lastModifiedTime in the response."""
 120     return find_key(response, "lt")
 121
 122
 123 def stateTag(response):
 124     """Return the state tag from the response."""
 125     return find_key(response, "st")
 126
 127
 128 def resid(response):
 129     """Return the resource id in the response."""
 130     return find_key(response, "ri")
 131
 132
 133 def parent(response):
 134     """Return the parent resource id in the response."""
 135     return find_key(response, "pi")
 136
 137
 138 def content(response):
 139     """Return the content the response."""
 140     return find_key(response, "con")
 141
 142
 143 def restype(response):
 144     """Return the resource type the response."""
 145     return find_key(response, "rty")
 146
 147
 148 def currentNumberOfInstances(response):
 149     """Return current number of instances from the response."""
 150     return find_key(response, "cni")
 151
 152
 153 def currentByteSize(response):
 154     """Return current byte size from the response."""
 155     return find_key(response, "cbs")
 156
 157
 158 def maxNumberOfInstances(response):
 159     """Return max number of instances from the response."""
 160     return find_key(response, "mni")
 161
 162
 163 def maxByteSize(response):
 164     """Return max byte size from the response."""
 165     return find_key(response, "mbs")
 166
 167
 168 def status(response):
 169     """Return the protocol status code in the response."""
 170     try:
 171         return response.status_code
 172     except (TypeError, AttributeError):
 173         return None
 174
 175
 176 def headers(response):
 177     """Return the protocol headers in the response."""
 178     try:
 179         return response.headers
 180     except (TypeError, AttributeError):
 181         return None
 182
 183
 184 def error(response):
 185     """Return the error string in the response."""
 186     try:
 187         return response.json()["error"]
 188     except (TypeError, AttributeError):
 189         return None
 190
 191
 192 def normalize(resource_uri):
 193     """Remove the first / of /InCSE1/ae1."""
 194     if resource_uri is not None:
 195         if resource_uri[0] == "/":
 196             return resource_uri[1:]
 197     return resource_uri
 198
 199
 200 class connect:
 201
 202     """Create the connection."""
 203
 204     def __init__(
 205         self,
 206         server="localhost",
 207         base="InCSE1",
 208         auth=("admin", "admin"),
 209         protocol="http",
 210     ):
 211         """Connect to a IoTDM server."""
 212         self.session = requests.Session()
 213         self.session.auth = auth
 214         self.session.headers.update({"content-type": "application/json"})
 215         self.timeout = 5
 216         self.payload = cse_payload % (base)
 217         self.headers = {
 218             # Admittedly these are "magic values" but are required
 219             # and until a proper defaulting initializer is in place
 220             # are hard-coded.
 221             "content-type": "application/vnd.onem2m-res+json",
 222             "X-M2M-Origin": "iotdm-robot-tests",
 223             "X-M2M-RI": "12345",
 224             "X-M2M-OT": "NOW",
 225         }
 226         self.server = "%s://" % (protocol) + server
 227         self.url = self.server + op_provision
 228         self.response = self.session.post(
 229             self.url, data=self.payload, timeout=self.timeout
 230         )
 231
 232     def modify_headers_origin(self, new_origin):
 233         """Modify the headers to test ACP."""
 234         self.headers["X-M2M-Origin"] = new_origin
 235
 236     def create(self, parent, restype, attr=None):
 237         """Create certain resource with attributes under parent URI.
 238
 239         Args:
 240             :param parent: the target URI
 241             :param restype: the resourceType of the resource
 242             :param attr: the payload of the resource
 243         """
 244         if parent is None:
 245             return None
 246         restype = int(restype)
 247         payload = payload_map[restype]
 248         payload = payload % (attr)
 249         self.headers["content-type"] = (
 250             "application/\
 251             vnd.onem2m-res+json;ty=%s"
 252             % (restype)
 253         )
 254         parent = normalize(parent)
 255         self.url = self.server + ":8282/%s?&rcn=1" % (parent)
 256         self.response = self.session.post(
 257             self.url, payload, timeout=self.timeout, headers=self.headers
 258         )
 259
 260     def create_with_command(self, parent, restype, command, attr=None):
 261         """Create certain resource with attributes under parent URI.
 262
 263         Args:
 264             :param parent: the target URI
 265             :param restype: the resourceType of the resource
 266             :param command: the command would be in the URI after &
 267             :param attr: the payload of the resource
 268         """
 269         if parent is None:
 270             return None
 271         restype = int(restype)
 272         payload = payload_map[restype]
 273         payload = payload % (attr)
 274         self.headers["content-type"] = (
 275             "application/\
 276             vnd.onem2m-res+json;ty=%s"
 277             % (restype)
 278         )
 279         parent = normalize(parent)
 280         self.url = self.server + ":8282/%s?%s" % (parent, command)
 281         self.response = self.session.post(
 282             self.url, payload, timeout=self.timeout, headers=self.headers
 283         )
 284
 285     def retrieve(self, resource_uri):
 286         """Retrieve resource using resource_uri."""
 287         if resource_uri is None:
 288             return None
 289         resource_uri = normalize(resource_uri)
 290         self.url = self.server + ":8282/%s?rcn=5" % (resource_uri)
 291         self.headers["X-M2M-NM"] = None
 292         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 293         self.response = self.session.get(
 294             self.url, timeout=self.timeout, headers=self.headers
 295         )
 296
 297     def retrieve_with_command(self, resource_uri, command):
 298         """Retrieve resource using resource_uri with command."""
 299         if resource_uri is None:
 300             return None
 301         if command is None:
 302             return None
 303         resource_uri = normalize(resource_uri)
 304         self.url = self.server + ":8282/%s?%s" % (resource_uri, command)
 305         self.headers["X-M2M-NM"] = None
 306         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 307         self.response = self.session.get(
 308             self.url, timeout=self.timeout, headers=self.headers
 309         )
 310
 311     def update(self, resource_uri, restype, attr=None):
 312         """Update resource at resource_uri with new attributes."""
 313         if resource_uri is None:
 314             return None
 315         resource_uri = normalize(resource_uri)
 316         restype = int(restype)
 317         payload = payload_map[restype]
 318         payload = payload % (attr)
 319         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 320         self.url = self.server + ":8282/%s" % (resource_uri)
 321         self.response = self.session.put(
 322             self.url, payload, timeout=self.timeout, headers=self.headers
 323         )
 324
 325     def update_with_command(self, resource_uri, restype, command, attr=None):
 326         """Update resource at resource_uri with new attributes."""
 327         if resource_uri is None:
 328             return None
 329         resource_uri = normalize(resource_uri)
 330         restype = int(restype)
 331         payload = payload_map[restype]
 332         payload = payload % (attr)
 333         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 334         self.url = self.server + ":8282/%s?%s" % (resource_uri, command)
 335         self.response = self.session.put(
 336             self.url, payload, timeout=self.timeout, headers=self.headers
 337         )
 338
 339     def delete(self, resource_uri):
 340         """Delete the resource at the resource_uri."""
 341         if resource_uri is None:
 342             return None
 343         resource_uri = normalize(resource_uri)
 344         self.url = self.server + ":8282/%s" % (resource_uri)
 345         self.headers["X-M2M-NM"] = None
 346         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 347         self.response = self.session.delete(
 348             self.url, timeout=self.timeout, headers=self.headers
 349         )
 350
 351     def delete_with_command(self, resource_uri, command):
 352         """Delete the resource at the resource_uri."""
 353         if resource_uri is None:
 354             return None
 355         resource_uri = normalize(resource_uri)
 356         self.url = self.server + ":8282/%s?%s" % (resource_uri, command)
 357         self.headers["X-M2M-NM"] = None
 358         self.headers["content-type"] = "application/vnd.onem2m-res+json"
 359         self.response = self.session.delete(
 360             self.url, timeout=self.timeout, headers=self.headers
 361         )
 362
 363     def tree(self):
 364         """Get the resource tree."""
 365         self.url = self.server + op_tree
 366         self.response = self.session.get(self.url)
 367
 368     def kill(self):
 369         """Kill the tree."""
 370         self.url = self.server + op_cleanup
 371         self.response = self.session.post(self.url)