test(http): exercise OSE through a running server
Test web and DSi responses, permissions, moderation, settings and multipart logo uploads over HTTP.
This commit is contained in:
rekushi committed
1 parent
95168fb1a5
commit
425da88711
1 file changed
+159
@@ -0,0 +1,159 @@
|
|||||||
|
"""Run the real proxy and browser routes against a disposable installation."""
|
||||||
|
from pathlib import Path
|
||||||
|
from tempfile import TemporaryDirectory
|
||||||
|
from http.client import HTTPConnection
|
||||||
|
from urllib.parse import urlencode
|
||||||
|
from contextlib import closing
|
||||||
|
import json
|
||||||
|
import gc
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
import socket
|
||||||
|
import sqlite3
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
|
||||||
|
ROOT = Path(__file__).resolve().parents[1]
|
||||||
|
|
||||||
|
|
||||||
|
class HttpIntegrationTests(unittest.TestCase):
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
cls.temporary = TemporaryDirectory()
|
||||||
|
cls.directory = Path(cls.temporary.name) / 'ose'
|
||||||
|
shutil.copytree(ROOT, cls.directory, ignore=shutil.ignore_patterns('__pycache__', '.venv', 'logs', 'backups', 'tests', '*.sqlite3*', 'server-settings.json'))
|
||||||
|
source = sqlite3.connect((ROOT/'database/reximemo.sqlite3').as_uri()+'?mode=ro', uri=True)
|
||||||
|
destination = sqlite3.connect(cls.directory/'database/reximemo.sqlite3')
|
||||||
|
source.backup(destination)
|
||||||
|
destination.execute('DELETE FROM ip_logins')
|
||||||
|
destination.commit()
|
||||||
|
destination.close()
|
||||||
|
source.close()
|
||||||
|
with socket.socket() as probe:
|
||||||
|
probe.bind(('127.0.0.1',0))
|
||||||
|
cls.port = probe.getsockname()[1]
|
||||||
|
config = {'site_name':'OSE HTTP Test','host':'127.0.0.1','port':cls.port}
|
||||||
|
(cls.directory/'server-settings.json').write_text(json.dumps(config), encoding='utf-8')
|
||||||
|
env = dict(os.environ, REXIMEMO_PORT=str(cls.port), REXIMEMO_HOST='127.0.0.1',
|
||||||
|
REXIMEMO_DB=str(cls.directory/'database/reximemo.sqlite3'),
|
||||||
|
REXIMEMO_CONFIG=str(cls.directory/'server-settings.json'), PYTHONUTF8='1')
|
||||||
|
cls.log = (cls.directory/'http-test.log').open('w+',encoding='utf-8')
|
||||||
|
cls.process = subprocess.Popen([sys.executable,'-u','server.py','docs:enabled'],cwd=cls.directory,env=env,stdout=cls.log,stderr=subprocess.STDOUT)
|
||||||
|
try:
|
||||||
|
for attempt in range(100):
|
||||||
|
if cls.process.poll() is not None:
|
||||||
|
cls.log.seek(0)
|
||||||
|
raise RuntimeError(cls.log.read())
|
||||||
|
try:
|
||||||
|
if cls.request('GET','/')[0] == 200:
|
||||||
|
break
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
time.sleep(.1)
|
||||||
|
else:
|
||||||
|
raise RuntimeError('The test server did not start.')
|
||||||
|
except Exception:
|
||||||
|
cls.tearDownClass()
|
||||||
|
raise
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownClass(cls):
|
||||||
|
if cls.process.poll() is None:
|
||||||
|
cls.process.terminate()
|
||||||
|
cls.process.wait(timeout=10)
|
||||||
|
cls.log.close()
|
||||||
|
# Windows can retain SQLite's shared-memory mapping briefly after exit.
|
||||||
|
gc.collect()
|
||||||
|
for attempt in range(20):
|
||||||
|
try:
|
||||||
|
cls.temporary.cleanup()
|
||||||
|
break
|
||||||
|
except (PermissionError, NotADirectoryError):
|
||||||
|
if attempt == 19:
|
||||||
|
raise
|
||||||
|
time.sleep(.1)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def request(cls, method, path, fields=None, *, cookie='', dsi=False, raw=None, content_type=None):
|
||||||
|
connection=HTTPConnection('127.0.0.1',cls.port,timeout=10)
|
||||||
|
headers={'Host':'flipnote.hatena.com' if dsi else 'localhost','Cookie':cookie}
|
||||||
|
if dsi:
|
||||||
|
headers['X-DSi-SID']='test-transport'
|
||||||
|
body = urlencode(fields).encode() if fields is not None else raw
|
||||||
|
if body is not None:
|
||||||
|
headers['Content-Type']=content_type or 'application/x-www-form-urlencoded'
|
||||||
|
connection.request(method,path,body=body,headers=headers)
|
||||||
|
response=connection.getresponse()
|
||||||
|
result=response.status,dict(response.getheaders()),response.read()
|
||||||
|
connection.close()
|
||||||
|
return result
|
||||||
|
|
||||||
|
def login(self):
|
||||||
|
status,headers,body=self.request('GET','/login')
|
||||||
|
self.assertEqual(status,200)
|
||||||
|
token=re.search(rb'name="csrf" value="([a-f0-9]{64})"',body)[1].decode()
|
||||||
|
cookie=headers['Set-Cookie'].split(';',1)[0]
|
||||||
|
result=self.request('POST','/login',{'username':'reximemo','password':'alpine','csrf':token},cookie=cookie)
|
||||||
|
self.assertEqual(result[0],303)
|
||||||
|
return token,cookie
|
||||||
|
|
||||||
|
def test_web_dsi_and_management_end_to_end(self):
|
||||||
|
for path in ('/','/browse?page=2','/search?q=Sample','/channels','/docs'):
|
||||||
|
self.assertEqual(self.request('GET',path)[0],200,path)
|
||||||
|
for region in ('xx','eu','us','jp'):
|
||||||
|
status,_,menu=self.request('GET',f'/ds/v2-{region}/index.uls',dsi=True)
|
||||||
|
self.assertEqual(status,200)
|
||||||
|
self.assertTrue(menu.startswith(b'UGAR'))
|
||||||
|
self.assertEqual(self.request('GET','/admin')[0],403)
|
||||||
|
token,cookie=self.login()
|
||||||
|
self.assertEqual(self.request('POST','/admin/channel-create',{'title':'Rejected'})[0],403)
|
||||||
|
for tab in ('flipnotes','comments','channels','server','bans'):
|
||||||
|
self.assertEqual(self.request('GET','/admin?tab='+tab,cookie=cookie)[0],200,tab)
|
||||||
|
created=self.request('POST','/admin/channel-create',{'csrf':token,'title':'Music','description':'Songs and animation'},cookie=cookie)
|
||||||
|
self.assertEqual(created[0],303)
|
||||||
|
with closing(sqlite3.connect(self.directory/'database/reximemo.sqlite3')) as db:
|
||||||
|
channel=db.execute("SELECT id FROM channels WHERE title='Music'").fetchone()[0]
|
||||||
|
note=db.execute("SELECT id,creator_fsid,filename FROM flipnotes WHERE public_id='f_sample1'").fetchone()
|
||||||
|
edited=self.request('POST','/watch/f_sample1/manage',{'csrf':token,'title':'Updated title','description':'A description <test>','channel_id':channel},cookie=cookie)
|
||||||
|
self.assertEqual(edited[0],303)
|
||||||
|
status,_,watch=self.request('GET','/watch/f_sample1',cookie=cookie)
|
||||||
|
self.assertEqual(status,200)
|
||||||
|
self.assertIn(b'A description <test>',watch)
|
||||||
|
dsi_path=f'/ds/v2-xx/movie/{note[1]}/{note[2]}.ppm'
|
||||||
|
original=(self.directory/'database/Creators'/note[1]/(note[2]+'.ppm')).read_bytes()
|
||||||
|
self.assertEqual(self.request('GET',dsi_path,dsi=True)[2],original)
|
||||||
|
self.assertEqual(self.request('GET','/media/f_sample1.ppm',cookie=cookie)[2],original)
|
||||||
|
self.assertEqual(self.request('POST','/admin/flipnote-visibility',{'csrf':token,'id':note[0],'hidden':'1','reason':'Review'},cookie=cookie)[0],303)
|
||||||
|
self.assertEqual(self.request('GET',dsi_path,dsi=True)[0],404)
|
||||||
|
self.assertEqual(self.request('GET','/thumb/f_sample1.png',cookie=cookie)[0],200)
|
||||||
|
self.request('GET','/logout',cookie=cookie)
|
||||||
|
for path in ('/watch/f_sample1','/media/f_sample1.ppm','/flipnote/f_sample1.ppm','/thumb/f_sample1.png'):
|
||||||
|
self.assertEqual(self.request('GET',path)[0],404,path)
|
||||||
|
token,cookie=self.login()
|
||||||
|
self.assertEqual(self.request('POST','/admin/flipnote-visibility',{'csrf':token,'id':note[0],'hidden':'0'},cookie=cookie)[0],303)
|
||||||
|
self.assertEqual(self.request('GET',dsi_path,dsi=True)[2],original)
|
||||||
|
settings={'csrf':token,'site_name':'Friendly OSE','tagline':'A test server','welcome_message':'Welcome everyone','accent_colour':'#336699','max_upload_kib':'2048'}
|
||||||
|
self.assertEqual(self.request('POST','/admin/server',settings,cookie=cookie)[0],303)
|
||||||
|
self.assertIn(b'Friendly OSE',self.request('GET','/')[2])
|
||||||
|
self.assertEqual(self.request('GET','/register')[0],403)
|
||||||
|
settings['csrf']='bad'
|
||||||
|
settings['site_name']='Not allowed'
|
||||||
|
self.assertEqual(self.request('POST','/admin/server',settings,cookie=cookie)[0],403)
|
||||||
|
self.assertIn(b'Friendly OSE',self.request('GET','/')[2])
|
||||||
|
logo=(self.directory/'web/static/RexiMemo_OSE_Logo.png').read_bytes()
|
||||||
|
boundary='ose-integration-boundary'
|
||||||
|
raw=(f'--{boundary}\r\nContent-Disposition: form-data; name="csrf"\r\n\r\n{token}\r\n--{boundary}\r\nContent-Disposition: form-data; name="logo"; filename="logo.png"\r\nContent-Type: image/png\r\n\r\n'.encode()+logo+f'\r\n--{boundary}--\r\n'.encode())
|
||||||
|
uploaded=self.request('POST','/admin/logo',cookie=cookie,raw=raw,content_type='multipart/form-data; boundary='+boundary)
|
||||||
|
self.assertEqual(uploaded[0],303,uploaded[2][:1000])
|
||||||
|
self.assertTrue(self.request('GET','/branding/logo.png')[2].startswith(b'\x89PNG'))
|
||||||
|
self.assertEqual(self.request('POST','/admin/channel-delete',{'csrf':token,'id':channel,'target_channel_id':'1'},cookie=cookie)[0],303)
|
||||||
|
with closing(sqlite3.connect(self.directory/'database/reximemo.sqlite3')) as db:
|
||||||
|
self.assertEqual(db.execute('SELECT channel_id FROM flipnotes WHERE id=?',(note[0],)).fetchone()[0],1)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__=='__main__':
|
||||||
|
unittest.main()
|
||||||
Reference in new issue
Block a user