1 """This is the base library for criotdm. Both work for IoTDM project."""
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"
89 def find_key(response, key, first=None):
90 """Deserialize response, return value for key or None."""
92 key1 = list(dic.keys())
94 raise ValueError("The response should be json object")
95 if key1[0] not in resources:
96 raise ValueError("The resource is not recognized")
98 return dic.get(key1[0], None)
100 return dic.get(key1[0], None).get(key, None)
103 def childResource(response):
104 """Return child resources from the response."""
105 return find_key(response, "ch")
108 def childResourceFirst(response):
109 """Return child resources from the response."""
110 return find_key(response, "ch", True)
114 """Return the resource name in the response."""
115 return find_key(response, "rn")
118 def lastModifiedTime(response):
119 """Return the lastModifiedTime in the response."""
120 return find_key(response, "lt")
123 def stateTag(response):
124 """Return the state tag from the response."""
125 return find_key(response, "st")
129 """Return the resource id in the response."""
130 return find_key(response, "ri")
133 def parent(response):
134 """Return the parent resource id in the response."""
135 return find_key(response, "pi")
138 def content(response):
139 """Return the content the response."""
140 return find_key(response, "con")
143 def restype(response):
144 """Return the resource type the response."""
145 return find_key(response, "rty")
148 def currentNumberOfInstances(response):
149 """Return current number of instances from the response."""
150 return find_key(response, "cni")
153 def currentByteSize(response):
154 """Return current byte size from the response."""
155 return find_key(response, "cbs")
158 def maxNumberOfInstances(response):
159 """Return max number of instances from the response."""
160 return find_key(response, "mni")
163 def maxByteSize(response):
164 """Return max byte size from the response."""
165 return find_key(response, "mbs")
168 def status(response):
169 """Return the protocol status code in the response."""
171 return response.status_code
172 except (TypeError, AttributeError):
176 def headers(response):
177 """Return the protocol headers in the response."""
179 return response.headers
180 except (TypeError, AttributeError):
185 """Return the error string in the response."""
187 return response.json()["error"]
188 except (TypeError, AttributeError):
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:]
202 """Create the connection."""
208 auth=("admin", "admin"),
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"})
216 self.payload = cse_payload % (base)
218 # Admittedly these are "magic values" but are required
219 # and until a proper defaulting initializer is in place
221 "content-type": "application/vnd.onem2m-res+json",
222 "X-M2M-Origin": "iotdm-robot-tests",
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
232 def modify_headers_origin(self, new_origin):
233 """Modify the headers to test ACP."""
234 self.headers["X-M2M-Origin"] = new_origin
236 def create(self, parent, restype, attr=None):
237 """Create certain resource with attributes under parent URI.
240 :param parent: the target URI
241 :param restype: the resourceType of the resource
242 :param attr: the payload of the resource
246 restype = int(restype)
247 payload = payload_map[restype]
248 payload = payload % (attr)
249 self.headers["content-type"] = (
251 vnd.onem2m-res+json;ty=%s"
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
260 def create_with_command(self, parent, restype, command, attr=None):
261 """Create certain resource with attributes under parent URI.
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
271 restype = int(restype)
272 payload = payload_map[restype]
273 payload = payload % (attr)
274 self.headers["content-type"] = (
276 vnd.onem2m-res+json;ty=%s"
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
285 def retrieve(self, resource_uri):
286 """Retrieve resource using resource_uri."""
287 if resource_uri is 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
297 def retrieve_with_command(self, resource_uri, command):
298 """Retrieve resource using resource_uri with command."""
299 if resource_uri is 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
311 def update(self, resource_uri, restype, attr=None):
312 """Update resource at resource_uri with new attributes."""
313 if resource_uri is 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
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:
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
339 def delete(self, resource_uri):
340 """Delete the resource at the resource_uri."""
341 if resource_uri is 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
351 def delete_with_command(self, resource_uri, command):
352 """Delete the resource at the resource_uri."""
353 if resource_uri is 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
364 """Get the resource tree."""
365 self.url = self.server + op_tree
366 self.response = self.session.get(self.url)
370 self.url = self.server + op_cleanup
371 self.response = self.session.post(self.url)