diff --git a/tests/test_database_features.py b/tests/test_database_features.py new file mode 100644 index 0000000..33d595f --- /dev/null +++ b/tests/test_database_features.py @@ -0,0 +1,191 @@ +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()