From 425da88711b533cea69503e6b3fb85bbf75a7986 Mon Sep 17 00:00:00 2001 From: rekushi <1+rekushi@noreply.flipnote.world> Date: Thu, 29 Oct 2026 04:44:30 +0000 Subject: [PATCH] test(http): exercise OSE through a running server Test web and DSi responses, permissions, moderation, settings and multipart logo uploads over HTTP. --- tests/test_http_integration.py | 159 +++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 tests/test_http_integration.py diff --git a/tests/test_http_integration.py b/tests/test_http_integration.py new file mode 100644 index 0000000..72bd44e --- /dev/null +++ b/tests/test_http_integration.py @@ -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 ','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()