scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

test_status_msg: Check if message_list_response exists, otherwise pass. It will post a LANforge error output if you don't write your LANforge URL properly.

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-12-09 09:07:16 -08:00
parent 70dd8c3e5f
commit 5c50b61621
1 changed files with 75 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
py-scripts/test_status_msg.py
View File

@@ -67,12 +67,13 @@ class TestStatusMessage(LFCliBase):
message_response = self.json_get(self.session_url) message_response = self.json_get(self.session_url)
if self.debug: if self.debug:
pprint(message_response) pprint(message_response)
if "empty" in message_response: if message_response:
self._pass("empty response, zero messages") if "empty" in message_response:
elif "messages" in message_response: self._pass("empty response, zero messages")
messages_a = message_response["messages"] elif "messages" in message_response:
if len(messages_a) > 0: messages_a = message_response["messages"]
self._fail("we should have zero messages") if len(messages_a) > 0:
self._fail("we should have zero messages")
for msg_num in (1, 2, 3, 4, 5): for msg_num in (1, 2, 3, 4, 5):
# print("----- ----- ----- ----- ----- ----- %s ----- ----- ----- ----- ----- " % msg_num) # print("----- ----- ----- ----- ----- ----- %s ----- ----- ----- ----- ----- " % msg_num)
@@ -84,9 +85,10 @@ class TestStatusMessage(LFCliBase):
"message": "message %s" % msg_num "message": "message %s" % msg_num
}) })
message_response = self.json_get(self.session_url) message_response = self.json_get(self.session_url)
if len(message_response["messages"]) != msg_num: if message_response:
pprint(message_response) if len(message_response["messages"]) != msg_num:
self._fail("we should have %s messages" % msg_num) pprint(message_response)
self._fail("we should have %s messages" % msg_num)
self._pass("created and listed %s messages counted" % msg_num) self._pass("created and listed %s messages counted" % msg_num)
@@ -97,81 +99,84 @@ class TestStatusMessage(LFCliBase):
""" """
message_list_response = self.json_get(self.session_url) message_list_response = self.json_get(self.session_url)
if "empty" in message_list_response: if message_list_response:
self._fail("empty response, we expect 1 or more messages") if "empty" in message_list_response:
msg_num = 0 self._fail("empty response, we expect 1 or more messages")
for message_o in message_list_response["messages"]: msg_num = 0
msg_url = message_o["_links"] for message_o in message_list_response["messages"]:
print("Message url: " + msg_url) msg_url = message_o["_links"]
message_response = self.json_get(msg_url) print("Message url: " + msg_url)
if self.debug: message_response = self.json_get(msg_url)
pprint(message_response) if self.debug:
for content_o in message_response["messages"]: pprint(message_response)
msg_num += 1 for content_o in message_response["messages"]:
print("id %s" % content_o["message_id"]) msg_num += 1
print("key %s" % content_o["message"]["key"]) print("id %s" % content_o["message_id"])
print("content-type %s" % content_o["message"]["content-type"]) print("key %s" % content_o["message"]["key"])
print("message %s" % content_o["message"]["message"]) print("content-type %s" % content_o["message"]["content-type"])
print("message %s" % content_o["message"]["message"])
if msg_num != self.msg_count: if msg_num != self.msg_count:
self._fail("(stop) expected %s messages, saw %s" % (self.msg_count, msg_num)) self._fail("(stop) expected %s messages, saw %s" % (self.msg_count, msg_num))
else: else:
self._pass("saw correct number of messages") self._pass("saw correct number of messages")
def cleanup(self): def cleanup(self):
"""delete messages and delete the session""" """delete messages and delete the session"""
message_list_response = self.json_get(self.session_url) message_list_response = self.json_get(self.session_url)
if "empty" in message_list_response: if message_list_response:
self._fail("empty response, we expect 1 or more messages") if "empty" in message_list_response:
last_link = "" self._fail("empty response, we expect 1 or more messages")
msg_num = 0 last_link = ""
for message_o in message_list_response["messages"]: msg_num = 0
# print("Delete Message url: "+msg_url) for message_o in message_list_response["messages"]:
last_link = message_o["_links"] # print("Delete Message url: "+msg_url)
msg_num += 1 last_link = message_o["_links"]
msg_num += 1
if msg_num != self.msg_count: if msg_num != self.msg_count:
self._fail("(cleanup) expected %s messages, saw %s" % (self.msg_count, msg_num)) self._fail("(cleanup) expected %s messages, saw %s" % (self.msg_count, msg_num))
message_response = self.json_delete(last_link) message_response = self.json_delete(last_link)
if self.debug: if self.debug:
pprint(message_response) pprint(message_response)
# check message removal # check message removal
message_list_response = self.json_get(self.session_url) message_list_response = self.json_get(self.session_url)
msg_num = len(message_list_response["messages"]) if message_list_response:
if msg_num != (self.msg_count - 1):
self._fail("(cleanup) expected %s messages, saw %s" % ((self.msg_count - 1), msg_num))
else:
self._pass("(cleanup) messages decreased by one")
all_url = self.session_url + "/all"
message_response = self.json_delete(all_url)
if self.debug:
pprint(message_response)
message_list_response = self.json_get(self.session_url)
if self.debug:
pprint(message_list_response)
if "messages" in message_list_response:
msg_num = len(message_list_response["messages"]) msg_num = len(message_list_response["messages"])
elif "empty" in message_list_response: if msg_num != (self.msg_count - 1):
msg_num = 0 self._fail("(cleanup) expected %s messages, saw %s" % ((self.msg_count - 1), msg_num))
else:
self._pass("(cleanup) messages decreased by one")
if msg_num == 0: all_url = self.session_url + "/all"
return "deleted all messages in session" message_response = self.json_delete(all_url)
else: if self.debug:
self._fail("failed to delete all messages in session") pprint(message_response)
if 'empty' in self.json_get(self.session_url).keys(): message_list_response = self.json_get(self.session_url)
if self.debug: if self.debug:
print("--- del -------------------- -------------------- --------------------") pprint(message_list_response)
self.exit_on_error = False if "messages" in message_list_response:
self.json_delete("%s/this" % self.session_url, debug_=False) msg_num = len(message_list_response["messages"])
if self.debug: elif "empty" in message_list_response:
print("--- ~del -------------------- -------------------- --------------------") msg_num = 0
else:
return 'ports deleted successfully' if msg_num == 0:
return "deleted all messages in session"
else:
self._fail("failed to delete all messages in session")
if 'empty' in message_list_response.keys():
if self.debug:
print("--- del -------------------- -------------------- --------------------")
self.exit_on_error = False
self.json_delete("%s/this" % self.session_url, debug_=False)
if self.debug:
print("--- ~del -------------------- -------------------- --------------------")
else:
return 'ports deleted successfully'
sessions_list_response = self.json_get("/status-msg") sessions_list_response = self.json_get("/status-msg")
if self.debug: if self.debug: