openlp
/
openlp_api_tester
mirror of https://gitlab.com/openlp/openlp_api_tester.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add yaml control.

This commit is contained in:
Tim 2020-06-26 08:38:30 +01:00
parent a96a96a4d4
commit 21dba39858
No known key found for this signature in database
GPG Key ID: 3D454289AF831A6D
11 changed files with 345 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
alert.yaml Normal file
View File

@ -0,0 +1,12 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service

12
blank.yaml Normal file
View File

@ -0,0 +1,12 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service

85
full_run_rand.yaml Normal file
View File

@ -0,0 +1,85 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service
step4:
delay: 1
name: search_and_add
payload:
plugin: songs
step5:
delay: 1
name: search_and_add
payload:
plugin: songs
step6:
delay: 1
name: search_and_add
payload:
plugin: bibles
step7:
delay: 1
name: search_and_add
payload:
plugin: bibles
step8:
delay: 1
name: search_and_add
payload:
plugin: custom
step9:
delay: 1
name: search_and_add
payload:
plugin: images
step10:
delay: 1
name: search_and_add
payload:
plugin: songs
step11:
delay: 1
name: search_and_add
payload:
plugin: presentation
step12:
delay: 1
name: search_and_add
payload:
plugin: bibles
step13:
delay: 1
name: search_and_add
payload:
plugin: bibles
step14:
delay: 1
name: search_and_add
payload:
plugin: songs
step15:
delay: 1
name: search_and_add
payload:
plugin: custom
step16:
delay: 1
name: search_and_add
payload:
plugin: bibles
step17:
delay: 1
name: search_and_add
payload:
plugin: bibles
step18:
delay: 0.1
name: load_service_sequential

85
full_run_seq.yaml Normal file
View File

@ -0,0 +1,85 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service
step4:
delay: 1
name: search_and_add
payload:
plugin: songs
step5:
delay: 1
name: search_and_add
payload:
plugin: songs
step6:
delay: 1
name: search_and_add
payload:
plugin: bibles
step7:
delay: 1
name: search_and_add
payload:
plugin: bibles
step8:
delay: 1
name: search_and_add
payload:
plugin: custom
step9:
delay: 1
name: search_and_add
payload:
plugin: images
step10:
delay: 1
name: search_and_add
payload:
plugin: songs
step11:
delay: 1
name: search_and_add
payload:
plugin: presentation
step12:
delay: 1
name: search_and_add
payload:
plugin: bibles
step13:
delay: 1
name: search_and_add
payload:
plugin: bibles
step14:
delay: 1
name: search_and_add
payload:
plugin: songs
step15:
delay: 1
name: search_and_add
payload:
plugin: custom
step16:
delay: 1
name: search_and_add
payload:
plugin: bibles
step17:
delay: 1
name: search_and_add
payload:
plugin: bibles
step18:
delay: 0.1
name: load_service_random

42
media.yaml Normal file
View File

@ -0,0 +1,42 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service
step4:
delay: 5
name: media_play
step5:
delay: 1
name: clear_preview_controller
step6:
max: 1
min: 1
name: new_service
step7:
delay: 1
name: search_and_add
payload:
plugin: media
step8:
delay: 5
name: media_play
step9:
delay: 5
name: media_pause
step10:
delay: 5
name: media_pause
step11:
delay: 5
name: media_play
step12:
delay: 5
name: media_stop

25
test.yaml Normal file
View File

@ -0,0 +1,25 @@
process_name: test process
step1:
max: 1
min: 1
name: clear_live_controller
step2:
delay: 1
name: clear_preview_controller
step3:
max: 1
min: 1
name: new_service
step4:
delay: 1
name: search_and_add
payload:
plugin: songs
step5:
delay: 1
name: search_and_add
payload:
plugin: songs
step6:
delay: 0.1
name: load_service_sequential

25
test2.py Normal file
View File

@ -0,0 +1,25 @@
import json
import yaml
# ...
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('rargs', nargs='*', default=[])
a = parser.parse_args()
with open(a.rargs[0], 'r') as file:
# The FullLoader parameter handles the conversion from YAML
# scalar values to Python the dictionary format
xx = yaml.load(file)
print(yaml.dump(xx))
for x in xx:
print(x)
print(xx[x])
c = xx[x]
print(c)
if isinstance(c, str):
pass
else:
try:
print(c['delay'])
except KeyError:
pass

27
test_api/apitest/callbacks.py
View File

