52 lines
3.9 KiB
Python
52 lines
3.9 KiB
Python
"""Integration checks in CT100191. Creates and always removes two test users."""
|
|
import datetime,json,secrets,sqlite3,urllib.error,urllib.parse,urllib.request
|
|
db=sqlite3.connect('file:/var/lib/jellyfin/data/jellyfin.db?mode=ro',uri=True)
|
|
admin=db.execute('SELECT AccessToken FROM ApiKeys ORDER BY Id LIMIT 1').fetchone()[0];db.close()
|
|
def api(path,data=None,method=None,token=admin):
|
|
headers={'Authorization':'MediaBrowser Token="'+token+'", Client="DecoderIntegration", Device="Maintenance", DeviceId="decoder-maintenance", Version="7"','Content-Type':'application/json','X-Emby-Authorization':'MediaBrowser Client="DecoderVerification", Device="IntegrationCheck", DeviceId="decoder-plugin-check", Version="1"'}
|
|
req=urllib.request.Request('http://127.0.0.1:8096'+path,data=None if data is None else json.dumps(data).encode(),headers=headers,method=method)
|
|
with urllib.request.urlopen(req,timeout=90) as response:
|
|
body=response.read();return json.loads(body) if body else None
|
|
report={'checked_utc':datetime.datetime.now(datetime.timezone.utc).isoformat()}
|
|
report['plugin']=next(x for x in api('/Plugins') if x['Name']=='Enigma2')
|
|
assert report['plugin']['Version']=='7.0.6.0' and report['plugin']['Status']=='Active'
|
|
report['epg']=api('/DecoderIntegration/Status')
|
|
report['channel_count']=api('/LiveTv/Channels?Limit=0')['TotalRecordCount']
|
|
report['timer_count']=api('/LiveTv/Timers?Limit=0')['TotalRecordCount']
|
|
users=[]
|
|
try:
|
|
tokens=[]
|
|
for i in range(2):
|
|
name='decoder-plugin-check-'+secrets.token_hex(4);password=secrets.token_urlsafe(24)
|
|
user=api('/Users/New',{'Name':name});users.append(user['Id'])
|
|
api('/Users/'+user['Id']+'/Password',{'CurrentPw':'','NewPw':password})
|
|
login=api('/Users/AuthenticateByName',{'Username':name,'Pw':password})
|
|
tokens.append(login['AccessToken'])
|
|
for token in tokens: assert api('/DecoderIntegration/Preferences',token=token)['DefaultView']=='all'
|
|
channel=api('/LiveTv/Channels?Limit=1')['Items'][0]['Id']
|
|
api('/DecoderIntegration/Preferences',{'FavoriteIds':[channel],'HiddenIds':[],'OrderedIds':[channel],'DefaultView':'favorites'},'PUT',tokens[0])
|
|
assert api('/DecoderIntegration/Preferences',token=tokens[0])['FavoriteIds']==[channel.replace('-','')]
|
|
assert api('/DecoderIntegration/Preferences',token=tokens[1])['FavoriteIds']==[]
|
|
report['per_user_preferences']='PASS'
|
|
all_channels=api('/LiveTv/Channels?Limit=2')['Items'];first,second=[c['Id'] for c in all_channels]
|
|
api('/DecoderIntegration/Preferences',{'HiddenIds':[first],'OrderedIds':[second,first]},'PUT',tokens[0])
|
|
native=api('/LiveTv/Channels?UserId='+users[0]+'&Limit=1',token=tokens[0])
|
|
assert native['TotalRecordCount']==report['channel_count']-1 and native['Items'][0]['Id']==second
|
|
assert api('/LiveTv/Channels?UserId='+users[0]+'&Limit=0',token=tokens[0])['Items']==[]
|
|
assert api('/LiveTv/Channels?UserId='+users[1]+'&Limit=0',token=tokens[1])['TotalRecordCount']==report['channel_count']
|
|
assert api('/LiveTv/Channels?UserId='+users[0]+'&DecoderIncludeHidden=true&Limit=0',token=tokens[0])['TotalRecordCount']==report['channel_count']
|
|
report['native_channel_preferences_and_pagination']='PASS'
|
|
try:api('/DecoderIntegration/Status',token=tokens[1]);raise AssertionError('non-admin status permitted')
|
|
except urllib.error.HTTPError as ex:assert ex.code==403
|
|
report['admin_status_policy']='PASS'
|
|
for token in tokens:
|
|
req=urllib.request.Request('http://127.0.0.1:8096/web/ConfigurationPage?name=decoder-channels',headers={'Authorization':'MediaBrowser Token="'+token+'", Client="DecoderIntegration", Device="Maintenance", DeviceId="decoder-maintenance", Version="7"'})
|
|
text=urllib.request.urlopen(req).read().decode();assert 'channelRows' in text
|
|
report['user_channel_page']='PASS'
|
|
finally:
|
|
for user in users:
|
|
api('/Users/'+user,method='DELETE')
|
|
from pathlib import Path
|
|
Path('/var/lib/jellyfin/decoder-integration/users/'+user.replace('-','')+'.json').unlink(missing_ok=True)
|
|
print(json.dumps(report,ensure_ascii=False))
|