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

Flake8 fixes

This commit is contained in:
Tim 2020-06-24 17:54:41 +01:00
parent 6b7d25cedc
commit 8b74dd5d6f
No known key found for this signature in database
GPG Key ID: 3D454289AF831A6D
2 changed files with 38 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
test_api/apitest/callbacks.py
View File

@ -25,7 +25,8 @@ import string
import time
from test_api.apitest.constants import BookNames
from test_api.apitest.logger import print_text, print_error, print_ok, print_info
from test_api.apitest.logger import print_text
from test_api.apitest.task import Task
def human_delay(delay: int = 2) -> None:
@ -124,17 +125,20 @@ def media_stop(rtc: object) -> None:
assert ret.status_code == 204, f'{ret.status_code} returned'
def load_service_sequential(self) -> None:
print_text('Load_and_process_service_sequential')
items = requests.get(f'http://{self.address}:{self.http_port}/api/v2/service/items')
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:
if item['plugin'] == 'video':
pass
else:
self.tasks.append(Task(self, service_item_show, poll_m=1, poll_o=2))
self.tasks.append(Task(self, play_live_item, poll_m=1, poll_o=2))
rtc.tasks.append(Task(rtc, service_item_show, poll_m=1, poll_o=2, ag=item))
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))
rtc.tasks.append(Task(rtc, new_service, poll_m=1, poll_o=1))
#self.load_service_random()
def service_item_show(rtc: object, item: dict) -> None:
@ -144,16 +148,22 @@ def service_item_show(rtc: object, item: dict) -> None:
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
live_item(rtc, item['plugin'])
def play_live_item(rtc: object, plugin: str) -> None:
def play_live_item(rtc: object, item: dict) -> None:
plugin = item['plugin']
print_text(f'test_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):
ret = requests.post(rtc.base_url + 'controller/show', json=dict(id=i))
rtc.tasks.append(Task(rtc, controller_item_show, poll_m=1, poll_o=2, ag=str(i)))
i += 1
human_delay()
assert ret.status_code == 204, f'{ret.status_code} returned from show'
def controller_item_show(rtc: object, id: str) -> None:
print_text('Controller_item_show')
ret = requests.post(rtc.base_url + 'controller/show', json=dict(uid=int(id)))
assert ret.status_code == 204, ret.status_code

42
test_api/apitest/runner.py
View File

@ -23,9 +23,9 @@ import websocket
from collections import deque
from websocket import create_connection
from test_api.apitest.callbacks import *
from test_api.apitest.callbacks import * #noqa E403
from test_api.apitest.logger import print_text, print_error, print_ok, print_info
# from test_api.apitest.logger import print_debug
# from test_api.apitest.logger import print_debug #n
from test_api.apitest.task import Task
@ -120,26 +120,22 @@ class RunTestsController(object):
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.load_service_sequential()
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.load_service_random()
#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):
@ -178,8 +174,6 @@ class RunTestsController(object):
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')