@ -25,7 +25,7 @@ import string
from test_api.apitest.constants import BookNames
from test_api.apitest.logger import print_text
from test_api.apitest.task import Task
from test_api.apitest.task import Task, human_delay
def clear_live_controller(rtc: object) -> None:
@ -47,7 +47,8 @@ def new_service(rtc: object) -> None:
assert ret.status_code == 204, f'{ret.status_code} returned'
def search_and_add(rtc: object, plugin: str) -> None:
def search_and_add(rtc: object, payload: dict) -> None:
plugin = payload['plugin']
print_text(f'Search_and_add for {plugin}')
if plugin == 'bibles':
bk_id = random.randint(1, len(BookNames) - 1)
@ -83,7 +84,8 @@ def search_and_add(rtc: object, plugin: str) -> None:
assert ret.status_code == 204, f'{ret.status_code} returned from add'
def search_and_live(rtc: object, plugin: str) -> None:
def search_and_live(rtc: object, payload: dict) -> None:
plugin = payload['plugin']
print_text(f'Search_and_live for {plugin}')
while True:
let = random.choice(string.ascii_letters)
@ -129,7 +131,8 @@ def load_service_sequential(rtc: object) -> None:
if item['plugin'] == 'video':
pass
else:
rtc.pending.append(Task(rtc, service_item_show, poll_m=1, poll_o=2, ag=item))
pl = {'max': 1, 'min': 1, 'name': 'service_item_show', 'payload': {'item': item}}
rtc.pending.append(Task(rtc, pl))
# rtc.tasks.append(Task(rtc, play_live_item, poll_m=1, poll_o=2, ag=item))
#rtc.tasks.append(Task(rtc, clear_live_controller, poll_m=1, poll_o=1))
@ -137,29 +140,35 @@ def load_service_sequential(rtc: object) -> None:
#self.load_service_random()
def service_item_show(rtc: object, item: dict) -> None:
def service_item_show(rtc: object, payload: dict) -> None:
item = payload['item']
print_text('Service_item_show')
title = item['title']
id = item['id']
print_text(f'test_service_song {title} {id}')
ret = requests.post(rtc.base_url + 'service/show', json=dict(uid=id))
assert ret.status_code == 204, ret.status_code
rtc.tasks.append(Task(rtc, play_live_item, poll_m=1, poll_o=2, ag=item))
pl = {'max': 1, 'min': 2, 'name': 'play_live_item', 'payload': {'item': item}}
rtc.tasks.append(Task(rtc, pl))
def play_live_item(rtc: object, item: dict) -> None:
def play_live_item(rtc: object, payload: dict) -> None:
item = payload['item']
plugin = item['plugin']
print_text(f'Play_live_item - {plugin}')
ret = requests.get(rtc.base_url + 'controller/live-item')
assert ret.status_code == 200, f'{ret.status_code} returned from live_item'
i = 0
for _ in json.loads(ret.text):
rtc.tasks.append(Task(rtc, controller_item_show, delay=0.1, poll_m=1, poll_o=2, ag=str(i)))
pl = {'max': 1, 'min': 2, 'delay': 1, 'name': 'controller_item_show', 'payload': {'id': i}}
rtc.tasks.append(Task(rtc, pl))
i += 1
assert ret.status_code == 200, f'{ret.status_code} returned from show'
def controller_item_show(rtc: object, id: str) -> None:
def controller_item_show(rtc: object, payload: dict) -> None:
id = payload['id']
print_text('Controller_item_show')
ret = requests.post(rtc.base_url + 'controller/show', json=dict(id=int(id)))
human_delay()
assert ret.status_code == 204, ret.status_code

75
test_api/apitest/runner.py
View File

