From 11c60f13692e87967a6ac479aa2968d7f5d3dd30 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Wed, 25 Nov 2020 10:46:01 -0800 Subject: [PATCH] test_status_msg.py: tests the /status-msg JSON url features --- py-scripts/test_status_msg.py | 150 ++++++++++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100755 py-scripts/test_status_msg.py diff --git a/py-scripts/test_status_msg.py b/py-scripts/test_status_msg.py new file mode 100755 index 00000000..e3b28ea6 --- /dev/null +++ b/py-scripts/test_status_msg.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 + +import os +import sys + +if sys.version_info[0] != 3: + print("This script requires Python 3") + exit(1) + +if 'py-json' not in sys.path: + sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) + +import argparse +from LANforge.lfcli_base import LFCliBase +import time +from uuid import uuid1 +import pprint +from pprint import pprint + +class TestStatusMessage(LFCliBase): + def __init__(self, host, port, + _debug_on=False, + _exit_on_error=False, + _exit_on_fail=False): + super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail) + self.check_connect() + + + def build(self): + """create a new session""" + new_session = uuid1() + self.status_msg_url = "/status-msg" + self.session_url = "/status-msg/"+str(new_session) + # print("----- ----- ----- ----- ----- PUT ----- ----- ----- ----- ----- ----- ") + self.json_put(self.session_url, _data={}) + + # we should see list of sessions + try: + #print("----- ----- ----- ----- ----- GET ----- ----- ----- ----- ----- ----- ") + session_response = self.json_get(self.status_msg_url) + if self.debug: + pprint(session_response) + if "sessions" not in session_response: + print("----- ----- ----- ----- ----- BAD ----- ----- ----- ----- ----- ----- ") + self._fail("response lacks sessions element") + if len(session_response["sessions"]) < 2: + self._fail("why do we have less than two sessions?") + for session in session_response["sessions"]: + #print("----- ----- ----- ----- ----- SESSION ----- ----- ----- ----- ----- ----- ") + pprint(session) + self._pass("session created") + except ValueError as ve: + print("----- ----- ----- ----- ----- what??? ----- ----- ----- ----- ----- ----- ") + self._fail(ve) + + + def start(self, print_pass=False, print_fail=False): + """ + create a series of messages + :return: None + """ + #print("----- ----- ----- ----- ----- START ----- %s ----- ----- ----- ----- ----- " % self.session_url) + message_response = self.json_get(self.session_url) + if self.debug: + pprint(message_response) + if "empty" in message_response: + self._pass("empty response, zero messages") + elif "messages" in message_response: + messages_a = message_response["messages"] + if len(messages_a) > 0: + self._fail("we should have zero messages") + + + for msg_num in ( 1, 2, 3, 4, 5 ): + #print("----- ----- ----- ----- ----- ----- %s ----- ----- ----- ----- ----- " % msg_num) + #print("session url: "+self.session_url) + self.json_post(self.session_url, { + "key": "test_status_message.py", + "content-type":"application/json", + "message":"message %s"%msg_num + }) + message_response = self.json_get(self.session_url) + if len(message_response["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) + + def stop(self): + """ + make sure we read those messages + :return: None + """ + message_response = self.json_get(self.session_url) + if "empty" in message_response: + self._fail("empty response, we expect 1 or more messages") + for message_o in message_response["messages"]: + msg_url = message_o["_links"] + print("Message url: "+msg_url) + + def cleanup(self): + """delete messages and delete the session""" + self._fail("TODO") + pass + +def main(): + lfjson_port = 8080 + parser = LFCliBase.create_bare_argparse( + prog=__file__, + # formatter_class=argparse.RawDescriptionHelpFormatter, + formatter_class=argparse.RawTextHelpFormatter, + + description=""" +Test the status message passing functions of /status-msg: +- create a session: PUT /status-msg/ +- post message: POST /status-msg/ +- list sessions: GET /status-msg/ +- list messages for session: GET /status-msg/ +- delete message: DELETE /status-msg//message-id +- delete session: DELETE /status-msg//this +- delete all messages in session: DELETE /status-msg//all +""") + parser.add_argument('--new', help='create new session') + parser.add_argument('--update', help='add message to session') + parser.add_argument('--read', help='read message(s) from session') + parser.add_argument('--list', help='list messages from session') + parser.add_argument('--delete', help='delete message') + args = parser.parse_args() + + status_messages = TestStatusMessage(args.mgr, + lfjson_port, + _debug_on=True, + _exit_on_error=True, + _exit_on_fail=True) + status_messages.build() + if not status_messages.passes(): + print(status_messages.get_fail_message()) + exit(1) + status_messages.start(False, False) + status_messages.stop() + if not status_messages.passes(): + print(status_messages.get_fail_message()) + exit(1) + status_messages.cleanup() + if status_messages.passes(): + print("Full test passed, all messages read and cleaned up") + + +if __name__ == "__main__": + main()