test(dsi): cover native features and compatibility
Verify menus, search, creator controls, thumbnails and unchanged original download bytes.
This commit is contained in:
rekushi committed
1 parent
51db1e4329
commit
95168fb1a5
1 file changed
+249
@@ -0,0 +1,249 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import ExitStack
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
import importlib.util
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from PIL import Image
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SAMPLE = next((ROOT / "database" / "Creators").glob("*/*.ppm")).read_bytes()
|
||||
|
||||
|
||||
class Request:
|
||||
def __init__(self, path="/ds/v2-xx/index.uls", args=None, headers=None, ip="192.0.2.40"):
|
||||
self.path = path.encode("ascii")
|
||||
self.args = args or {}
|
||||
self.headers = {b"host": b"flipnote.hatena.com", b"x-dsi-sid": b"test", **(headers or {})}
|
||||
self.response_headers = {}
|
||||
self.response_code = 200
|
||||
self.content = BytesIO()
|
||||
self.ip = ip
|
||||
|
||||
def getAllHeaders(self): return self.headers
|
||||
def getClientIP(self): return self.ip
|
||||
def setHeader(self, key, value): self.response_headers[key] = value
|
||||
def setResponseCode(self, value): self.response_code = value
|
||||
|
||||
|
||||
def module(name):
|
||||
source = ROOT / "hatenadir" / "ds" / "v2-xx" / name
|
||||
spec = importlib.util.spec_from_file_location("test_ose_" + name.replace(".", "_"), source)
|
||||
loaded = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(loaded)
|
||||
return loaded
|
||||
|
||||
|
||||
class DsiFeatureTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.stack = ExitStack()
|
||||
directory = Path(self.stack.enter_context(TemporaryDirectory()))
|
||||
from database import DatabaseBackend
|
||||
self.db = DatabaseBackend(directory / "test.sqlite3")
|
||||
self.stack.callback(self.db.close)
|
||||
import DB
|
||||
import reximemo
|
||||
self.stack.enter_context(patch.object(DB, "Database", self.db))
|
||||
self.stack.enter_context(patch.object(reximemo, "Database", self.db))
|
||||
self.room = self.db.get_or_create_creator("192.0.2.40", fsid="A000000000000001", display_name="Search Artist")
|
||||
with self.db.lock, self.db.conn:
|
||||
for index in range(56):
|
||||
self.db.conn.execute("INSERT INTO flipnotes(public_id,creator_room_id,creator_fsid,filename,title,channel_id,posted_at) VALUES(?,?,?,?,?,1,?)",
|
||||
(f"f_test{index}", self.room["id"], "A000000000000001", f"test{index}", f"Search Animation {index}", index))
|
||||
self.stack.enter_context(patch.object(self.db, "GetFlipnoteTMB", return_value=SAMPLE[:0x6A0]))
|
||||
|
||||
def tearDown(self):
|
||||
self.stack.close()
|
||||
|
||||
def menu(self, data):
|
||||
from Hatenatools import UGO
|
||||
parsed = UGO().Read(data)
|
||||
self.assertIsNot(parsed, False)
|
||||
return parsed.Items
|
||||
|
||||
def test_channels_and_creator_rooms_reach_notes_after_first_fifty(self):
|
||||
channel = module("ch.py").ChannelResource(1, "uls")
|
||||
creator = module("creator.py").CreatorLeaf(self.room["id"], "movies.uls")
|
||||
for resource, path in ((channel, "/ds/v2-xx/ch/1.uls"),
|
||||
(creator, f"/ds/v2-xx/creator/{self.room['id']}/movies.uls")):
|
||||
with self.subTest(path=path):
|
||||
first = self.menu(resource.render_GET(Request(path)))
|
||||
second = self.menu(resource.render_GET(Request(path, {b"page": [b"2"]})))
|
||||
self.assertTrue(any(item[0] == "button" and item[2] == "Next Page" for item in first))
|
||||
self.assertTrue(any(item[0] == "button" and item[2] == "Previous Page" for item in second))
|
||||
self.assertEqual(sum(item[0] == "button" and item[1] == 3 for item in second), 6)
|
||||
|
||||
def test_search_keyboard_keeps_query_and_scope_and_results_are_paginated(self):
|
||||
keyboard = module("search.kbd.py").PyResource()
|
||||
request = Request("/ds/v2-jp/search.kbd", {b"kind": [b"creators"]}, {b"x-email-addr": b"Search Artist"})
|
||||
self.assertEqual(keyboard.render_POST(request), b"OK")
|
||||
target = request.response_headers[b"X-DSi-Forwarder"]
|
||||
self.assertIn(b"http://ugomemo.hatena.ne.jp/ds/v2-jp/search.uls?", target)
|
||||
self.assertIn(b"q=Search+Artist", target)
|
||||
self.assertIn(b"kind=creators", target)
|
||||
nested = module("search.py").PyResource().getChild(b"creators.kbd", request)
|
||||
nested_request = Request("/ds/v2-jp/search/creators.kbd", headers={b"x-email-addr": b"Search Artist"})
|
||||
self.assertEqual(nested.render_POST(nested_request), b"OK")
|
||||
self.assertIn(b"kind=creators", nested_request.response_headers[b"X-DSi-Forwarder"])
|
||||
search = module("search.ugo.py").PyResource()
|
||||
request = Request("/ds/v2-xx/search.uls", {b"q": [b"Search"], b"page": [b"3"]})
|
||||
items = self.menu(search.render_GET(request))
|
||||
self.assertEqual(sum(item[0] == "button" and item[1] == 3 for item in items), 6)
|
||||
self.assertTrue(any(item[0] == "button" and item[2] == "Previous Page" and "q=Search" in item[3] for item in items))
|
||||
creators = self.menu(search.render_GET(Request("/ds/v2-xx/search.uls", {b"q": [b"Artist"], b"kind": [b"creators"]})))
|
||||
self.assertTrue(any(item[0] == "button" and item[2] == "Search Artist" for item in creators))
|
||||
|
||||
def test_comment_pagination_preserves_native_play_and_save_urls(self):
|
||||
flipnote = self.db.get_flipnote_by_public_id("f_test0")
|
||||
for index in range(31):
|
||||
self.db.add_text_comment(flipnote["id"], "Comment " + str(index), ip_address="192.0.2.40")
|
||||
movie = module("movie.py")
|
||||
path = "/ds/v2-xx/movie/A000000000000001/test0.htm"
|
||||
request = Request(path, {b"comments_page": [b"2"]})
|
||||
result = movie.details_page(request, flipnote).decode("utf-8")
|
||||
self.assertIn("Comment 30", result)
|
||||
self.assertNotIn("Comment 29", result)
|
||||
self.assertIn("comments_page=1", result)
|
||||
self.assertIn('<meta name="savebutton" content="http://flipnote.hatena.com/ds/v2-xx/movie/A000000000000001/test0.ppm">', result)
|
||||
self.assertIn('<meta name="upperlink" content="http://flipnote.hatena.com/ds/v2-xx/movie/A000000000000001/test0.ppm">', result)
|
||||
|
||||
def test_hidden_flipnote_and_comment_endpoints_are_not_downloadable(self):
|
||||
flipnote = self.db.get_flipnote_by_public_id("f_test0")
|
||||
movie = module("movie.py").FlipnoteFile("A000000000000001", "test0", "ppm")
|
||||
self.db.set_flipnote_hidden(flipnote["id"], True, "Review")
|
||||
request = Request()
|
||||
movie.render_GET(request)
|
||||
self.assertEqual(request.response_code, 404)
|
||||
|
||||
def test_hidden_mini_comment_and_hidden_parent_do_not_expose_ppm_or_npf(self):
|
||||
flipnote = self.db.get_flipnote_by_public_id("f_test0")
|
||||
with self.db.lock, self.db.conn:
|
||||
cursor = self.db.conn.execute("INSERT INTO comments(public_id,flipnote_id,creator_room_id,type,memo_filename,posted_at) VALUES('c_test',?,?, 'memo','example.ppm',1)",
|
||||
(flipnote["id"], self.room["id"]))
|
||||
comment_id = cursor.lastrowid
|
||||
resource = module("comment.py").PyResource()
|
||||
self.db.set_comment_hidden(comment_id, True, "Review")
|
||||
for suffix in ("ppm", "npf"):
|
||||
request = Request()
|
||||
child = resource.getChild(f"{comment_id}.{suffix}".encode(), request)
|
||||
child.render(request)
|
||||
self.assertEqual(request.response_code, 404)
|
||||
self.db.set_comment_hidden(comment_id, False)
|
||||
self.db.set_flipnote_hidden(flipnote["id"], True)
|
||||
request = Request()
|
||||
resource.getChild(f"{comment_id}.ppm".encode(), request).render(request)
|
||||
self.assertEqual(request.response_code, 404)
|
||||
|
||||
def test_owner_keyboard_edits_preserve_other_details_and_reject_another_ip(self):
|
||||
movie = module("movie.py")
|
||||
fn = self.db.get_flipnote_by_public_id("f_test0")
|
||||
self.db.update_flipnote(fn["id"], fn["title"], "Original description", 1)
|
||||
resource = movie.MetadataKeyboard(fn["creator_fsid"], fn["filename"], "title")
|
||||
denied = Request(headers={b"x-email-addr": b"Not yours"}, ip="192.0.2.99")
|
||||
self.assertNotEqual(resource.render_POST(denied), b"OK")
|
||||
self.assertEqual(self.db.get_flipnote_by_id(fn["id"])["title"], fn["title"])
|
||||
request = Request(headers={b"x-email-addr": b'An "updated" title'})
|
||||
self.assertEqual(resource.render_POST(request), b"OK")
|
||||
updated = self.db.get_flipnote_by_id(fn["id"])
|
||||
self.assertEqual(updated["title"], 'An "updated" title')
|
||||
self.assertEqual(updated["description"], "Original description")
|
||||
self.assertEqual(updated["channel_id"], 1)
|
||||
detail = movie.details_page(Request(), updated).decode()
|
||||
self.assertIn('An "updated" title', detail)
|
||||
self.assertNotIn(""", detail.split("<body>", 1)[1])
|
||||
self.assertIn('name="deletebutton"', detail)
|
||||
self.assertIn("/title.kbd", detail)
|
||||
other_detail = movie.details_page(Request(ip="192.0.2.99"), updated).decode()
|
||||
self.assertNotIn('name="deletebutton"', other_detail)
|
||||
|
||||
def test_original_download_bytes_and_separate_download_counter_are_preserved(self):
|
||||
movie = module("movie.py")
|
||||
with patch.object(self.db, "GetFlipnotePPM", return_value=SAMPLE):
|
||||
self.assertEqual(movie.FlipnoteFile("A000000000000001", "test0", "ppm").render_GET(Request()), SAMPLE)
|
||||
fn = self.db.get_flipnote_by_public_id("f_test0")
|
||||
self.assertEqual((fn["views"], fn["downloads"]), (1, 0))
|
||||
movie.FlipnoteFile("A000000000000001", "test0", "dl").render_GET(Request())
|
||||
self.assertEqual(self.db.get_flipnote_by_id(fn["id"])["downloads"], 1)
|
||||
|
||||
def test_guest_posting_and_upload_limit_settings_apply_to_dsi(self):
|
||||
from server_config import DEFAULTS
|
||||
config = {**DEFAULTS, "guest_posting": False}
|
||||
channel = module("ch.py").ChannelResource(1, "post")
|
||||
request = Request()
|
||||
with patch("server_config.load_config", return_value=config), patch.object(self.db, "AddFlipnote") as add:
|
||||
self.assertIn("Sign in", channel.render_POST(request).decode("utf-16le"))
|
||||
add.assert_not_called()
|
||||
config.update(guest_posting=True, max_upload_bytes=65536)
|
||||
request.content = BytesIO(bytes(65537))
|
||||
with patch("server_config.load_config", return_value=config), patch.object(self.db, "AddFlipnote") as add:
|
||||
self.assertIn("upload limit", channel.render_POST(request).decode("utf-16le"))
|
||||
add.assert_not_called()
|
||||
|
||||
def test_new_ugo_routes_have_uls_aliases_and_channel_directory_paginates(self):
|
||||
from twisted.web.resource import Resource
|
||||
from hatena import LoadHatenadirStructure
|
||||
root = Resource()
|
||||
LoadHatenadirStructure(root, str(ROOT / "hatenadir" / "ds" / "v2-xx"))
|
||||
for slug in (b"channels", b"search", b"allflipnotes", b"index"):
|
||||
self.assertIs(root.children[slug + b".ugo"], root.children[slug + b".uls"])
|
||||
self.assertIn(b"search.kbd", root.children)
|
||||
self.assertIn(b"search", root.children)
|
||||
for index in range(26):
|
||||
self.db.create_channel("Channel " + str(index))
|
||||
items = self.menu(root.children[b"channels.uls"].render_GET(Request(args={b"page": [b"2"]})))
|
||||
self.assertEqual(sum(item[0] == "button" and item[1] == 101 for item in items), 2)
|
||||
|
||||
def test_native_owner_delete_requires_ownership(self):
|
||||
movie = module("movie.py").FlipnoteFile("A000000000000001", "test0", "delete")
|
||||
denied = Request(ip="192.0.2.99")
|
||||
self.assertNotEqual(movie.render_POST(denied), b"OK")
|
||||
self.assertIsNotNone(self.db.get_flipnote_by_public_id("f_test0"))
|
||||
request = Request()
|
||||
self.assertEqual(movie.render_POST(request), b"OK")
|
||||
self.assertIn(b"/creator/", request.response_headers[b"X-DSi-Forwarder"])
|
||||
self.assertIsNone(self.db.get_flipnote_by_public_id("f_test0"))
|
||||
|
||||
|
||||
class ThumbnailTests(unittest.TestCase):
|
||||
def test_saved_thumbnail_frame_is_used_while_first_frame_helper_stays_first(self):
|
||||
from Hatenatools import PPM
|
||||
import reximemo
|
||||
ppm = PPM().Read(SAMPLE, ReadFrames=True, ReadSound=False)
|
||||
first = ppm.GetFrame(0).tobytes()
|
||||
index = next(index for index in range(1, ppm.FrameCount) if ppm.GetFrame(index).tobytes() != first)
|
||||
changed = bytearray(SAMPLE)
|
||||
changed[0x12:0x14] = index.to_bytes(2, "little")
|
||||
raw = bytes(changed)
|
||||
with Image.open(BytesIO(reximemo.ppm_thumbnail_png(raw))) as image:
|
||||
self.assertEqual(image.tobytes(), reximemo.ppm_frame_image(raw, index).convert("RGB").tobytes())
|
||||
self.assertEqual(reximemo.ppm_first_frame_image(raw).tobytes(), reximemo.ppm_frame_image(raw, 0).tobytes())
|
||||
with Image.open(BytesIO(reximemo.ppm_thumbnail_png(raw, first_frame=True))) as image:
|
||||
self.assertEqual(image.tobytes(), reximemo.ppm_first_frame_image(raw).convert("RGB").tobytes())
|
||||
|
||||
def test_both_hidden_layers_leave_only_paper_and_do_not_change_original_bytes(self):
|
||||
import reximemo
|
||||
from Hatenatools import PPM
|
||||
changed = bytearray(SAMPLE)
|
||||
changed[0x6A6] |= 0x30
|
||||
original = bytes(changed)
|
||||
ppm = PPM().Read(original, ReadFrames=True, ReadSound=False)
|
||||
self.assertEqual(ppm.LayerVisibility, (False, False))
|
||||
image = reximemo.ppm_first_frame_image(original)
|
||||
self.assertEqual(len(image.getcolors(2)), 1)
|
||||
self.assertEqual(bytes(changed), original)
|
||||
|
||||
def test_thumbnail_png_is_reused_without_decoding_again(self):
|
||||
import reximemo
|
||||
reximemo._THUMBNAILS.clear()
|
||||
expected = reximemo.ppm_thumbnail_png(SAMPLE, (80, 60))
|
||||
with patch.object(reximemo, "ppm_frame_image", side_effect=AssertionError("Decoded again")):
|
||||
self.assertEqual(reximemo.ppm_thumbnail_png(SAMPLE, (80, 60)), expected)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user