monkee.ch/SickGear
1
0
Fork 0
mirror of https://github.com/SickGear/SickGear.git synced 2024-12-21 10:13:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
SickGear/tests/webapi_tests.py
JackDandy 8e8568adb3 Change add jobs to centralise scheduler activities. 
Change refactor scene_exceptions.
2023-04-27 12:47:00 +01:00

844 lines
48 KiB
Python
Raw Blame History

# coding=UTF-8
# Author:
# URL: https://github.com/SickGear/SickGear
#
# This file is part of SickGear.
#
# SickGear 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.
#
# SickGear 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 SickGear. If not, see <http://www.gnu.org/licenses/>.
from __future__ import print_function
import datetime
import itertools
import os
import unittest
import warnings
import sys
import test_lib as test
warnings.filterwarnings('ignore', module=r'.*ssl_.*', message='.*SSLContext object.*')
import sickgear
from exceptions_helper import ex
from sickgear.classes import SearchResult
from sickgear.common import Quality, ARCHIVED, DOWNLOADED, WANTED, UNAIRED, SNATCHED, SNATCHED_PROPER, SNATCHED_BEST, \
SNATCHED_ANY, SUBTITLED, statusStrings, UNKNOWN
from sickgear.event_queue import Events
from sickgear.tv import TVEpisode, TVShow
from sickgear.webserveInit import WebServer
from sickgear import webapi, scheduler, search_backlog, search_queue, show_queue, history, db
from sickgear.scene_numbering import set_scene_numbering_helper
from lib import requests
from six import integer_types, iteritems, iterkeys, itervalues, string_types
# noinspection PyUnreachableCode
if False:
from typing import Any, AnyStr
NoneType = type(None)
today = datetime.date.today()
last_week = today - datetime.timedelta(days=5)
old_date = today - datetime.timedelta(days=14)
future = today + datetime.timedelta(days=1)
far_future = today + datetime.timedelta(days=30)
if 'win32' == sys.platform:
root_folder_tests = [
# root_dirs, path, expected
('1|C:\\dir', 'C:\\folder', None),
('1|c:\\dir', 'c:\\dir', 'c:\\dir'),
('1|c:\\dir2', 'c:\\dir2\\dir', 'c:\\dir2'),
('1|c:\\tv_complete|c:\\tv', 'c:\\tv', 'c:\\tv')
]
else:
root_folder_tests = [
# root_dirs, path, expected
('1|~/dir', '~/dir/dir', '~/dir'),
('1|/mnt/hdd/dir', '/mnt/hdd/folder', None),
('1|/mnt/hdd/dir', '/mnt/hdd/dir', '/mnt/hdd/dir'),
('1|/mnt/hdd/dir2', '/mnt/hdd/dir2/dir', '/mnt/hdd/dir2'),
('1|/mnt/hdd/tv_complete|/mnt/hdd/tv', '/mnt/hdd/tv', '/mnt/hdd/tv')
]
test_shows = [
{'tvid': 1, 'prodid': 1234, 'name': 'Test Show', 'runtime': 45, 'airs': 'Mondays, 00:45', 'imdbid': 'tt1234567',
'_location': r'C:\series\show dir', 'network': 'Network', 'overview': 'Overview text', 'status': 'Continuing',
'quality_init': [], 'quality_upgrade': [],
'episodes': {
1: {
1: {'name': 'ep1', 'status': Quality.composite_status(DOWNLOADED, Quality.HDWEBDL),
'airdate': old_date, 'description': 'ep1 description'},
2: {'name': 'ep2', 'status': WANTED, 'airdate': last_week, 'description': 'ep2 description'},
3: {'name': 'ep3', 'status': WANTED, 'airdate': today, 'description': 'ep3 description'},
4: {'name': 'ep4', 'status': UNAIRED, 'airdate': future, 'description': 'ep4 description'},
5: {'name': 'ep5', 'status': UNAIRED, 'airdate': far_future, 'description': 'ep5 description'},
}
}
},
{'tvid': 1, 'prodid': 5678, 'name': 'Test Show 2', 'runtime': 45, 'airs': 'Tuesdays, 22:15', 'imdbid': 'tt7775567',
'_location': r'C:\series\show 2', 'network': 'Network 2', 'overview': 'Overview text 2', 'status': 'Continuing',
'quality_init': [Quality.HDTV, Quality.FULLHDWEBDL], 'quality_upgrade': [Quality.FULLHDWEBDL],
'episodes': {
1: {
1: {'name': 'new ep1', 'status': UNAIRED, 'airdate': far_future, 'description': 'ep1 description'},
2: {'name': 'new ep2', 'status': UNAIRED, 'airdate': far_future + datetime.timedelta(days=7),
'description': 'ep2 description'},
3: {'name': 'new ep3', 'status': UNAIRED, 'airdate': far_future + datetime.timedelta(days=14),
'description': 'ep3 description'},
4: {'name': 'new ep4', 'status': UNAIRED, 'airdate': far_future + datetime.timedelta(days=28),
'description': 'ep4 description'},
5: {'name': 'new ep5', 'status': UNAIRED, 'description': 'ep5 description'},
}
}
},
]
def fake_action(*args, **kwargs):
pass
class WebAPICase(test.SickbeardTestDBCase):
webserver = None
instance = None
def __init__(self, *args, **kwargs):
super(WebAPICase, self).__init__(*args, **kwargs)
self.org_mass_action = None
self.show_save_db = None
@classmethod
def setUpClass(cls):
super(WebAPICase, cls).setUpClass()
# web server options
sickgear.WEB_PORT = 8080
cls.web_options = dict(
host='127.0.0.1',
port=sickgear.WEB_PORT,
web_root=None,
data_root=os.path.join(sickgear.PROG_DIR, 'gui', sickgear.GUI_NAME),
log_dir=sickgear.LOG_DIR,
username=sickgear.WEB_USERNAME,
password=sickgear.WEB_PASSWORD,
handle_reverse_proxy=sickgear.HANDLE_REVERSE_PROXY,
enable_https=False,
https_cert=None,
https_key=None,
)
# start web server
try:
# used to check if existing SG instances have been started
sickgear.helpers.wait_for_free_port(
sickgear.WEB_IPV6 and '::1' or cls.web_options['host'], cls.web_options['port'])
cls.webserver = WebServer(options=cls.web_options)
cls.webserver.start()
# wait for server thread to be started
cls.webserver.wait_server_start()
cls.webserver.switch_handlers()
sickgear.started = True
sickgear.API_KEYS = [['unit test key', '1234567890']]
sickgear.USE_API = True
except (BaseException, Exception) as e:
print('Failed to start WebServer: %s' % ex(e))
@classmethod
def tearDownClass(cls):
super(WebAPICase, cls).tearDownClass()
# shutdown web server
if cls.webserver:
cls.webserver.shut_down()
try:
cls.webserver.join(10)
except (BaseException, Exception):
pass
if cls.instance:
super(WebAPICase, cls.instance).tearDown()
def setUp(self):
self.reset_show_data = False
self.org_mass_action = None
self.show_save_db = None
if not WebAPICase.instance:
WebAPICase.instance = self
super(WebAPICase, self).setUp()
sickgear.events = Events(None)
sickgear.show_queue_scheduler = scheduler.Scheduler(
show_queue.ShowQueue(),
cycle_time=datetime.timedelta(seconds=3),
thread_name='SHOWQUEUE')
sickgear.search_queue_scheduler = scheduler.Scheduler(
search_queue.SearchQueue(),
cycle_time=datetime.timedelta(seconds=3),
thread_name='SEARCHQUEUE')
sickgear.search_backlog_scheduler = search_backlog.BacklogSearchScheduler(
search_backlog.BacklogSearcher(),
cycle_time=datetime.timedelta(minutes=60),
run_delay=datetime.timedelta(minutes=60),
thread_name='BACKLOG')
sickgear.indexermapper.indexer_list = [i for i in sickgear.indexers.indexer_api.TVInfoAPI().all_sources]
for root_dirs, path, expected in root_folder_tests:
sickgear.ROOT_DIRS = root_dirs
for cur_show in test_shows:
show_obj = TVShow(cur_show['tvid'], cur_show['prodid'])
for k, v in iteritems(cur_show):
if k in ('tvid', 'prodid', 'episodes', 'quality_init', 'quality_upgrade'):
continue
if '_%s' % k in show_obj.__dict__:
show_obj.__dict__['_%s' % k] = v
elif k in show_obj.__dict__:
show_obj.__dict__[k] = v
if 'quality_init' in cur_show and cur_show['quality_init']:
show_obj.quality = Quality.combine_qualities(cur_show['quality_init'],
cur_show.get('quality_upgrade', []))
show_obj.dirty = True
show_obj.save_to_db(True)
sickgear.showList.append(show_obj)
sickgear.showDict.update({show_obj.sid_int: show_obj})
for season, eps in iteritems(cur_show['episodes']):
for ep, data in iteritems(eps):
ep_obj = TVEpisode(show_obj, season, ep)
for k, v in iteritems(data):
if '_%s' % k in ep_obj.__dict__:
ep_obj.__dict__['_%s' % k] = v
elif k in ep_obj.__dict__:
ep_obj.__dict__[k] = v
show_obj.sxe_ep_obj.setdefault(season, {})[ep] = ep_obj
ep_obj.save_to_db(True)
status, quality = Quality.split_composite_status(ep_obj.status)
if status in (DOWNLOADED, SNATCHED):
s_r = SearchResult([ep_obj])
s_r.show_obj, s_r.quality, s_r.provider, s_r.name = \
show_obj, quality, None, '%s.S%sE%s.group' % (
show_obj.name, ep_obj.season, ep_obj.episode)
history.log_snatch(s_r)
if DOWNLOADED == status:
history.log_download(ep_obj, '%s.S%sE%s.group.mkv' % (
show_obj.name, ep_obj.season, ep_obj.episode), quality, 'group')
sickgear.webserve.Home.make_showlist_unique_names()
def tearDown(self):
if None is not self.org_mass_action:
db.DBConnection.mass_action = self.org_mass_action
if None is not self.show_save_db:
sickgear.tv.TVShow.save_to_db = self.show_save_db
sickgear.show_queue_scheduler.action.queue = []
sickgear.search_queue_scheduler.action.queue = []
if self.reset_show_data:
for cur_show in test_shows:
show_obj = sickgear.helpers.find_show_by_id({cur_show['tvid']: cur_show['prodid']})
if 'quality_init' in cur_show and cur_show['quality_init']:
show_obj.quality = Quality.combine_qualities(cur_show['quality_init'],
cur_show.get('quality_upgrade', []))
else:
show_obj.quality = int(sickgear.QUALITY_DEFAULT)
show_obj.upgrade_once = int(cur_show.get('upgrade_once', 0))
show_obj.scene = int(cur_show.get('scene', 0))
show_obj.save_to_db()
for season, data in iteritems(cur_show['episodes']):
for ep_nb, cur_ep in iteritems(data):
ep_obj = show_obj.get_episode(season, ep_nb)
ep_obj.status = cur_ep.get('status')
ep_obj.save_to_db()
set_scene_numbering_helper(
cur_show['tvid'], cur_show['prodid'], season, ep_nb,
scene_season=cur_ep.get('scene_season'), scene_episode=cur_ep.get('scene_episode'))
@staticmethod
def _request_from_api(cmd, params=None):
param = {'cmd': webapi._functionMaper_reversed[cmd]}
if isinstance(params, dict):
param.update(params)
return requests.get('http://127.0.0.1:%s/api/%s' % (sickgear.WEB_PORT, sickgear.API_KEYS[0][1]),
params=param).json()
@staticmethod
def _check_types(data, fields):
result, msg = True, []
missing_list, wrong_type = [], {}
for f in fields:
if f[0] not in data:
missing_list.append(f[0])
result = False
elif not isinstance(data[f[0]], f[1]):
wrong_type[f[0]] = 'Expected: %s, Got: %s' % (type(data[f[0]]), str(f[1]))
result = False
if missing_list:
msg.append('Missing fields: %s' % ', '.join(missing_list))
if wrong_type:
msg.append('Wrong field type: %s' % ', '.join(['%s: %s' % (k, v) for k, v in iteritems(wrong_type)]))
return result, ('', ', %s' % ', '.join(msg))[0 < len(msg)]
def _check_success_base_response(self, data, endpoint, message='', data_type=dict):
# type: (Any, Any, AnyStr, type) -> None
r, msg = self._check_types(data, [('result', string_types), ('data', data_type), ('message', string_types)])
self.assertTrue(r, msg='Failed command: %s%s' % (webapi._functionMaper_reversed[endpoint], msg))
self.assertEqual(data['result'], 'success')
self.assertEqual(data['message'], message)
def _check_episode_data(self, data, endpoint):
r, msg = self._check_types(
data,
[('absolute_number', integer_types), ('airdate', string_types), ('description', string_types),
('name', string_types), ('quality', string_types), ('scene_absolute_number', integer_types),
('scene_episode', integer_types), ('scene_season', integer_types), ('status', string_types),
('timezone', (NoneType, string_types))]
)
self.assertTrue(r, msg='Failed command: %s - data episode dict%s' % (
webapi._functionMaper_reversed[endpoint], msg))
def test_sg(self):
data = self._request_from_api(webapi.CMD_SickGear)
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='Failed command: %s%s' % (webapi._functionMaper_reversed[webapi.CMD_SickGear], msg))
r, msg = self._check_types(data['data'], [('api_commands', list), ('api_version', integer_types),
('fork', string_types), ('sb_version', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGear], msg))
needed_ele = [(k, string_types) for k in iterkeys(webapi._functionMaper) if 'listcommands' != k]
r = all(v[0] in data['data']['api_commands'] for v in needed_ele)
if not r:
i = list(set(n[0] for n in needed_ele) - set(data['data']['api_commands']))
else:
i = []
self.assertTrue(r, msg='Failed command: %s - api_commands list%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGear], ('', ', missing: %s' % ','.join(i))[0 < len(i)]))
def _check_show_fields(self, data, endpoint, include_season_list=True):
r, msg = self._check_types(
data,
[('air_by_date', integer_types), ('airs', string_types), ('anime', integer_types), ('cache', dict),
('classification', string_types), ('flatten_folders', integer_types), ('genre', list),
('global_exclude_ignore', string_types), ('global_exclude_require', string_types), ('ids', dict),
('ignorewords', string_types), ('imdb_id', string_types), ('indexer', integer_types),
('indexerid', integer_types), ('language', string_types), ('location', string_types),
('network', string_types), ('next_ep_airdate', string_types), ('paused', integer_types),
('prune', integer_types), ('quality', string_types), ('quality_details', dict),
('requirewords', string_types), ('runtime', integer_types), ('scenenumbering', bool),
('show_name', string_types), ('sports', integer_types), ('startyear', integer_types),
('status', string_types), ('subtitles', integer_types), ('tag', string_types),
('timezone', (NoneType, string_types)), ('tvrage_id', integer_types),
('tvrage_name', string_types), ('upgrade_once', integer_types)] +
([], [('season_list', list)])[include_season_list]
)
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[endpoint], msg))
r, msg = self._check_types(data['ids'],
[('%s' % k, integer_types) for k in sickgear.indexermapper.indexer_list])
self.assertTrue(r, msg='Failed shows "ids" check: %s' % msg)
r, msg = self._check_types(
data['quality_details'],
[('archive', list), ('initial', list)]
)
self.assertTrue(r, msg='Failed shows "quality_details" check: %s' % msg)
def test_show(self):
# test not found
data = self._request_from_api(webapi.CMD_SickGearShow, params={'indexer': 1, 'indexerid': 98765})
r, msg = self._check_types(data, [('result', string_types), ('data', dict), ('message', string_types)])
self.assertTrue(r, msg='Failed command: %s%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShow], msg))
self.assertEqual(data['result'], 'failure')
self.assertEqual(data['message'], 'Show not found')
# test existing show
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShow,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShow)
self._check_show_fields(data['data'], webapi.CMD_SickGearShow)
def test_seasons(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowSeasons,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowSeasons)
for season, eps in iteritems(cur_show['episodes']):
r, msg = self._check_types(data['data'], [('%s' % season, dict)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasons], msg))
for cur_ep in iterkeys(eps):
r, msg = self._check_types(data['data']['%s' % season], [('%s' % cur_ep, dict)])
self.assertTrue(r, msg='Failed command: %s - data season dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasons], msg))
self._check_episode_data(data['data']['%s' % season]['%s' % cur_ep], webapi.CMD_SickGearShowSeasons)
def test_coming_episodes(self):
data = self._request_from_api(webapi.CMD_SickGearComingEpisodes)
self._check_success_base_response(data, webapi.CMD_SickGearComingEpisodes)
r, msg = self._check_types(data['data'], [('later', list), ('missed', list), ('soon', list), ('today', list)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearComingEpisodes], msg))
self.assertTrue(all(0 < len(data['data'][t]) for t in ('later', 'missed', 'soon', 'today')),
msg='Not all categories returned')
for t_p in ('later', 'missed', 'soon', 'today'):
for cur_ep in data['data'][t_p]:
r, msg = self._check_types(
cur_ep,
[('airdate', string_types), ('airs', string_types), ('data_network', string_types),
('data_show_name', string_types), ('ep_name', string_types), ('ep_plot', string_types),
('episode', integer_types), ('ids', dict), ('local_datetime', string_types),
('network', string_types), ('parsed_datetime', string_types), ('paused', integer_types),
('prod_id', integer_types), ('quality', string_types), ('runtime', integer_types),
('season', integer_types), ('show_name', string_types), ('show_status', string_types),
('status', integer_types), ('status_str', string_types), ('timezone', (NoneType, string_types)),
('tv_id', integer_types), ('tvdbid', integer_types), ('weekday', integer_types)]
)
self.assertTrue(r, msg='Failed command: %s - data %s dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearComingEpisodes], t_p, msg))
r, msg = self._check_types(cur_ep['ids'],
[('%s' % k, integer_types) for k in sickgear.indexermapper.indexer_list])
self.assertTrue(r, msg='Failed %s "ids" check: %s' % (t_p, msg))
def test_all_shows(self):
data = self._request_from_api(webapi.CMD_SickGearShows)
self._check_success_base_response(data, webapi.CMD_SickGearShows)
for show_id, cur_show in iteritems(data['data']):
self._check_show_fields(cur_show, webapi.CMD_SickGearShows, include_season_list=False)
def test_episode(self):
# not found episode
data = self._request_from_api(webapi.CMD_SickGearEpisode,
params={'indexer': 1, 'indexerid': 1234, 'season': 10, 'episode': 11})
r, msg = self._check_types(data, [('result', string_types), ('data', dict), ('message', string_types)])
self.assertTrue(r, msg='Failed command: %s%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearEpisode], msg))
self.assertEqual(data['result'], 'error')
self.assertEqual(data['message'], 'Episode not found')
# found episode
for cur_show in test_shows:
for season, eps in iteritems(cur_show['episodes']):
for cur_ep in iterkeys(eps):
data = self._request_from_api(webapi.CMD_SickGearEpisode,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid'],
'season': season, 'episode': cur_ep})
self._check_success_base_response(data, webapi.CMD_SickGearEpisode)
r, msg = self._check_types(
data['data'],
[('absolute_number', integer_types), ('airdate', string_types), ('description', string_types),
('file_size', integer_types), ('file_size_human', string_types), ('location', string_types),
('name', string_types), ('quality', string_types), ('release_name', string_types),
('scene_absolute_number', (NoneType, integer_types)),
('scene_episode', (NoneType, integer_types)), ('scene_season', (NoneType, integer_types)),
('status', string_types), ('subtitles', string_types), ('timezone', (NoneType, string_types))]
)
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearEpisode], msg))
def test_shutdown(self):
data = self._request_from_api(webapi.CMD_SickGearShutdown)
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='basic test failed for shutdown')
self.assertEqual(data['message'], 'SickGear is shutting down...')
self.assertEqual(data['result'], 'success')
def test_restart(self):
data = self._request_from_api(webapi.CMD_SickGearRestart)
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='basic test failed for shutdown')
self.assertEqual(data['message'], 'SickGear is restarting...')
self.assertEqual(data['result'], 'success')
def test_ping(self):
data = self._request_from_api(webapi.CMD_SickGearPing)
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='basic test failed for shutdown')
self.assertEqual(data['message'], 'Pong')
self.assertEqual(data['result'], 'success')
def test_get_indexers(self):
data = self._request_from_api(webapi.CMD_SickGearGetIndexers)
self._check_success_base_response(data, webapi.CMD_SickGearGetIndexers)
r, msg = self._check_types(data['data'], [('%s' % k, dict) for k in sickgear.indexermapper.indexer_list])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearGetIndexers], msg))
for i in sickgear.indexermapper.indexer_list:
r, msg = self._check_types(
data['data']['%s' % i],
[('id', integer_types), ('main_url', string_types), ('name', string_types), ('searchable', bool),
('show_url', string_types)])
self.assertTrue(r, msg='Failed command: %s - data %s dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearGetIndexers], i, msg))
def test_get_seasonlist(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowSeasonList,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowSeasonList, data_type=list)
r = all(isinstance(v, integer_types) for v in data['data'])
self.assertTrue(r, msg='Failed command: %s - data dict incorrect type' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasonList]))
def test_get_episodelist(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowSeasons,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowSeasons)
r, msg = self._check_types(data['data'], [('%s' % i, dict) for i in iterkeys(cur_show['episodes'])])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasons], msg))
for season, eps in iteritems(cur_show['episodes']):
r, msg = self._check_types(data['data'], [('%s' % season, dict)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasons], msg))
for cur_ep in iterkeys(eps):
r, msg = self._check_types(data['data']['%s' % season], [('%s' % cur_ep, dict)])
self.assertTrue(r, msg='Failed command: %s - data season dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowSeasons], msg))
self._check_episode_data(data['data']['%s' % season]['%s' % cur_ep], webapi.CMD_SickGearShowSeasons)
def test_get_require_words(self):
# global
data = self._request_from_api(webapi.CMD_SickGearListRequireWords)
self._check_success_base_response(data, webapi.CMD_SickGearListRequireWords, message='Global require word list')
r, msg = self._check_types(data['data'], [('require words', list), ('type', string_types), ('use regex', bool)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearListRequireWords], msg))
# show based
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearListRequireWords,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearListRequireWords,
message='%s: require word list' % cur_show['name'])
r, msg = self._check_types(
data['data'],
[('require words', list), ('type', string_types), ('use regex', bool),
('global exclude require', list), ('indexer', integer_types), ('indexerid', integer_types),
('show name', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearListRequireWords], msg))
def test_get_ignore_words(self):
# global
data = self._request_from_api(webapi.CMD_SickGearListIgnoreWords)
self._check_success_base_response(data, webapi.CMD_SickGearListIgnoreWords, message='Global ignore word list')
r, msg = self._check_types(data['data'], [('ignore words', list), ('type', string_types), ('use regex', bool)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearListIgnoreWords], msg))
# show based
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearListIgnoreWords,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearListIgnoreWords,
message='%s: ignore word list' % cur_show['name'])
r, msg = self._check_types(
data['data'],
[('ignore words', list), ('type', string_types), ('use regex', bool),
('global exclude ignore', list), ('indexer', integer_types), ('indexerid', integer_types),
('show name', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearListIgnoreWords], msg))
def test_get_search_queue(self):
data = self._request_from_api(webapi.CMD_SickGearSearchQueue)
self._check_success_base_response(data, webapi.CMD_SickGearSearchQueue)
r, msg = self._check_types(
data['data'],
[(t, list) for t in ('backlog', 'failed', 'manual', 'proper')] + [('recent', integer_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearSearchQueue], msg))
def test_get_system_default(self):
data = self._request_from_api(webapi.CMD_SickGearGetDefaults)
self._check_success_base_response(data, webapi.CMD_SickGearGetDefaults)
r, msg = self._check_types(
data['data'],
[('archive', list), ('flatten_folders', integer_types), ('future_show_paused', integer_types),
('initial', list), ('status', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearGetDefaults], msg))
def test_get_all_qualities(self):
data = self._request_from_api(webapi.CMD_SickGearGetQualities)
self._check_success_base_response(data, webapi.CMD_SickGearGetQualities)
r, msg = self._check_types(data['data'], [(q, integer_types) for q in iterkeys(webapi.quality_map)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearGetQualities], msg))
def test_get_human_qualities(self):
data = self._request_from_api(webapi.CMD_SickGearGetqualityStrings)
self._check_success_base_response(data, webapi.CMD_SickGearGetqualityStrings)
r, msg = self._check_types(data['data'], [('%s' % q, string_types) for q in iterkeys(Quality.qualityStrings)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearGetqualityStrings], msg))
def test_get_scene_qualities(self):
# global
data = self._request_from_api(webapi.CMD_SickGearExceptions)
self._check_success_base_response(data, webapi.CMD_SickGearExceptions)
r = all(isinstance(e, string_types) for e in data['data'])
self.assertTrue(r, msg='Failed command: %s - data dict' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearExceptions]))
# show specific
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearExceptions,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearExceptions, data_type=list)
r = all(isinstance(e, string_types) for e in data['data'])
self.assertTrue(r, msg='Failed command: %s - data dict' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearExceptions]))
def test_history(self):
data = self._request_from_api(webapi.CMD_SickGearHistory)
self._check_success_base_response(data, webapi.CMD_SickGearHistory, data_type=list)
for cur_show in data['data']:
self.assertTrue(isinstance(cur_show, dict), msg='wrong type')
r, msg = self._check_types(
cur_show,
[('date', string_types), ('episode', integer_types), ('indexer', integer_types),
('indexerid', integer_types), ('provider', string_types), ('quality', string_types),
('resource', string_types), ('resource_path', string_types), ('season', integer_types),
('show_name', string_types), ('status', string_types), ('tvdbid', integer_types),
('version', integer_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearHistory], msg))
def test_shows_stats(self):
data = self._request_from_api(webapi.CMD_SickGearShowsStats)
self._check_success_base_response(data, webapi.CMD_SickGearShowsStats)
r, msg = self._check_types(
data['data'],
[('ep_downloaded', integer_types), ('ep_total', integer_types), ('shows_active', integer_types),
('shows_total', integer_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowsStats], msg))
def test_show_stats(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowStats,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowStats)
r, msg = self._check_types(
data['data'],
[(statusStrings.statusStrings[status].lower().replace(" ", "_").replace("(", "").replace(
")", ""), integer_types) for status in statusStrings.statusStrings
if status not in SNATCHED_ANY + [UNKNOWN, DOWNLOADED]] + [('downloaded', dict), ('snatched', dict)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowStats], msg))
for s in ('downloaded', 'snatched'):
r, msg = self._check_types(
data['data'][s],
[(t.lower().replace(" ", "_").replace("(", "").replace(")", ""), integer_types)
for k, t in iteritems(Quality.qualityStrings) if Quality.NONE != k])
self.assertTrue(r, msg='Failed command: %s - data dict - %s:%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowStats], s, msg))
def test_get_root_dirs(self):
data = self._request_from_api(webapi.CMD_SickGearGetRootDirs)
self._check_success_base_response(data, webapi.CMD_SickGearGetRootDirs, data_type=list)
for r_d in data['data']:
r, msg = self._check_types(r_d, [('default', integer_types), ('location', string_types),
('valid', integer_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearListIgnoreWords], msg))
def test_get_schedules(self):
data = self._request_from_api(webapi.CMD_SickGearCheckScheduler)
self._check_success_base_response(data, webapi.CMD_SickGearCheckScheduler)
r, msg = self._check_types(
data['data'],
[('backlog_is_paused', integer_types), ('backlog_is_running', integer_types),
('last_backlog', string_types), ('next_backlog', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearCheckScheduler], msg))
def test_do_show_update(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowUpdate,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowUpdate,
message='%s has queued to be updated' % cur_show['name'])
# check that duplicate adding fails
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowUpdate,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowUpdate], msg))
self.assertTrue('Unable to update %s.' % cur_show['name'] in data['message'], msg='Wrong failure message')
def test_do_show_refresh(self):
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowRefresh,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
self._check_success_base_response(data, webapi.CMD_SickGearShowUpdate,
message='%s has queued to be refreshed' % cur_show['name'])
# check that duplicate adding fails
for cur_show in test_shows:
data = self._request_from_api(webapi.CMD_SickGearShowRefresh,
params={'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']})
r, msg = self._check_types(data, [('data', dict), ('message', string_types), ('result', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearShowRefresh], msg))
self.assertTrue('Unable to refresh %s.' % cur_show['name'] in data['message'], msg='Wrong failure message')
def test_pause_show(self):
self.reset_show_data = True
self.show_save_db = sickgear.tv.TVShow.save_to_db
sickgear.tv.TVShow.save_to_db = fake_action
for set_pause in (None, 0, 1, 0):
for cur_show in test_shows:
params = {'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']}
if None is not set_pause:
params.update({'pause': set_pause})
data = self._request_from_api(webapi.CMD_SickGearShowPause, params=params)
self._check_success_base_response(data, webapi.CMD_SickGearShowPause,
message='%s has been %spaused' % (
cur_show['name'], ('', 'un')[set_pause in (None, 0, False)]))
def test_set_scene_numbering(self):
self.reset_show_data = True
self.show_save_db = sickgear.tv.TVShow.save_to_db
sickgear.tv.TVShow.save_to_db = fake_action
for set_scene in (1, 0):
for cur_show in test_shows:
params = {'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']}
if None is not set_scene:
params.update({'activate': set_scene})
data = self._request_from_api(webapi.CMD_SickGearActivateSceneNumber, params=params)
self._check_success_base_response(data, webapi.CMD_SickGearActivateSceneNumber,
message='Scene Numbering %sactivated' % (
('', 'de')[set_scene in (None, 0, False)]))
r, msg = self._check_types(
data['data'],
[('indexer', integer_types), ('indexerid', integer_types), ('scenenumbering', bool),
('show_name', string_types)])
self.assertTrue(r, msg='Failed command: %s - data dict%s' % (
webapi._functionMaper_reversed[webapi.CMD_SickGearActivateSceneNumber], msg))
self.assertTrue(data['data']['scenenumbering'] == bool(set_scene))
self.assertTrue(data['data']['show_name'] == cur_show['name'])
def test_set_show_quality(self):
self.reset_show_data = True
self.show_save_db = sickgear.tv.TVShow.save_to_db
sickgear.tv.TVShow.save_to_db = fake_action
for set_quality in [
{'init': [Quality.SDTV],
'upgrade': [],
'upgrade_once': 0},
{'init': [Quality.SDTV],
'upgrade': [],
'upgrade_once': 1},
{'init': [Quality.SDTV],
'upgrade': [Quality.FULLHDWEBDL, Quality.FULLHDBLURAY],
'upgrade_once': 0},
{'init': [Quality.SDTV],
'upgrade': [Quality.FULLHDWEBDL, Quality.FULLHDBLURAY],
'upgrade_once': 1},
{'init': [Quality.SDTV, Quality.SDDVD, Quality.HDTV],
'upgrade': [],
'upgrade_once': 0},
{'init': [Quality.SDTV, Quality.SDDVD, Quality.HDTV],
'upgrade': [],
'upgrade_once': 1},
{'init': [Quality.SDTV, Quality.SDDVD, Quality.HDTV],
'upgrade': [Quality.SDDVD, Quality.HDWEBDL],
'upgrade_once': 0},
{'init': [Quality.SDTV, Quality.SDDVD, Quality.HDTV],
'upgrade': [Quality.SDDVD, Quality.HDWEBDL],
'upgrade_once': 1},
]:
for cur_show in test_shows:
show_obj = sickgear.helpers.find_show_by_id({cur_show['tvid']: cur_show['prodid']})
params = {'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']}
for t in ('init', 'upgrade', 'upgrade_once'):
if set_quality[t]:
params.update({t: set_quality[t]})
data = self._request_from_api(webapi.CMD_SickGearShowSetQuality, params=params)
self._check_success_base_response(
data,
webapi.CMD_SickGearShowSetQuality,
message='%s quality has been changed to %s' % (
cur_show['name'], webapi._get_quality_string(show_obj.quality)))
self.assertEqual(show_obj.upgrade_once, int(set_quality['upgrade_once']))
def test_set_show_scene_numbers(self):
self.reset_show_data = True
self.show_save_db = sickgear.tv.TVShow.save_to_db
sickgear.tv.TVShow.save_to_db = fake_action
for set_numbers in [
{'forSeason': 1,
'forEpisode': 1,
'forAbsolute': None,
'sceneSeason': 1,
'sceneEpisode': 2,
'sceneAbsolute': None,
},
]:
for cur_show in test_shows:
params = {'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid']}
set_scene_params = params.copy()
set_scene_params.update({'activate': 1})
data = self._request_from_api(webapi.CMD_SickGearActivateSceneNumber,
params=set_scene_params)
self._check_success_base_response(data, webapi.CMD_SickGearActivateSceneNumber,
message='Scene Numbering activated')
for t in ('forSeason', 'forEpisode', 'forAbsolute', 'sceneSeason', 'sceneEpisode', 'sceneAbsolute'):
if set_numbers[t]:
params.update({t: set_numbers[t]})
data = self._request_from_api(webapi.CMD_SickGearSetSceneNumber, params=params)
self._check_success_base_response(data, webapi.CMD_SickGearSetSceneNumber)
self._check_types(
data['data'],
[(t, integer_types) for t in ('forSeason', 'forEpisode', 'sceneSeason', 'sceneEpisode')] +
[('success', bool)])
for t in ('forSeason', 'forEpisode', 'sceneSeason', 'sceneEpisode'):
self.assertEqual(data['data'][t], set_numbers[t])
def test_set_episode_status(self):
self.org_mass_action = db.DBConnection.mass_action
db.DBConnection.mass_action = fake_action
self.reset_show_data = True
failed_msg = 'Failed to set all or some status. Check data.'
success_msg = 'All status set successfully.'
for cur_quality_str, cur_quality in itertools.chain(iteritems(webapi.quality_map), iteritems({'None': None})):
for cur_value, cur_status in iteritems(statusStrings.statusStrings):
if (cur_quality and cur_value not in (SNATCHED, DOWNLOADED, ARCHIVED)) or \
(None is cur_quality and SNATCHED == cur_value):
continue
cur_status = cur_status.lower()
# print('Testing setting episode status to: %s %s' % (cur_status, cur_quality_str))
if cur_value in (UNKNOWN, UNAIRED, SNATCHED_PROPER, SNATCHED_BEST, SUBTITLED):
continue
for cur_show in test_shows:
for season, eps in iteritems(cur_show['episodes']):
for ep_nb, cur_ep in iteritems(eps):
ep_obj = sickgear.helpers.find_show_by_id({cur_show['tvid']: cur_show['prodid']}).\
get_episode(season, ep_nb)
params = {'indexer': cur_show['tvid'], 'indexerid': cur_show['prodid'], 'season': season,
'episode': ep_nb, 'status': cur_status}
if cur_quality:
params.update({'quality': cur_quality_str})
old_status = ep_obj.status
status, quality = Quality.split_composite_status(ep_obj.status)
expect_fail = UNAIRED == status or (DOWNLOADED == status and not cur_quality)
expected_msg = (success_msg, failed_msg)[expect_fail]
data = self._request_from_api(webapi.CMD_SickGearEpisodeSetStatus, params=params)
r, msg = self._check_types(data,
[('result', string_types), ('data', (dict, list)[expect_fail]),
('message', string_types)])
self.assertTrue(r, msg=msg)
self.assertTrue(data['message'].startswith(expected_msg))
# reset status
ep_obj.status = old_status
# ep_obj.save_to_db()
if __name__ == '__main__':
print('==================')
print('STARTING - WebAPI TESTS')
print('==================')
print('######################################################################')
suite = unittest.TestLoader().loadTestsFromTestCase(WebAPICase)
unittest.TextTestRunner(verbosity=2).run(suite)
Reference in a new issue View git blame Copy permalink