Files
agents/tests/venue/playlists.py
T
2026-10-31 03:18:19 +07:00

179 lines
14 KiB
Python

"""Build the frontend, then run: python3 tests/venue/playlists.py. Requires Firefox."""
import base64, json, mimetypes, pathlib, socket, subprocess, tempfile, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
DIST=pathlib.Path(__file__).resolve().parents[2]/'evening_detective_frontend/dist'
state={'playlists':[], 'calls':[], 'roles':['organizer'], 'failList':False, 'failGroup':False, 'failUpload':False}
class Handler(BaseHTTPRequestHandler):
def log_message(self,*args):pass
def respond(self,data):
self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(json.dumps(data,ensure_ascii=False).encode())
def do_GET(self):self.handle_request()
def do_POST(self):self.handle_request()
def do_PUT(self):self.handle_request()
def do_DELETE(self):self.handle_request()
def handle_request(self):
path=urlparse(self.path).path
if path.startswith('/__mock'):
path=path.removeprefix('/__mock');payload=json.loads(self.rfile.read(int(self.headers.get('Content-Length',0))) or '{}')
if self.command!='GET':state['calls'].append((path,payload))
if path=='/api/users/me':return self.respond({'user':{'ID':1,'username':'Тест','roles':state['roles']}})
if path=='/api/ui/permissions':return self.respond({'permissions':['games_page'],'environment':'prod'})
if path=='/api/playlists':
if self.command=='GET':return self.respond({'code':13,'message':'Ошибка списка'} if state['failList'] else {'playlists':state['playlists']})
p={'ID':max((p['ID'] for p in state['playlists']),default=0)+1,'name':payload['name'],'groups':[],'tracks':[]};state['playlists'].append(p)
return self.respond({'playlist':p})
if path.startswith('/api/playlists/'):
ident=int(path.split('/')[3]);p=next(p for p in state['playlists'] if p['ID']==ident)
if path.endswith('/tracks') and self.command=='POST':
assert payload['data'] and payload['filename'].endswith('.mp3')
if state['failUpload']:return self.respond({'code':9,'message':'Группа используется'})
if payload.get('groupID') and not any(g['ID']==payload['groupID'] for g in p['groups']):return self.respond({'code':5,'message':'Группа не найдена'})
p['tracks'].append({'ID':max((t['ID'] for t in p['tracks']),default=0)+1,'name':payload['name'],'filename':payload['filename'],'groupId':payload.get('groupID',0)})
elif '/tracks/' in path and self.command=='DELETE':p['tracks']=[t for t in p['tracks'] if t['ID']!=int(path.split('/')[-1])]
elif len(path.split('/'))==4 and self.command=='DELETE':state['playlists'].remove(p)
elif path.endswith('/groups/order'):
p['groups'].sort(key=lambda g:payload['groupIDs'].index(g['ID']))
elif path.endswith('/groups'):
if state['failGroup']:return self.respond({'code':13,'message':'Тест: группа не сохранена'})
if payload.get('groupID'):
g=next(g for g in p['groups'] if g['ID']==payload['groupID']);g.update(name=payload['name'])
else:
g={'ID':max((g['ID'] for g in p['groups']),default=0)+1,'name':payload['name'],'position':len(p['groups'])};p['groups'].append(g)
if payload.get('endMode') and payload['endMode']!='PLAYLIST_GROUP_END_MODE_REPEAT':g['endMode']=payload['endMode']
else:g.pop('endMode',None)
elif path.endswith('/group'):
t=next(t for t in p['tracks'] if t['ID']==int(path.split('/')[5]));t['groupId']=payload.get('groupID',0)
elif '/groups/' in path and self.command=='DELETE':
groupID=int(path.split('/')[-1]);p['groups']=[g for g in p['groups'] if g['ID']!=groupID]
for t in p['tracks']:
if t.get('groupId')==groupID:t['groupId']=0
elif 'trackIDs' in payload:
p['tracks'].sort(key=lambda t:payload['trackIDs'].index(t['ID']))
elif 'name' in payload:p['name']=payload['name']
return self.respond({'playlist':p})
return self.respond({})
f=DIST/path.lstrip('/')
if f.is_file():body=f.read_bytes();mime=mimetypes.guess_type(f.name)[0] or 'application/octet-stream'
else:
inject='''<script>localStorage.setItem('accessToken','fixture');localStorage.setItem('refreshToken','fixture');window.confirm=()=>window.allowDiscard!==false;window.testErrors=[];addEventListener('error',e=>testErrors.push(e.message));addEventListener('unhandledrejection',e=>testErrors.push(String(e.reason)));const originalFetch=fetch.bind(window);window.fetch=(u,o)=>{const p=new URL(u,location.href);return originalFetch(p.pathname.startsWith('/api/')?'/__mock'+p.pathname+p.search:u,o)};</script>'''
body=(DIST/'index.html').read_text().replace('<head>','<head>'+inject).encode();mime='text/html'
self.send_response(200);self.send_header('Content-Type',mime);self.end_headers();self.wfile.write(body)
server=ThreadingHTTPServer(('127.0.0.1',0),Handler);threading.Thread(target=server.serve_forever,daemon=True).start()
profile=pathlib.Path(tempfile.mkdtemp(prefix='ed059-playlists-'))
with socket.socket() as s:s.bind(('127.0.0.1',0));port=s.getsockname()[1]
(profile/'user.js').write_text(f'user_pref("marionette.port", {port});\nuser_pref("browser.shell.checkDefaultBrowser", false);')
log=(profile/'firefox.log').open('w');firefox=subprocess.Popen(['firefox','--headless','--no-remote','--profile',str(profile),'--marionette','--remote-allow-system-access','about:blank'],stdout=log,stderr=log);sock=None
try:
for _ in range(100):
try:sock=socket.create_connection(('127.0.0.1',port),timeout=1);break
except OSError:time.sleep(.1)
assert sock is not None,(profile/'firefox.log').read_text()
sock.settimeout(20)
def read():
size=b''
while not size.endswith(b':'):size+=sock.recv(1)
data=b''
while len(data)<int(size[:-1]):data+=sock.recv(int(size[:-1])-len(data))
return json.loads(data)
read();counter=0
def command(name,args=None):
global counter
counter+=1;p=json.dumps([0,counter,name,args or {}]).encode();sock.sendall(str(len(p)).encode()+b':'+p);r=read();assert not r[2],r[2];return r[3]
def js(script):
r=command('WebDriver:ExecuteScript',{'script':script,'args':[],'newSandbox':False,'sandbox':None});return r.get('value') if isinstance(r,dict) and 'value' in r else r
def wait(condition):
for _ in range(100):
if js('return '+condition):return
time.sleep(.1)
raise AssertionError(condition+'; body='+str(js('return document.body.innerText')))
def click(label,scope='document'):
code=f"const b=[...{scope}.querySelectorAll('button')].find(b=>b.textContent.trim()==={json.dumps(label)}||b.getAttribute('aria-label')==={json.dumps(label)});if(!b||b.disabled)return false;b.click();return true"
wait(f"[...{scope}.querySelectorAll('button')].some(b=>(b.textContent.trim()==={json.dumps(label)}||b.getAttribute('aria-label')==={json.dumps(label)})&&!b.disabled)")
assert js(code),label
time.sleep(.15)
def input_value(selector,value):
assert js(f'const e=document.querySelector({json.dumps(selector)});if(!e)return false;e.value={json.dumps(value)};e.dispatchEvent(new Event("input",{{bubbles:true}}));return true')
command('WebDriver:NewSession',{'capabilities':{'alwaysMatch':{}}});command('Marionette:SetContext',{'value':'chrome'});js('document.documentElement.style.minWidth="0";document.documentElement.setAttribute("minwidth","0");return true');command('Marionette:SetContext',{'value':'content'})
def choose(selector,label):
assert js(f'const e=document.querySelector({json.dumps(selector)});if(!e)return false;e.click();return true'),selector
wait('!!document.querySelector(".n-base-select-option")')
assert js(f'const e=[...document.querySelectorAll(".n-base-select-option")].find(e=>e.textContent.trim()==={json.dumps(label)});if(!e)return false;e.click();return true'),label
def navigate():
command('WebDriver:Navigate',{'url':f'http://127.0.0.1:{server.server_port}/playlists'})
def upload(name, content="fake-mp3", group="Без группы"):
assert js(f'const e=document.querySelector({json.dumps('section[aria-label="'+group+'"] input[type=file]')});const d=new DataTransfer();d.items.add(new File([{json.dumps(content)}],{json.dumps(name)},{{type:"audio/mpeg"}}));e.files=d.files;e.dispatchEvent(new Event("change",{{bubbles:true}}));return true')
command('WebDriver:SetWindowRect',{'width':1280,'height':1000})
state['failList']=True
navigate();wait('document.body.innerText.includes("Ошибка списка")')
state['failList']=False;click('Повторить загрузку')
wait('document.body.innerText.includes("Плейлистов пока нет")')
assert not js('return document.body.innerText.includes("Использовать в игре")')
input_value('input[placeholder="Новый плейлист"]','Первый')
click('Создать');wait('!!document.querySelector("#music-playlist-name")')
wait('document.querySelector("#music-playlist-name").value==="Первый"')
wait(r'document.querySelector("input[placeholder=\"Новый плейлист\"]").value===""')
input_value('#music-playlist-name','Переименованный');click('Сохранить название')
wait('!document.querySelector("#music-playlist-name").disabled')
assert state['playlists'][0]['name']=='Переименованный'
input_value('input[placeholder="Новая группа"]','Группа')
state['failGroup']=True;click('Добавить группу');wait('document.body.innerText.includes("Тест: группа не сохранена")')
assert js(r'return document.querySelector("input[placeholder=\"Новая группа\"]").value')=='Группа'
state['failGroup']=False;click('Добавить группу');wait('!!document.querySelector("section[aria-label=Группа]")')
wait('!document.querySelector("#music-playlist-name").disabled')
upload('empty.mp3','');wait('document.body.innerText.includes("Выберите непустой MP3 размером до 64 МиБ")')
assert not state['playlists'][0]['tracks']
input_value('#music-playlist-name','Черновик загрузки')
state['failUpload']=True;upload('rejected.mp3',group='Группа');wait('document.body.innerText.includes("Группа используется")')
assert js('return document.querySelector("#music-playlist-name").value')=='Черновик загрузки'
assert not state['playlists'][0]['tracks']
state['failUpload']=False;input_value('#music-playlist-name','Переименованный')
upload('grouped.mp3',group='Группа');wait('document.querySelectorAll(".track-row").length===1')
assert state['playlists'][0]['tracks'][0]['groupId']==1
assert not any(path.endswith('/group') for path,payload in state['calls']), 'upload must assign the group in one request'
wait('!document.querySelector("#music-playlist-name").disabled')
upload('first.mp3');wait('document.querySelectorAll(".track-row").length===2')
wait('!document.querySelector("#music-playlist-name").disabled')
upload('second.mp3');wait('document.querySelectorAll(".track-row").length===3')
wait('!document.querySelector("#music-playlist-name").disabled')
click('Поднять трек 2 в группе Без группы');click('Сохранить порядок')
wait('!document.querySelector("#music-playlist-name").disabled')
assert [t['name'] for t in state['playlists'][0]['tracks'] if not t['groupId']]==['second.mp3','first.mp3']
choose('section[aria-label="Без группы"] .track-row .n-base-selection','Группа')
wait('document.querySelector("section[aria-label=Группа] .track-row")!==null')
wait('!document.querySelector("#music-playlist-name").disabled')
assert state['playlists'][0]['tracks'][1]['groupId']==1
input_value('#music-playlist-name','Черновик')
js('window.allowDiscard=false;return true')
click('Главная');time.sleep(.2)
assert js('return location.pathname')=='/playlists'
assert js('return document.querySelector("#music-playlist-name").value')=='Черновик'
state['failList']=True
# A fresh visit tests the initial load failure again without altering production data.
js('window.allowDiscard=true;return true')
navigate();wait('document.body.innerText.includes("Ошибка списка")')
state['failList']=False;click('Повторить загрузку');wait('!!document.querySelector("#music-playlist-name")')
for width in [1280,375]:
command('WebDriver:SetWindowRect',{'width':width,'height':1000});time.sleep(.3)
assert js('return document.documentElement.scrollWidth<=innerWidth'), 'horizontal overflow'
image=command('WebDriver:TakeScreenshot',{})['value'];pathlib.Path(f'/tmp/ed059-playlists-{width}.png').write_bytes(base64.b64decode(image))
js('document.querySelector("#music-playlist-name").focus();return true')
command('WebDriver:PerformActions',{'actions':[{'type':'key','id':'keyboard','actions':[{'type':'keyDown','value':'\ue004'},{'type':'keyUp','value':'\ue004'}]}]})
assert js('return document.activeElement.tagName')=='BUTTON'
click('Удалить', 'document.querySelector(".track-row")');click('Подтвердить')
wait('document.querySelectorAll(".track-row").length===2')
click('Удалить группу');click('Подтвердить');wait('!document.querySelector("section[aria-label=Группа]")')
click('Удалить плейлист');click('Подтвердить');wait('document.body.innerText.includes("Плейлистов пока нет")')
state['roles']=['author'];navigate();wait('!!document.querySelector("[data-testid=music-editor]")')
state['roles']=['user'];navigate();wait('document.body.innerText.includes("Нет доступа к плейлистам")')
assert not js('return !!document.querySelector("[data-testid=music-editor]")')
assert js('return testErrors')==[],js('return testErrors')
print('PLAYLISTS_PASS: direct group upload and rejection, load error/retry, empty/create/select/rename, group error/retry, upload/order/membership/delete, dirty route guard, author/organizer access, keyboard, desktop/mobile')
finally:
if sock:sock.close()
firefox.terminate()
try:firefox.wait(timeout=10)
except subprocess.TimeoutExpired:firefox.kill();firefox.wait()
log.close();server.shutdown();server.server_close()