This commit is contained in:
Tim Bentley 2021-08-21 15:16:29 +00:00
parent 7c64e935b5
commit 5acab38871
23 changed files with 220 additions and 50 deletions

View File

@ -9,4 +9,4 @@ lint-yaml:
image: python:latest
script:
- pip install yamllint
- yamllint *.yaml
- yamllint scripts/*.yaml

View File

@ -1,14 +1,18 @@
# OpenLP API tester.
A commandline utility to access a running OpenLP instance via it's API's and test all functionality.
A command line utility to access a running OpenLP instance via it's API's and test all functionality.
It will find OpenLP on the network via ZeroConf and then using the API's build and run a service.
This harness will find a running OpenLP instance on the network via ZeroConf and then using the API's build and run a service.
After each command the results are tested to make sure the WebSockets have changed.
The API's need to be configured as insecure as this is not being tested.
Tests are mastered in JSON files and run using the framework.
## Test Structure
The bible test framework is random so some bible references will not be correct and therefore will not load! The chapter and verse numbers are random numbers!!
## Sample Test Structure
```
process_name: test process
step1:
@ -38,29 +42,43 @@ step6:
The step number, needs to be unique and in order!
* delay
* delay (optional)
How long to wait after a call has been made. Default is 2 seconds for all calls.
How long to wait after a call has been made. Default is 0.3 seconds for all calls.
* max
* max (optional)
How long to wait after a call has been made. Default is 2 seconds for all calls.
Max number of fields which will change on the Web Sockets interface.
* min
* min (optional)
How long to wait after a call has been made. Default is 2 seconds for all calls.
Max number of fields which will change on the Web Sockets interface.
* payload
How long to wait after a call has been made. Default is 2 seconds for all calls.
Json to support addition information to the call.
* name
The step name (function name) to be called to run the step.
## Commands
## Commands and Payloads
| Rule Name | Payload |
|--------------------------|------------------------------|
| trigger_alert | text: <Some text to display> |
| search_and_add | plugin: <plugin name>|
| search_and_live | plugin: <plugin name>|
| service_item_show | item: item_id|
| play_live_item | ?????|
| controller_item_next |?????|
| controller_item_previous |???????|
## Adding API's and Internals
API's and their handers are defining in the callback.py file. This should be the only file that needs to change when adding new API's
The test scripts are in the scripts directory and are a pre defined set of tests.

View File

@ -1,13 +0,0 @@
---
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

View File

@ -1,13 +0,0 @@
---
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

View File

@ -0,0 +1,7 @@
{
"folders": [
{
"path": "."
}
]
}

View File

@ -1,3 +1,3 @@
python3-colorama==0.4.3
websocket-client==0.56.0
websocket-client==0.57.0

6
scripts/alert.yaml Normal file
View File

@ -0,0 +1,6 @@
---
process_name: Show basic alert process
step1:
payload:
text: Show some text
name: trigger_alert

26
scripts/blank.yaml Normal file
View File

@ -0,0 +1,26 @@
---
process_name: Test Display hiding
step1:
delay: 1
name: display_hide
step2:
delay: 1
name: display_show
step3:
delay: 1
name: display_blank
step4:
delay: 1
name: display_show
step5:
delay: 1
name: display_theme
step6:
delay: 1
name: display_show
step7:
delay: 1
name: display_desktop
step8:
delay: 1
name: display_show

4
scripts/themes.yaml Normal file
View File

@ -0,0 +1,4 @@
---
process_name: themes
step1:
name: loop_theme_default

View File

@ -167,7 +167,7 @@ def play_live_items(rtc: object) -> None:
ret = requests.get(rtc.base_url + 'controller/live-items')
assert ret.status_code == 200, f'{ret.status_code} returned from live_item'
i = 0
for _ in json.loads(ret.text):
for _ in json.loads(ret.text)['slides']:
pl = {'max': 1, 'min': 2, 'delay': 0.5, 'name': 'controller_item_show', 'payload': {'id': i}}
rtc.tasks.append(Task(rtc, pl))
i += 1
@ -187,3 +187,131 @@ def controller_item_show(rtc: object, payload: dict) -> None:
print_text('Controller_item_show')
ret = requests.post(rtc.base_url + 'controller/show', json=dict(id=int(id)))
assert ret.status_code == 204, ret.status_code
def controller_item_next(rtc: object, payload: dict) -> None:
print_text('Controller_item_next')
ret = requests.post(rtc.base_url + 'controller/progress', json=dict(action='next'))
assert ret.status_code == 204, ret.status_code
def controller_item_previous(rtc: object, payload: dict) -> None:
print_text('Controller_item_previous')
ret = requests.post(rtc.base_url + 'controller/progress', json=dict(action='previous'))
assert ret.status_code == 204, ret.status_code
def trigger_alert(rtc: object, payload: dict) -> None:
p_text = payload['text']
print_text('trigger_alert')
ret = requests.post(rtc.base_url + 'plugins/alerts', json=dict(text=p_text))
assert ret.status_code == 204, ret.status_code
def get_plugins(rtc: object) -> None:
print_text('get_plugins')
ret = requests.get(rtc.base_url + 'plugins')
assert ret.status_code == 204, ret.status_code
def get_system(rtc: object) -> None:
print_text('get_system')
ret = requests.get(rtc.base_url + 'system')
assert ret.status_code == 204, ret.status_code
def get_live_image(rtc: object) -> None:
print_text('get_live_image')
ret = requests.get(rtc.base_url + 'live-image')
assert ret.status_code == 204, ret.status_code
def loop_theme_default(rtc: object) -> None:
print_text('loop_theme_default')
items = get_themes(rtc)
for item in json.loads(items.text):
print(item)
# items = requests.get(rtc.base_url + 'service/items')
# service = json.loads(items.text)
# limit = len(service)
# random_service = [random.randint(1, limit) for itr in range(limit)]
# # test sequentially
# for item in random_service:
# pl = {'max': 1, 'min': 1, 'name': 'service_item_show', 'payload': {'item': item}}
# rtc.pending.append(Task(rtc, pl))
def get_themes(rtc: object) -> str:
print_text('get_themes')
ret = requests.get(rtc.base_url + 'controller/themes')
assert ret.status_code == 200, ret.status_code
return ret.text
def get_default_theme(rtc: object) -> None:
print_text('get_default_theme')
ret = requests.get(rtc.base_url + 'controller/theme')
assert ret.status_code == 204, ret.status_code
def get_themes_name(rtc: object, payload: dict) -> None:
t_name = payload['theme_name']
print_text('get_themes_name')
ret = requests.get(rtc.base_url + f'themes/{t_name}')
assert ret.status_code == 204, ret.status_code
def get_live_theme(rtc: object) -> None:
print_text('get_live_theme')
ret = requests.get(rtc.base_url + 'live_theme')
assert ret.status_code == 204, ret.status_code
def get_theme_level(rtc: object) -> None:
print_text('get_theme_level')
ret = requests.get(rtc.base_url + 'theme_level')
assert ret.status_code == 204, ret.status_code
def set_theme_level(rtc: object, payload: dict) -> None:
t_level = payload['theme_level']
print_text('set_theme_level')
ret = requests.post(rtc.base_url + 'theme_level', json=dict(level=t_level))
assert ret.status_code == 204, ret.status_code
def set_theme(rtc: object, payload: dict) -> None:
t_level = payload['theme_level']
print_text('set_theme_level')
ret = requests.post(rtc.base_url + 'theme_level', json=dict(level=t_level))
assert ret.status_code == 204, ret.status_code
def display_hide(rtc: object) -> None:
print_text('display_hide')
ret = requests.post(rtc.base_url + 'core/display', json=dict(display='hide'))
assert ret.status_code == 204, ret.status_code
def display_show(rtc: object) -> None:
print_text('display_show')
ret = requests.post(rtc.base_url + 'core/display', json=dict(display='show'))
assert ret.status_code == 204, ret.status_code
def display_blank(rtc: object) -> None:
print_text('display_blank')
ret = requests.post(rtc.base_url + 'core/display', json=dict(display='blank'))
assert ret.status_code == 204, ret.status_code
def display_theme(rtc: object) -> None:
print_text('display_theme')
ret = requests.post(rtc.base_url + 'core/display', json=dict(display='theme'))
assert ret.status_code == 204, ret.status_code
def display_desktop(rtc: object) -> None:
print_text('display_desktop')
ret = requests.post(rtc.base_url + 'core/display', json=dict(display='desktop'))
assert ret.status_code == 204, ret.status_code

View File

@ -23,24 +23,24 @@ init(autoreset=True)
def print_text(text: str):
print(Fore.LIGHTMAGENTA_EX + '[*] = ' + text)
print(Fore.LIGHTMAGENTA_EX + f'[*] {text}')
def print_error(text: str):
print(Fore.RED + '[-] = ' + text)
print(Fore.RED + f'[-] {text}')
def print_ok(text: str):
print(Fore.GREEN + '[_] = ' + text)
print(Fore.GREEN + f'[_] {text}')
def print_warn(text: str):
print(Fore.YELLOW + '[?] = ' + text)
print(Fore.YELLOW + f'[?] {text}')
def print_info(text: str):
print(Fore.MAGENTA + '[!] = ' + text)
print(Fore.MAGENTA + f'[!] {text}')
def print_debug(text: str):
print(Fore.CYAN + '[#] = ' + text)
print(Fore.CYAN + f'[#] {text}')

View File

@ -116,7 +116,8 @@ class RunTestsController(object):
parser = argparse.ArgumentParser()
parser.add_argument('rargs', nargs='*', default=[])
a = parser.parse_args()
with open(a.rargs[0], 'r') as file:
f_open = f'{a.rargs[0]}'
with open(f_open, 'r') as file:
commands = yaml.load(file, Loader=yaml.FullLoader)
for step in commands:
if step == 'process_name':

View File

@ -23,6 +23,11 @@ from test_api.apitest.callbacks import * # noqa E403
def human_delay(delay: int = 2) -> None:
"""
Delay time in Seconds
:param delay:
:return:
"""
time.sleep(delay)

View File

@ -36,8 +36,9 @@ def check_for_openlp(type: str) -> str:
def add_service(self, zeroconf: Zeroconf, type: str, name: str):
info = zeroconf.get_service_info(type, name)
print_info(f'Service {name} added, service info: {info}')
# print_info(f'Service {name} added, service info: {info}')
if info.name.startswith("OpenLP"):
print_info(f'Service {name} found')
self.port = str(info.port)
self.address = socket.inet_ntoa(info.addresses[0])

View File

@ -27,11 +27,11 @@ def start() -> None:
Instantiate and run the tests.
:return:
"""
print_text('OpenLP - API Test Runner V0_2')
print_text('OpenLP - API Test Runner V0_3')
print_text('Check OpenLP is running')
op_address, op_http_port = check_for_openlp('http')
if not op_http_port:
print_error('OpenLP is not running ')
print_error('OpenLP is not running - aborting')
return
else:
print_ok(f'OpenLP is running on port (http) {op_http_port}')