feat: port Hatena server to Python 3
* Update server-side database and dynamic resource code for Python 3 * Preserve existing Flipnote, UGO, NTFT, CSS, localization, and Hatena assets * Port Flipnote upload and movie endpoints * Update front page UGO menu generators * Preserve existing database and sample Flipnote data * Add regression tests for database operations and binary formats * Add server resource loading tests * Maintain compatibility with the original DSi Flipnote Studio protocol
This commit is contained in:
Rekushi authored and GitHub committed
1 parent
687a2bcb43
commit
e4bb0b00ba
48 files changed
+3463
No files matched your search
@@ -0,0 +1,59 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import database as database_module
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SAMPLE_CREATOR = "913304C0CC74CC1F"
|
||||
SAMPLE_NAME = "74CC1F_0D3AA40F2B3A0_000"
|
||||
|
||||
|
||||
class DatabaseTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.old_cwd = os.getcwd()
|
||||
temp_root = Path(self.tmp.name)
|
||||
shutil.copytree(ROOT / "database", temp_root / "database")
|
||||
os.chdir(temp_root)
|
||||
self.db = database_module.Database.__class__()
|
||||
|
||||
def tearDown(self):
|
||||
# Prevent the per-test instance from having anything left to flush at interpreter exit.
|
||||
self.db.Creator.clear()
|
||||
self.db.new = False
|
||||
os.chdir(self.old_cwd)
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_read_sample(self):
|
||||
self.assertTrue(self.db.CreatorExists(SAMPLE_CREATOR))
|
||||
row = self.db.GetFlipnote(SAMPLE_CREATOR, SAMPLE_NAME)
|
||||
self.assertTrue(row)
|
||||
self.assertEqual(row[0], SAMPLE_NAME)
|
||||
self.assertEqual(len(row), 9)
|
||||
self.assertTrue(self.db.GetFlipnotePPM(SAMPLE_CREATOR, SAMPLE_NAME).startswith(b"PARA"))
|
||||
self.assertEqual(len(self.db.GetFlipnoteTMB(SAMPLE_CREATOR, SAMPLE_NAME)), 0x6A0)
|
||||
|
||||
def test_counters_and_write(self):
|
||||
self.assertTrue(self.db.AddView(SAMPLE_CREATOR, SAMPLE_NAME))
|
||||
self.assertTrue(self.db.AddStar(SAMPLE_CREATOR, SAMPLE_NAME, 5, "green"))
|
||||
self.assertTrue(self.db.AddDownload(SAMPLE_CREATOR, SAMPLE_NAME))
|
||||
self.assertEqual(self.db.Views, 1)
|
||||
self.assertEqual(self.db.Stars, 1)
|
||||
self.assertEqual(self.db.Downloads, 1)
|
||||
row = self.db.GetFlipnote(SAMPLE_CREATOR, SAMPLE_NAME, True)
|
||||
self.assertEqual(int(row[1]), 3)
|
||||
self.assertEqual(int(row[3]), 5)
|
||||
self.assertEqual(int(row[8]), 1)
|
||||
self.db.write()
|
||||
text = Path("database/Creators") / SAMPLE_CREATOR / "flipnotes.dat"
|
||||
self.assertIn("\t3\t", text.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,98 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from Hatenatools import NTFT, PPM, TMB, UGO
|
||||
from Hatenatools.PPM import WriteImage, get_metadata
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SAMPLE_PPM = ROOT / "database" / "Creators" / "913304C0CC74CC1F" / "74CC1F_0D3AA40F2B3A0_000.ppm"
|
||||
NEWS_NTFT = ROOT / "hatenadir" / "ds" / "v2-xx" / "__ntfts" / "News.ntft"
|
||||
KAERU_NTFT = ROOT / "hatenadir" / "images" / "ds" / "kaeru.ntft"
|
||||
INDEX_UGOXML = ROOT / "hatenadir" / "ds" / "v2-xx" / "index.ugoxml"
|
||||
INBOX_UGOXML = ROOT / "hatenadir" / "ds" / "v2-xx" / "inbox.ugoxml"
|
||||
|
||||
|
||||
class PPMTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.data = SAMPLE_PPM.read_bytes()
|
||||
|
||||
def test_tmb_metadata_and_exact_pack(self):
|
||||
tmb = TMB().Read(self.data[:0x6A0], DecodeThumbnail=True)
|
||||
self.assertTrue(tmb)
|
||||
self.assertEqual(tmb.EditorAuthorID, "913304C0CC74CC1F")
|
||||
self.assertEqual(tmb.CurrentFilename, "74CC1F_0D3AA40F2B3A0_000.tmb")
|
||||
self.assertEqual(tmb.FrameCount, 3)
|
||||
self.assertEqual(tmb.GetThumbnail().shape, (64, 48))
|
||||
self.assertEqual(tmb.Pack(), self.data[:0x6A0])
|
||||
|
||||
def test_ppm_full_parse_and_frame(self):
|
||||
ppm = PPM().Read(self.data, DecodeThumbnail=True, ReadFrames=True, ReadSound=True)
|
||||
self.assertTrue(ppm)
|
||||
self.assertEqual(ppm.EditorAuthorID, "913304C0CC74CC1F")
|
||||
self.assertEqual(ppm.CurrentFilename, "74CC1F_0D3AA40F2B3A0_000.ppm")
|
||||
self.assertEqual(ppm.FrameCount, 3)
|
||||
self.assertEqual(ppm.GetThumbnail().shape, (64, 48))
|
||||
frame = ppm.GetFrame(0)
|
||||
self.assertIsInstance(frame, np.ndarray)
|
||||
self.assertEqual(frame.shape, (256, 192))
|
||||
self.assertEqual(len(ppm.SoundData), 4)
|
||||
|
||||
def test_metadata_helper_infers_filetype(self):
|
||||
ppm = PPM().Read(self.data, ReadFrames=False)
|
||||
metadata = get_metadata(ppm)
|
||||
self.assertEqual(metadata["Current filename"], "74CC1F_0D3AA40F2B3A0_000.ppm")
|
||||
self.assertEqual(metadata["Number of frames"], 3)
|
||||
self.assertIn("Frame speed", metadata)
|
||||
|
||||
def test_image_export(self):
|
||||
ppm = PPM().Read(self.data, DecodeThumbnail=True, ReadFrames=False)
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
output = Path(tmp) / "thumb.png"
|
||||
self.assertTrue(WriteImage(ppm.GetThumbnail(), str(output)))
|
||||
self.assertGreater(output.stat().st_size, 0)
|
||||
|
||||
|
||||
class NTFTTests(unittest.TestCase):
|
||||
def test_exact_roundtrip_news(self):
|
||||
raw = NEWS_NTFT.read_bytes()
|
||||
image = NTFT().Read(raw, (32, 32))
|
||||
self.assertTrue(image)
|
||||
self.assertEqual(image.Pack(), raw)
|
||||
|
||||
def test_exact_roundtrip_kaeru(self):
|
||||
raw = KAERU_NTFT.read_bytes()
|
||||
image = NTFT().Read(raw, (36, 30))
|
||||
self.assertTrue(image)
|
||||
self.assertEqual(image.Pack(), raw)
|
||||
|
||||
|
||||
class UGOTests(unittest.TestCase):
|
||||
def _roundtrip_xml(self, path: Path):
|
||||
ugo = UGO().ReadXML(str(path), False)
|
||||
self.assertTrue(ugo)
|
||||
packed = ugo.Pack()
|
||||
self.assertIsInstance(packed, bytes)
|
||||
self.assertEqual(packed[:4], b"UGAR")
|
||||
decoded = UGO().Read(packed)
|
||||
self.assertTrue(decoded)
|
||||
# Embedded-file names are XML-side metadata and are not stored in the
|
||||
# binary UGO format. The meaningful binary round-trip is idempotence.
|
||||
self.assertEqual(decoded.Pack(), packed)
|
||||
|
||||
def test_index_xml(self):
|
||||
self._roundtrip_xml(INDEX_UGOXML)
|
||||
|
||||
def test_inbox_xml(self):
|
||||
self._roundtrip_xml(INBOX_UGOXML)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,124 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def install_twisted_stub():
|
||||
"""Install only the Twisted surface needed to smoke-test this project."""
|
||||
if importlib.util.find_spec("twisted") is not None:
|
||||
return
|
||||
|
||||
twisted = types.ModuleType("twisted")
|
||||
web = types.ModuleType("twisted.web")
|
||||
resource_mod = types.ModuleType("twisted.web.resource")
|
||||
static_mod = types.ModuleType("twisted.web.static")
|
||||
internet_mod = types.ModuleType("twisted.internet")
|
||||
|
||||
class Resource:
|
||||
isLeaf = False
|
||||
|
||||
def __init__(self):
|
||||
self.children = {}
|
||||
|
||||
def putChild(self, name, child):
|
||||
if not isinstance(name, bytes):
|
||||
raise TypeError("Twisted child names must be bytes")
|
||||
self.children[name] = child
|
||||
|
||||
def getChild(self, name, request):
|
||||
return self
|
||||
|
||||
def getChildWithDefault(self, name, request):
|
||||
return self.children.get(name, self.getChild(name, request))
|
||||
|
||||
class File(Resource):
|
||||
def __init__(self, path):
|
||||
super().__init__()
|
||||
self.path = path
|
||||
|
||||
def render(self, request):
|
||||
p = Path(self.path)
|
||||
return p.read_bytes() if p.is_file() else b""
|
||||
|
||||
class Reactor:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def callLater(self, delay, func, *args, **kwargs):
|
||||
self.calls.append((delay, func, args, kwargs))
|
||||
return object()
|
||||
|
||||
def callInThread(self, func, *args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
|
||||
resource_mod.Resource = Resource
|
||||
static_mod.File = File
|
||||
web.resource = resource_mod
|
||||
web.static = static_mod
|
||||
internet_mod.reactor = Reactor()
|
||||
twisted.web = web
|
||||
twisted.internet = internet_mod
|
||||
|
||||
sys.modules.update(
|
||||
{
|
||||
"twisted": twisted,
|
||||
"twisted.web": web,
|
||||
"twisted.web.resource": resource_mod,
|
||||
"twisted.web.static": static_mod,
|
||||
"twisted.internet": internet_mod,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ResourceTreeTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
install_twisted_stub()
|
||||
cls.old_cwd = Path.cwd()
|
||||
import os
|
||||
os.chdir(ROOT)
|
||||
import hatena
|
||||
cls.hatena = hatena
|
||||
cls.root = hatena.Setup()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
import os
|
||||
os.chdir(cls.old_cwd)
|
||||
|
||||
def test_core_resource_tree_loads(self):
|
||||
region = self.root.dsResource.region
|
||||
expected = {
|
||||
b"index.ugo",
|
||||
b"inbox.ugo",
|
||||
b"movie",
|
||||
b"post",
|
||||
b"frontpage",
|
||||
b"help",
|
||||
b"eula_list.tsv",
|
||||
}
|
||||
self.assertTrue(expected.issubset(region.children))
|
||||
self.assertIn(b"flipnote.post", region.children[b"post"].children)
|
||||
self.assertIn(b"hotmovies.ugo", region.children[b"frontpage"].children)
|
||||
self.assertIn(b"likedmovies.ugo", region.children[b"frontpage"].children)
|
||||
self.assertIn(b"newmovies.ugo", region.children[b"frontpage"].children)
|
||||
|
||||
def test_prepacked_ugo_resources_are_binary(self):
|
||||
region = self.root.dsResource.region
|
||||
self.assertTrue(region.children[b"index.ugo"].ugofile.startswith(b"UGAR"))
|
||||
self.assertTrue(region.children[b"inbox.ugo"].ugofile.startswith(b"UGAR"))
|
||||
|
||||
def test_hot_page_builds_from_database(self):
|
||||
hot = self.root.dsResource.region.children[b"frontpage"].children[b"hotmovies.ugo"]
|
||||
self.assertGreaterEqual(len(hot.pages), 1)
|
||||
self.assertTrue(hot.pages[0].startswith(b"UGAR"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user