From 95168fb1a59037fa615d33ff865fac5262f52f43 Mon Sep 17 00:00:00 2001 From: rekushi <1+rekushi@noreply.flipnote.world> Date: Thu, 29 Oct 2026 04:43:08 +0000 Subject: [PATCH] test(dsi): cover native features and compatibility Verify menus, search, creator controls, thumbnails and unchanged original download bytes. --- tests/test_dsi_features.py | 249 +++++++++++++++++++++++++++++++++++++ 1 file changed, 249 insertions(+) create mode 100644 tests/test_dsi_features.py diff --git a/tests/test_dsi_features.py b/tests/test_dsi_features.py new file mode 100644 index 0000000..c0e1efb --- /dev/null +++ b/tests/test_dsi_features.py @@ -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('', result) + self.assertIn('', 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("", 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()