op_tree = ":8181/restconf/operational/onem2m:onem2m-resource-tree"
op_cleanup = ":8181/restconf/operations/onem2m:onem2m-cleanup-store"
-cse_payload = '''
+cse_payload = """
{ "input": {
"onem2m-primitive": [
{
]
}
}
-'''
+"""
-application_payload = '''
+application_payload = """
{
any:
[
{"aei":"jb", "api":"jb", "apn":"jb2", "or":"http://hey/you" %s}
]
}
-'''
+"""
-_container_payload = '''
+_container_payload = """
{
any:
[
}
]
}
-'''
+"""
-container_payload = '''
+container_payload = """
{
any:
[
}
]
}
-'''
+"""
-contentInstance_payload = '''
+contentInstance_payload = """
{
"any": [
{
}
]
}
-'''
+"""
def which_payload(restype):
def find_key(response, key):
try:
val = response.json()
- return val['any'][0][key]
- except:
+ return val["any"][0][key]
+ except Exception:
return None
"""Return the protocol status code in the response."""
try:
return response.status_code
- except:
+ except Exception:
return None
"""Return the protocol headers in the response."""
try:
return response.headers
- except:
+ except Exception:
return None
def error(response):
"""Return the error string in the response."""
try:
- return response.json()['error']
- except:
+ return response.json()["error"]
+ except Exception:
return None
class connect:
- def __init__(self, server="localhost", base='InCSE1',
- auth=('admin', 'admin'), protocol="http"):
+ 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.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'
+ "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)
+ self.r = self.s.post(self.url, data=self.payload, timeout=self.timeout)
def create(self, parent, restype, name=None, attr=None):
"""Create resource."""
payload = which_payload(restype)
payload = payload % (attr2str(attr))
if name is None:
- self.headers['X-M2M-NM'] = None
+ self.headers["X-M2M-NM"] = None
else:
- self.headers['X-M2M-NM'] = name
+ 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)
+ self.r = self.s.post(
+ self.url, payload, timeout=self.timeout, headers=self.headers
+ )
return self.r
def retrieve(self, id):
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)
+ 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):
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)
+ 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):