"..", "..", "..", "karaf", "target", "assembly", "bin", "client")
return subprocess.run([executable],
input='feature:install ' + feature_name + '\n feature:list | grep tapi \n logout \n',
- universal_newlines=True)
+ universal_newlines=True, check=False)
def get_request(url):
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
auth=(ODL_LOGIN, ODL_PWD))
- else:
- return requests.request(
- "POST", url.format(RESTCONF_BASE_URL),
- headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
def post_xmlrequest(url, data):
data=data,
headers=TYPE_APPLICATION_XML,
auth=(ODL_LOGIN, ODL_PWD))
+ return None
def put_request(url, data):
def service_path_request(operation: str, servicename: str, wavenumber: str, nodes):
attr = {"renderer:input": {
- "renderer:service-name": servicename,
- "renderer:wave-number": wavenumber,
- "renderer:modulation-format": "qpsk",
- "renderer:operation": operation,
- "renderer:nodes": nodes}}
+ "renderer:service-name": servicename,
+ "renderer:wave-number": wavenumber,
+ "renderer:modulation-format": "qpsk",
+ "renderer:operation": operation,
+ "renderer:nodes": nodes}}
return post_request(URL_SERVICE_PATH, attr)
def create_ots_oms_request(nodeid: str, lcp: str):
attr = {"input": {
- "node-id": nodeid,
- "logical-connection-point": lcp}}
+ "node-id": nodeid,
+ "logical-connection-point": lcp}}
return post_request(URL_CREATE_OTS_OMS, attr)
return subprocess.Popen(
[HONEYNODE_EXECUTABLE, node_port, os.path.join(SAMPLES_DIRECTORY, node_config_file_name)],
stdout=outfile, stderr=outfile)
+ return None
def wait_until_log_contains(log_file, regexp, time_to_wait=20):
+ # pylint: disable=lost-exception
stringfound = False
filefound = False
line = None
signal.alarm(self.seconds)
def __exit__(self, type, value, traceback):
+ # pylint: disable=W0622
signal.alarm(0)