rekushi
e20850a498
Verify migrations, search, pagination, ownership, channel reassignment and hidden-media filtering.
192 lines
11 KiB
Python
192 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from unittest.mock import patch
|
|
import sqlite3
|
|
import unittest
|
|
|
|
import database
|
|
from database import DatabaseBackend
|
|
|
|
|
|
class DatabaseFeatureTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.creators = self.root / "Creators"
|
|
self.comments = self.root / "Comments"
|
|
self.creators.mkdir()
|
|
self.comments.mkdir()
|
|
self.patches = [patch.object(database, "CREATORS_DIR", self.creators),
|
|
patch.object(database, "COMMENTS_DIR", self.comments)]
|
|
for item in self.patches:
|
|
item.start()
|
|
self.db = DatabaseBackend(self.root / "test.sqlite3")
|
|
self.alice = self.db.create_account("Alice", "test password")
|
|
self.bob = self.db.create_account("Bob", "test password")
|
|
self.room = self.db.get_or_create_creator("192.0.2.1", self.alice,
|
|
fsid="A000000000000001")
|
|
self.flipnote = self.add_flipnote("A lovely cat", "A gentle animation")
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
for item in reversed(self.patches):
|
|
item.stop()
|
|
self.temporary.cleanup()
|
|
|
|
def add_flipnote(self, title, description="", *, room=None, channel=1):
|
|
room = room or self.room
|
|
with self.db.conn:
|
|
cursor = self.db.conn.execute(
|
|
"INSERT INTO flipnotes(public_id,creator_room_id,creator_fsid,filename,title,description,channel_id,posted_at) "
|
|
"VALUES(?,?,?,?,?,?,?,?)",
|
|
(self.db._public_id("f_"), room["id"], room["fsid"], self.db._public_id("note_"),
|
|
title, description, channel, 100),
|
|
)
|
|
return self.db.get_flipnote_by_id(cursor.lastrowid)
|
|
|
|
def test_search_pagination_counts_and_literal_punctuation(self):
|
|
self.add_flipnote("Cat number two")
|
|
self.add_flipnote("A dog")
|
|
self.add_flipnote("100% _ literal")
|
|
self.assertEqual(self.db.count_flipnotes(q="CAT"), 2)
|
|
first = self.db.list_flipnotes(q="cat", page=1, per_page=1)
|
|
second = self.db.list_flipnotes(q="cat", page=2, per_page=1)
|
|
self.assertEqual(len(first), 1)
|
|
self.assertNotEqual(first[0]["id"], second[0]["id"])
|
|
self.assertEqual(self.db.count_flipnotes(q="gentle"), 1)
|
|
self.assertEqual(self.db.count_flipnotes(q="Alice"), 4)
|
|
self.assertEqual(self.db.count_flipnotes(q="% _"), 1)
|
|
self.assertEqual(self.db.count_flipnotes(q="' OR 1=1 --"), 0)
|
|
|
|
def test_creator_search_only_lists_creators_with_visible_work(self):
|
|
bob_room = self.db.get_or_create_creator("192.0.2.2", self.bob, fsid="B000000000000002")
|
|
bob_note = self.add_flipnote("Bob's film", room=bob_room)
|
|
self.assertEqual(self.db.count_creators(), 2)
|
|
self.assertEqual(self.db.count_creators("ali"), 1)
|
|
self.assertEqual(self.db.search_creators("A000000000000001")[0]["display_name"], "Alice")
|
|
self.assertNotEqual(self.db.search_creators(page=1, per_page=1)[0]["id"],
|
|
self.db.search_creators(page=2, per_page=1)[0]["id"])
|
|
self.db.set_flipnote_hidden(bob_note["id"], True)
|
|
self.assertEqual(self.db.count_creators(), 1)
|
|
|
|
def test_channel_edit_and_deletion_moves_all_content(self):
|
|
channel = self.db.create_channel("Animation", "Drawn stories")
|
|
self.assertTrue(self.db.update_channel(channel, "Animated films", "Stories"))
|
|
self.assertEqual(self.db.get_channel(channel)["title"], "Animated films")
|
|
self.db.update_flipnote(self.flipnote["id"], "Cat", "Description", channel)
|
|
self.db.set_flipnote_hidden(self.flipnote["id"], True)
|
|
self.assertTrue(self.db.delete_channel(channel))
|
|
self.assertIsNone(self.db.get_channel(channel))
|
|
self.assertEqual(self.db.get_flipnote_by_id(self.flipnote["id"], include_hidden=True)["channel_id"], 1)
|
|
with self.assertRaises(ValueError):
|
|
self.db.delete_channel(1)
|
|
with self.assertRaises(ValueError):
|
|
self.db.create_channel("", "")
|
|
with self.assertRaises(ValueError):
|
|
self.db.update_flipnote(self.flipnote["id"], "x" * 81)
|
|
with self.assertRaises(ValueError):
|
|
self.db.update_flipnote(self.flipnote["id"], "Cat", "x" * 501)
|
|
|
|
def test_ownership_does_not_allow_an_ip_to_claim_an_account(self):
|
|
self.assertTrue(self.db.owns_flipnote(self.alice, "elsewhere", self.flipnote))
|
|
self.assertFalse(self.db.owns_flipnote(self.bob, "192.0.2.1", self.flipnote))
|
|
self.assertFalse(self.db.owns_flipnote(None, "192.0.2.1", self.flipnote))
|
|
guest_room = self.db.get_or_create_creator("192.0.2.8", fsid="C000000000000003")
|
|
guest = self.add_flipnote("Guest film", room=guest_room)
|
|
self.assertTrue(self.db.owns_flipnote(None, "192.0.2.8", guest))
|
|
self.assertFalse(self.db.owns_flipnote(None, "192.0.2.9", guest))
|
|
self.assertFalse(self.db.owns_flipnote(self.alice, "192.0.2.8", guest))
|
|
cid = self.db.add_text_comment(self.flipnote["id"], "Hello", account_id=self.bob)
|
|
comment = self.db.get_comment(cid)
|
|
self.assertTrue(self.db.owns_comment(self.bob, "", comment))
|
|
self.assertFalse(self.db.owns_comment(self.alice, "", comment))
|
|
|
|
def test_hidden_content_is_absent_in_every_default_read(self):
|
|
cid = self.db.add_text_comment(self.flipnote["id"], "A comment", account_id=self.bob)
|
|
fsid, filename = self.flipnote["creator_fsid"], self.flipnote["filename"]
|
|
path = self.db.FlipnotePath(fsid, filename)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"original PPM")
|
|
self.assertEqual(self.db.GetFlipnotePPM(fsid, filename), b"original PPM")
|
|
self.db.set_flipnote_hidden(self.flipnote["id"], True, "Under review")
|
|
self.assertIsNone(self.db.get_flipnote_by_id(self.flipnote["id"]))
|
|
self.assertIsNone(self.db.get_flipnote_by_public_id(self.flipnote["public_id"]))
|
|
self.assertFalse(self.db.FlipnoteExists(fsid, filename))
|
|
self.assertEqual(self.db.list_flipnotes(), [])
|
|
self.assertEqual(self.db.count_flipnotes(), 0)
|
|
self.assertEqual(self.db.list_creator_flipnotes(self.room["id"]), [])
|
|
self.assertEqual(self.db.count_comments(self.flipnote["id"]), 0)
|
|
self.assertEqual(self.db.list_comments(self.flipnote["id"]), [])
|
|
self.assertIsNone(self.db.get_comment(cid))
|
|
self.assertEqual(self.db.get_comment(cid, include_hidden=True)["id"], cid)
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.db.GetFlipnotePPM(fsid, filename)
|
|
with self.assertRaises(FileNotFoundError):
|
|
self.db.GetFlipnoteTMB(fsid, filename)
|
|
with self.assertRaises(ValueError):
|
|
self.db.add_text_comment(self.flipnote["id"], "Cannot add")
|
|
self.assertEqual(self.db.GetFlipnotePPM(fsid, filename, include_hidden=True), b"original PPM")
|
|
self.assertEqual(self.db.list_flipnotes(include_hidden=True)[0]["moderation_reason"], "Under review")
|
|
self.db.set_flipnote_hidden(self.flipnote["id"], False)
|
|
self.assertEqual(self.db.get_flipnote(fsid, filename)["moderation_reason"], "")
|
|
self.assertIsNotNone(self.db.get_comment(cid))
|
|
|
|
def test_hidden_mini_comment_cannot_be_downloaded(self):
|
|
with self.db.conn:
|
|
cur = self.db.conn.execute(
|
|
"INSERT INTO comments(public_id,flipnote_id,creator_room_id,type,memo_filename,posted_at) VALUES(?,?,?,?,?,?)",
|
|
("m_example", self.flipnote["id"], self.room["id"], "memo", "sample.ppm", 100),
|
|
)
|
|
cid = cur.lastrowid
|
|
(self.comments / "sample.ppm").write_bytes(b"mini PPM")
|
|
self.assertEqual(self.db.comment_ppm(cid), b"mini PPM")
|
|
self.db.set_comment_hidden(cid, True, "Review comment")
|
|
self.assertIsNone(self.db.comment_ppm(cid))
|
|
self.assertEqual(self.db.comment_ppm(cid, include_hidden=True), b"mini PPM")
|
|
self.assertEqual(self.db.count_comments(self.flipnote["id"]), 0)
|
|
self.assertEqual(self.db.count_comments(self.flipnote["id"], include_hidden=True), 1)
|
|
self.assertEqual(self.db.list_moderation_comments(q="Alice")[0]["moderation_reason"], "Review comment")
|
|
self.assertEqual(self.db.count_moderation_comments(q="Alice"), 1)
|
|
self.db.set_comment_hidden(cid, False)
|
|
self.assertEqual(self.db.comment_ppm(cid), b"mini PPM")
|
|
self.db.delete_comment(cid)
|
|
self.assertIsNone(self.db.comment_ppm(cid, include_hidden=True))
|
|
|
|
def test_deleted_parent_also_hides_comments_from_staff_lists(self):
|
|
cid = self.db.add_text_comment(self.flipnote["id"], "Hello")
|
|
self.db.delete_flipnote(self.flipnote["id"])
|
|
self.assertIsNone(self.db.get_comment(cid, include_hidden=True))
|
|
self.assertEqual(self.db.count_moderation_comments(), 0)
|
|
self.assertEqual(self.db.list_moderation_comments(), [])
|
|
|
|
def test_schema_upgrade_preserves_legacy_records(self):
|
|
path = self.root / "legacy.sqlite3"
|
|
with sqlite3.connect(path) as connection:
|
|
connection.executescript("""
|
|
CREATE TABLE flipnotes(id INTEGER PRIMARY KEY,public_id TEXT UNIQUE,creator_room_id INTEGER,
|
|
creator_fsid TEXT,filename TEXT,title TEXT,channel_id INTEGER,views INTEGER DEFAULT 0,
|
|
downloads INTEGER DEFAULT 0,stars INTEGER DEFAULT 0,posted_at INTEGER,deleted INTEGER DEFAULT 0,
|
|
UNIQUE(creator_fsid,filename));
|
|
INSERT INTO flipnotes VALUES(1,'f_old',1,'A000000000000001','old','Old title',1,7,3,9,50,0);
|
|
CREATE TABLE comments(id INTEGER PRIMARY KEY,public_id TEXT UNIQUE,flipnote_id INTEGER,
|
|
creator_room_id INTEGER,type TEXT,text_content TEXT,memo_filename TEXT,posted_at INTEGER,
|
|
deleted INTEGER DEFAULT 0);
|
|
INSERT INTO comments VALUES(1,'m_old',1,1,'text','Old comment',NULL,51,0);
|
|
""")
|
|
upgraded = DatabaseBackend(path)
|
|
try:
|
|
note = upgraded._one("SELECT * FROM flipnotes WHERE id=1")
|
|
comment = upgraded._one("SELECT * FROM comments WHERE id=1")
|
|
self.assertEqual((note["title"], note["views"], note["stars"]), ("Old title", 7, 9))
|
|
self.assertEqual(comment["text_content"], "Old comment")
|
|
for item in (note, comment):
|
|
self.assertEqual((item["description"], item["hidden"], item["moderation_reason"]), ("", 0, ""))
|
|
finally:
|
|
upgraded.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|