181 lines
7.0 KiB
Python
181 lines
7.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
##########################################################################
|
|
# OpenLP - Open Source Lyrics Projection #
|
|
# ---------------------------------------------------------------------- #
|
|
# Copyright (c) 2008-2021 OpenLP Developers #
|
|
# ---------------------------------------------------------------------- #
|
|
# This program is free software: you can redistribute it and/or modify #
|
|
# it under the terms of the GNU General Public License as published by #
|
|
# the Free Software Foundation, either version 3 of the License, or #
|
|
# (at your option) any later version. #
|
|
# #
|
|
# This program is distributed in the hope that it will be useful, #
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
|
|
# GNU General Public License for more details. #
|
|
# #
|
|
# You should have received a copy of the GNU General Public License #
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
|
|
##########################################################################
|
|
import argparse
|
|
import threading
|
|
import websocket
|
|
import json
|
|
|
|
from collections import deque
|
|
from websocket import create_connection
|
|
|
|
from test_api.apitest.logger import print_error, print_ok, print_info, print_debug
|
|
|
|
from test_api.apitest.callbacks import * # noqa E403
|
|
|
|
|
|
from test_api.apitest.task import Task
|
|
from test_api.apitest.task import human_delay
|
|
|
|
|
|
class RunTestsController(object):
|
|
|
|
def __init__(self, address, http_port: str, ws_port: str):
|
|
self.address = address
|
|
self.http_port = http_port
|
|
self.ws_port = ws_port
|
|
reserved_result_stage = {'results': ''}
|
|
reserved_result_stage = json.dumps(reserved_result_stage)
|
|
self.reserved_result_stage = bytes(reserved_result_stage, 'utf-8')
|
|
self.received = False
|
|
self.tasks = deque()
|
|
self.pending = deque()
|
|
self.ws_data = deque()
|
|
self.base_url = f'http://{self.address}:{self.http_port}/api/v2/'
|
|
self.debug = False
|
|
|
|
def connect(self) -> None:
|
|
print_info("Starting thread")
|
|
self.ws = websocket.WebSocketApp(f'ws://{self.address}:{self.ws_port}',
|
|
on_message=self.on_message,
|
|
on_close=self.on_close,
|
|
on_open=self.on_open,
|
|
on_error=self.on_error,
|
|
)
|
|
self.wst = threading.Thread(target=lambda: self.ws.run_forever())
|
|
self.wst.daemon = True
|
|
self.wst.start()
|
|
|
|
conn_timeout = 5
|
|
while (not self.ws.sock or not self.ws.sock.connected) and conn_timeout:
|
|
human_delay(1)
|
|
conn_timeout -= 1
|
|
if not conn_timeout:
|
|
print_error("Could not connect to WS! Exiting.")
|
|
|
|
def on_message(self, ws_obj: object, message: str) -> None:
|
|
if self.debug:
|
|
print_debug(f"Message returned: {message}")
|
|
if not self.received:
|
|
self.reserved_result_stage = message
|
|
else:
|
|
self.ws_data.append(message)
|
|
|
|
@staticmethod
|
|
def on_open(ws_obj: object) -> None:
|
|
print_info(f"Socket Listener Opened")
|
|
|
|
@staticmethod
|
|
def on_close(ws_obj: object) -> None:
|
|
print_info("Socket Listener Closed")
|
|
|
|
@staticmethod
|
|
def on_error(ws_obj: object, error: str) -> None:
|
|
print_error(f'WebSocket Error: {error}')
|
|
|
|
def load_and_check_sockets(self, first_run: bool = False) -> bool:
|
|
"""
|
|
Method to make connect to websockets
|
|
:param first_run:
|
|
:return:
|
|
"""
|
|
ws = create_connection(f'ws://{self.address}:{self.ws_port}', ping_interval=None)
|
|
if first_run:
|
|
print_text('Establishing Connection to State')
|
|
self.result_stage = ws.recv()
|
|
good = True
|
|
if self.result_stage:
|
|
if first_run:
|
|
print_ok('Connected to state... Good To test')
|
|
else:
|
|
if not self.reserved_result_stage:
|
|
self.reserved_result_stage = self.result_stage
|
|
else:
|
|
good = False
|
|
print_error('Not Connected to state ... you have issues')
|
|
ws.close()
|
|
return good
|
|
|
|
def read_command_file(self) -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('rargs', nargs='*', default=[])
|
|
a = parser.parse_args()
|
|
f_open = f'{a.rargs[0]}'
|
|
print(f_open)
|
|
with open(f_open, 'r') as file:
|
|
commands = json.load(file)
|
|
for step in commands:
|
|
if step == 'process_name':
|
|
print_info(f'Processing file {commands[step]}')
|
|
continue
|
|
self.tasks.append(Task(self, commands[step]))
|
|
self.process_queue()
|
|
|
|
def compare_stage(self):
|
|
"""
|
|
:return:
|
|
"""
|
|
self.stage_diff = {}
|
|
reserved = json.loads(self.reserved_result_stage.decode('utf-8'))['results']
|
|
current = json.loads(self.result_stage.decode('utf-8'))['results']
|
|
# compare_strings(reserved, current)
|
|
if len(reserved) > 0:
|
|
assert len(reserved) == len(current)
|
|
for a, _ in current.items():
|
|
# compare_strings(reserved[a], current[a])
|
|
if reserved[a] != current[a]:
|
|
self.stage_diff[a] = {'before': reserved[a], 'after': current[a]}
|
|
self.reserved_result_stage = self.result_stage
|
|
|
|
def process_queue(self):
|
|
while len(self.tasks) > 0:
|
|
task = self.tasks.popleft()
|
|
task.run()
|
|
# we have finished but have pending tasks to complete
|
|
if len(self.tasks) == 0 and len(self.pending) > 0:
|
|
pend = self.pending.popleft()
|
|
self.tasks.append(pend)
|
|
|
|
def check_websocket_changes(self, mandatory: int, optional: int) -> None:
|
|
while len(self.ws_data) > 0:
|
|
message = self.ws_data.popleft()
|
|
self.result_stage = message
|
|
self.compare_stage()
|
|
if mandatory <= len(self.stage_diff) <= mandatory + optional:
|
|
pass
|
|
else:
|
|
print(f'{mandatory} stage field(s) must have changed and '
|
|
f'{optional} may change changed- {str(self.stage_diff)}')
|
|
|
|
|
|
def compare_strings(a, b):
|
|
import difflib
|
|
print('{} => {}'.format(a, b))
|
|
if isinstance(a, str):
|
|
for i, s in enumerate(difflib.ndiff(a, b)):
|
|
if s[0] == ' ':
|
|
continue
|
|
elif s[0] == '-':
|
|
print(u'Delete "{}" from position {}'.format(s[-1], i))
|
|
elif s[0] == '+':
|
|
print(u'Add "{}" to position {}'.format(s[-1], i))
|
|
else:
|
|
if a != b:
|
|
print('{} => {} are different'.format(a, b))
|