@ -18,9 +18,11 @@
# 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 time
import yaml
from collections import deque
from websocket import create_connection
@ -79,7 +81,7 @@ class RunTestsController(object):
print_error(f'WebSocket Error: {error}')
def load_and_check_sockets(self, first_run: bool = False) -> bool:
ws = create_connection(f'ws://{self.address}:{self.ws_port}')
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()
@ -96,6 +98,20 @@ class RunTestsController(object):
ws.close()
return good
def read_command_file(self) -> None:
parser = argparse.ArgumentParser()
parser.add_argument('rargs', nargs='*', default=[])
a = parser.parse_args()
with open(a.rargs[0], 'r') as file:
commands = yaml.load(file, Loader=yaml.FullLoader)
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:
@ -114,32 +130,6 @@ class RunTestsController(object):
self.stage_diff[a] = {'before': reserved[a], 'after': current[a]}
self.reserved_result_stage = self.result_stage
def marshal_full(self):
print_ok('Running full test script')
if self.load_and_check_sockets(True):
self.tasks.append(Task(self, clear_live_controller, poll_m=1, poll_o=1))
self.tasks.append(Task(self, clear_preview_controller, delay=1))
self.tasks.append(Task(self, new_service, poll_m=1, poll_o=1))
self.tasks.append(Task(self, search_and_add, delay=1, ag='songs'))
self.tasks.append(Task(self, search_and_add, delay=1, ag='songs'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='custom'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='images'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='songs'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='presentations'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='songs'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='custom'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='custom'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='songs'))
#self.tasks.append(Task(self, search_and_add, delay=1, ag='bibles'))
self.tasks.append(Task(self, load_service_sequential, delay=0.1))
self.process_queue()
def process_queue(self):
while len(self.tasks) > 0:
task = self.tasks.popleft()
@ -149,37 +139,6 @@ class RunTestsController(object):
pend = self.pending.popleft()
self.tasks.append(pend)
def marshal_media(self):
print_ok('Running media test script')
if self.load_and_check_sockets(True):
self.tasks.append(Task(self, clear_live_controller, poll_m=1, poll_o=1))
self.tasks.append(Task(self, clear_preview_controller, delay=1))
self.tasks.append(Task(self, new_service, poll_m=1, poll_o=1))
self.tasks.append(Task(self, search_and_live, delay=1, ag='songs'))
self.tasks.append(Task(self, media_play, delay=2))
self.tasks.append(Task(self, clear_preview_controller, delay=1))
self.tasks.append(Task(self, new_service, poll_m=1, poll_o=1))
self.tasks.append(Task(self, search_and_live, delay=1, ag='media'))
self.tasks.append(Task(self, media_play, delay=5))
self.tasks.append(Task(self, media_pause, delay=5))
self.tasks.append(Task(self, media_pause, delay=5))
self.tasks.append(Task(self, media_play, delay=5))
self.tasks.append(Task(self, media_stop, delay=1))
def marshal_blank(self):
print_ok('Running blank test script')
if self.load_and_check_sockets(True):
self.tasks.append(Task(self, clear_live_controller, poll_m=1, poll_o=1))
self.tasks.append(Task(self, clear_preview_controller, delay=1))
self.tasks.append(Task(self, new_service, poll_m=1, poll_o=1))
def marshal_alert(self):
print_ok('Running alert test script')
if self.load_and_check_sockets(True):
self.tasks.append(Task(self, clear_live_controller, poll_m=1, poll_o=1))
self.tasks.append(Task(self, clear_preview_controller, delay=1))
self.tasks.append(Task(self, new_service, poll_m=1, poll_o=1))
def load_service_random(self) -> None:
print_text('Load_and process_service_random')
items = requests.get(f'http://{self.address}:{self.http_port}/api/v2/service/items')

37
test_api/apitest/task.py
View File

@ -18,7 +18,9 @@
# 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 importlib
import time
from test_api.apitest.callbacks import * # noqa E403
def human_delay(delay: int = 2) -> None:
@ -27,22 +29,27 @@ def human_delay(delay: int = 2) -> None:
class Task(object):
def __init__(self, caller: object, callback: object, delay: float = 0, poll_m: int = -1,
poll_o: int = -1, ag: str = None) -> None:
def __init__(self, caller: object, request: dict) -> None:
self.caller = caller
self.delay = delay
self.poll_m = poll_m
self.poll_o = poll_o
self.__setattr__("callback", callback)
self.arg = ag
self.request = request
callback = self.request['name']
function_string = f'test_api.apitest.callbacks.{callback}'
mod_name, func_name = function_string.rsplit('.', 1)
mod = importlib.import_module(mod_name)
func = getattr(mod, func_name)
self.__setattr__("callback", func)
def run(self):
if self.arg:
self.callback(self.caller, self.arg)
else:
try:
payload = self.request['payload']
self.callback(self.caller, payload)
except KeyError:
self.callback(self.caller)
if self.delay > 0:
human_delay(self.delay)
if self.poll_m >= 0:
self.caller.check_websocket_changes(self.poll_m, self.poll_m)
try:
human_delay(self.request['delay'])
except KeyError:
pass
try:
self.caller.check_websocket_changes(self.request['max'], self.request['min'])
except KeyError:
pass

3
test_api/runner.py
View File

@ -42,7 +42,8 @@ def start() -> None:
print_ok(f'OpenLP is running on port (ws) {op_ws_port}')
rtc = RunTestsController(op_address, op_http_port, op_ws_port)
rtc.connect()
rtc.marshal_full()
rtc.read_command_file()
#rtc.marshal_full()
# nrtc.marshal_media()
print_text('Finished running tests')