+++ /dev/null
-import requests
-from datetime import timedelta
-
-zero = timedelta(0)
-
-application = 2
-container = 3
-contentInstance = 4
-
-op_provision = ":8181/restconf/operations/onem2m:onem2m-cse-provisioning"
-op_tree = ":8181/restconf/operational/onem2m:onem2m-resource-tree"
-op_cleanup = ":8181/restconf/operations/onem2m:onem2m-cleanup-store"
-
-cse_payload = '''
-{ "input": {
- "onem2m-primitive": [
- {
- "name": "CSE_ID",
- "value": "%s"
- },
- {
- "name": "CSE_TYPE",
- "value": "IN-CSE"
- }
- ]
- }
-}
-'''
-
-application_payload = '''
-{
- any:
- [
- {"aei":"jb", "api":"jb", "apn":"jb2", "or":"http://hey/you" %s}
- ]
-}
-'''
-
-_container_payload = '''
-{
- any:
- [
- {
- "cr": "jb",
- "mni": 1,
- "mbs": 3,
- "or": "http://hey/you"
- %s
- }
- ]
-}
-'''
-
-container_payload = '''
-{
- any:
- [
- {
- "cr": "jb",
- "or": "http://hey/you"
- %s
- }
- ]
-}
-'''
-
-contentInstance_payload = '''
-{
- "any": [
- {
- "cnf": "1",
- "or": "http://hey/you"
- %s
- }
- ]
-}
-'''
-
-
-def which_payload(restype):
- """Return a template payload for the known datatypes."""
- if restype == application:
- return application_payload
- elif restype == container:
- return container_payload
- elif restype == contentInstance:
- return contentInstance_payload
- else:
- return ""
-
-
-def find_key(response, key):
- try:
- val = response.json()
- return val['any'][0][key]
- except:
- return None
-
-
-def name(response):
- """Return the resource name in the response."""
- return find_key(response, "rn")
-
-
-def resid(response):
- """Return the resource id in the response."""
- return find_key(response, "ri")
-
-
-def parent(response):
- """Return the parent resource id in the response."""
- return find_key(response, "pi")
-
-
-def content(response):
- """Return the content the response."""
- return find_key(response, "con")
-
-
-def restype(response):
- """Return the resource type the response."""
- return find_key(response, "rty")
-
-
-def status(response):
- """Return the protocol status code in the response."""
- try:
- return response.status_code
- except:
- return None
-
-
-def headers(response):
- """Return the protocol headers in the response."""
- try:
- return response.headers
- except:
- return None
-
-
-def error(response):
- """Return the error string in the response."""
- try:
- return response.json()['error']
- except:
- return None
-
-
-def normalize(id):
- if id is not None:
- if id[0] == "/":
- return id[1:]
- return id
-
-
-def attr2str(attr):
- """Convert a dictionary into a string for a protocol payload."""
- content = ""
- if attr is not None:
- content = ","
- n = len(attr)
- c = 1
- sep = ","
- for i in attr:
- if n == c:
- sep = ""
- n = str(attr[i])
- if i != "con" and n.isdigit():
- content = content + "'%s':%d%s" % (i, int(n), sep)
- else:
- content = content + "'%s':'%s'%s" % (i, n, sep)
- c = c + 1
- return content
-
-
-class connect:
- def __init__(self, server="localhost", base='InCSE1',
- auth=('admin', 'admin'), protocol="http"):
- """Connect to a IoTDM server."""
- self.s = requests.Session()
- self.s.auth = auth
- self.s.headers.update({'content-type': 'application/json'})
- self.timeout = (5, 5)
- self.payload = cse_payload % (base)
- self.headers = {
- # Admittedly these are "magic values" but are required
- # and until a proper defaulting initializer is in place
- # are hard-coded.
- 'content-type': 'application/json',
- 'X-M2M-Origin': '//localhost:10000',
- 'X-M2M-RI': '12345',
- 'X-M2M-OT': 'NOW'
- }
- self.server = "http://" + server
- if base is not None:
- self.url = self.server + op_provision
- self.r = self.s.post(self.url,
- data=self.payload, timeout=self.timeout)
-
- def create(self, parent, restype, name=None, attr=None):
- """Create resource."""
- if parent is None:
- return None
- payload = which_payload(restype)
- payload = payload % (attr2str(attr))
- if name is None:
- self.headers['X-M2M-NM'] = None
- else:
- self.headers['X-M2M-NM'] = name
- parent = normalize(parent)
- self.url = self.server + ":8282/%s?ty=%s&rcn=1" % (parent, restype)
- self.r = self.s.post(self.url, payload,
- timeout=self.timeout, headers=self.headers)
- return self.r
-
- def retrieve(self, id):
- """Retrieve resource."""
- if id is None:
- return None
- id = normalize(id)
- self.url = self.server + ":8282/%s?rcn=5&drt=2" % (id)
- self.headers['X-M2M-NM'] = None
- self.r = self.s.get(self.url, timeout=self.timeout,
- headers=self.headers)
- return self.r
-
- def update(self, id, attr=None):
- """Update resource attr"""
- if id is None:
- return None
- id = normalize(id)
- return None
-
- def delete(self, id):
- """Delete the resource with the provided ID."""
- if id is None:
- return None
- id = normalize(id)
- self.url = self.server + ":8282/%s" % (id)
- self.headers['X-M2M-NM'] = None
- self.r = self.s.delete(self.url, timeout=self.timeout,
- headers=self.headers)
- return self.r
-
- def tree(self):
- """Get the resource tree."""
- self.url = self.server + op_tree
- self.r = self.s.get(self.url)
- return self.r
-
- def kill(self):
- """Kill the tree."""
- self.url = self.server + op_cleanup
- self.r = self.s.post(self.url)
- return self.r
Code Review / integration.git / blobdiff