import ciotdm
-def connect_to_iotdm(host, user, password, prot):
+def connect_to_iotdm(host, user, password, prot="http"):
"""According to protocol, connect to iotdm."""
- return ciotdm.connect(host, base="InCSE1", auth=(
- user, password), protocol=prot)
+ return ciotdm.connect(host, base="InCSE1", auth=(user, password), protocol=prot)
def modify_headers_origin(connection, new_origin):
return connection.response
-def create_resource_with_command(connection, parent, restype,
- command, attribute=None):
+def create_resource_with_command(connection, parent, restype, command, attribute=None):
"""According to command in the header, create the resource."""
- connection.create_with_command(parent, restype,
- command, attribute)
+ connection.create_with_command(parent, restype, command, attribute)
check_response(connection.response, "create")
return connection.response
def create_subscription(connection, parent, ip, port):
"""Create subscription."""
uri = "http://%s:%d" % (ip, int(port))
- connection.create(parent, "subscription", {
- "notificationURI": uri,
- "notificationContentType": "wholeResource"})
+ connection.create(
+ parent,
+ "subscription",
+ {"notificationURI": uri, "notificationContentType": "wholeResource"},
+ )
check_response(connection.response, "create")
return connection.response
return connection.response
-def update_resource_with_command(connection, resid,
- restype, command, attr):
+def update_resource_with_command(connection, resid, restype, command, attr):
"""According to command, update resource with resourceID."""
connection.update_with_command(resid, restype, command, attr)
check_response(connection.response, "update")
"""Return resourceName."""
resource_name = ciotdm.name(response)
if resource_name is None:
- raise AssertionError('Cannot find this resource')
+ raise AssertionError("Cannot find this resource")
return resource_name
def location(response):
"""Return response content-location."""
- return response.headers['Content-Location']
+ return response.headers["Content-Location"]
def kill_the_tree(host, cseid, username, password):
"""Delete the whole tree."""
- connection = ciotdm.connect(host, base=cseid,
- auth=(username, password), protocol="http")
+ connection = ciotdm.connect(
+ host, base=cseid, auth=(username, password), protocol="http"
+ )
connection.kill()
def check_response(response, operation):
"""Check whether the connection is none."""
if response is None:
- raise AssertionError('Cannot %s this resource') % (operation)
- elif hasattr(response, 'status_code'):
+ raise AssertionError("Cannot %s this resource") % (operation)
+ elif hasattr(response, "status_code"):
if response.status_code < 200 or response.status_code > 299:
raise AssertionError(
- 'Cannot %s this resource [%d] : %s' %
- (operation, response.status_code, response.text))
+ "Cannot %s this resource [%d] : %s"
+ % (operation, response.status_code, response.text)
+ )