326 lines
12 KiB
Python
326 lines
12 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 json
|
|
import random
|
|
import requests
|
|
import string
|
|
|
|
from test_api.apitest.constants import BookNames
|
|
from test_api.apitest.logger import print_text
|
|
from test_api.apitest.task import Task
|
|
|
|
|
|
def clear_live_controller(rtc: object) -> None:
|
|
print_text('Clear_live_controllers')
|
|
ret = requests.post(rtc.base_url + 'controller/clear/live')
|
|
assert ret.status_code == 204, f'{ret.status_code} returned from clear live'
|
|
|
|
|
|
def clear_preview_controller(rtc: object) -> None:
|
|
print_text('Clear_preview_controllers')
|
|
# Preview clear should not impact the WebSockets.
|
|
ret = requests.post(rtc.base_url + 'controller/clear/preview')
|
|
assert ret.status_code == 204, f'{ret.status_code} returned from clear preview'
|
|
|
|
|
|
def new_service(rtc: object) -> None:
|
|
print_text('New_service')
|
|
ret = requests.get(rtc.base_url + 'service/new')
|
|
assert ret.status_code == 204, f'{ret.status_code} returned'
|
|
|
|
|
|
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)
|
|
bk = BookNames[bk_id]
|
|
ch = random.randint(1, 10)
|
|
vse = random.randint(1, 10)
|
|
let = f'{bk} {ch}:1-{vse}'
|
|
ret = requests.get(rtc.base_url + f'plugins/{plugin}/search?text={let}')
|
|
assert ret.status_code == 200, f'{ret.status_code} returned from search'
|
|
items = json.loads(ret.text)
|
|
if items:
|
|
ret = requests.post(rtc.base_url + f'plugins/{plugin}/add', json=dict(id=items[0][0]))
|
|
assert ret.status_code == 204, f'{ret.status_code} returned from add'
|
|
else:
|
|
if plugin == 'media':
|
|
while True:
|
|
let = random.choice(string.ascii_letters)
|
|
ret = requests.get(rtc.base_url + f'plugins/{plugin}/search?text={let}')
|
|
if ret.status_code == 200 and ret.text != '[]\n':
|
|
break
|
|
else:
|
|
let = random.choice(string.ascii_letters)
|
|
ret = requests.get(rtc.base_url + f'plugins/{plugin}/search?text={let}')
|
|
assert ret.status_code == 200, f'{ret.status_code} returned from search'
|
|
items = json.loads(ret.text)
|
|
limit = len(items) - 1
|
|
if limit >= 0:
|
|
random_item = 0
|
|
if limit > 0:
|
|
random_item = random.randint(1, limit) - 1
|
|
item = items[random_item]
|
|
ret = requests.post(rtc.base_url + f'plugins/{plugin}/add', json=dict(id=item[0]))
|
|
assert ret.status_code == 204, f'{ret.status_code} returned from add'
|
|
|
|
|
|
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)
|
|
ret = requests.get(rtc.base_url + f'plugins/{plugin}/search?text={let}')
|
|
if ret.status_code == 200 and len(json.loads(ret.text)) > 0:
|
|
break
|
|
assert ret.status_code == 200, f'{ret.status_code} returned from search'
|
|
items = json.loads(ret.text)
|
|
limit = len(items) - 1
|
|
if limit >= 0:
|
|
random_service = 0
|
|
if limit > 0:
|
|
random_service = random.randint(1, limit) - 1
|
|
item = items[random_service]
|
|
ret = requests.post(rtc.base_url + f'plugins/{plugin}/live', json=dict(id=item[0]))
|
|
assert ret.status_code == 204, f'{ret.status_code} returned from add'
|
|
|
|
|
|
def flush_pending(rtc: object) -> None:
|
|
print_text('Flush pending to start of tasks')
|
|
pend = rtc.pending.popleft()
|
|
rtc.tasks.insert(0, pend)
|
|
|
|
|
|
def media_play(rtc: object) -> None:
|
|
print_text('Media_play')
|
|
ret = requests.post(rtc.base_url + 'media/play')
|
|
assert ret.status_code == 400, f'{ret.status_code} returned'
|
|
|
|
|
|
def media_pause(rtc: object) -> None:
|
|
print_text('Media_pause')
|
|
ret = requests.post(rtc.base_url + 'media/pause')
|
|
assert ret.status_code == 400, f'{ret.status_code} returned'
|
|
|
|
|
|
def media_stop(rtc: object) -> None:
|
|
print_text('Media_stop')
|
|
ret = requests.post(rtc.base_url + 'media/stop')
|
|
assert ret.status_code == 400, f'{ret.status_code} returned'
|
|
|
|
|
|
def load_service_sequential(rtc: object) -> None:
|
|
print_text('Load_service_sequential')
|
|
items = requests.get(rtc.base_url + 'service/items')
|
|
service = json.loads(items.text)
|
|
# test sequentially
|
|
for item in service:
|
|
pl = {'max': 1, 'min': 1, 'name': 'service_item_show', 'payload': {'item': item}}
|
|
rtc.pending.append(Task(rtc, pl))
|
|
|
|
|
|
def load_service_random(rtc: object) -> None:
|
|
print_text('Load_service_sequential')
|
|
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 service_item_show(rtc: object, payload: dict) -> None:
|
|
print_text('Service_item_show')
|
|
item = payload['item']
|
|
title = item['title']
|
|
id = item['id']
|
|
print_text(f'test_service_item {title} {id}')
|
|
ret = requests.post(rtc.base_url + 'service/show', json=dict(uid=id))
|
|
assert ret.status_code == 204, ret.status_code
|
|
if not item['plugin'] == 'media':
|
|
pl = {'max': 1, 'min': 2, 'delay': 0.5, 'name': 'play_live_items'}
|
|
rtc.tasks.append(Task(rtc, pl))
|
|
|
|
|
|
def play_live_items(rtc: object) -> None:
|
|
print_text('Play_live_items')
|
|
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)['slides']:
|
|
pl = {'max': 1, 'min': 2, 'delay': 0.5, '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 play_live_item(rtc: object, payload: dict) -> None:
|
|
print_text('Play_live_item')
|
|
ret = requests.get(rtc.base_url + 'controller/live-item')
|
|
assert ret.status_code == 200, f'{ret.status_code} returned from live_item'
|
|
aa = json.loads(ret.text)
|
|
print_text(str(aa))
|
|
|
|
|
|
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)))
|
|
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 change_default_theme(rtc: object) -> None:
|
|
print_text('loop_theme_default')
|
|
items = get_themes(rtc)
|
|
default = get_default_theme(rtc)
|
|
first = None
|
|
for item in json.loads(items):
|
|
if item['name'].replace("\"", "") != default and first is None:
|
|
first = item['name']
|
|
set_theme(rtc, first)
|
|
|
|
# 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) -> list:
|
|
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) -> str:
|
|
print_text('get_default_theme')
|
|
ret = requests.get(rtc.base_url + 'controller/theme')
|
|
assert ret.status_code == 200, ret.status_code
|
|
return ret.text.replace("\"", "").replace("\n", "")
|
|
|
|
|
|
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
|
|
return ret.text
|
|
|
|
|
|
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 + 'controller/theme')
|
|
assert ret.status_code == 200, ret.status_code
|
|
return ret.text
|
|
|
|
|
|
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 + 'controller/theme-level', json=dict(level=t_level))
|
|
assert ret.status_code == 204, ret.status_code
|
|
return ret.text
|
|
|
|
|
|
def set_theme(rtc: object, theme_name: str) -> None:
|
|
print_text('set_theme')
|
|
ret = requests.post(rtc.base_url + 'controller/theme', json=dict(theme=theme_name))
|
|
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
|