RexiMemo-OSE/tests/test_web_features.py

"""Creator and administrator tools enforce the same visibility on every route."""
from __future__ import annotations

from contextlib import ExitStack
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
import io
import unittest

import database
from database import DatabaseBackend
import server_config
import web_features
import webapp


TOKEN = "a" * 64


class Request:
    def __init__(self, path, *, ip="192.0.2.9", fields=None, query=None, csrf=TOKEN):
        self.path = path.encode("utf-8")
        self._ip = ip
        self.args = {key.encode(): [str(value).encode()] for key, value in (query or {}).items()}
        if fields is not None:
            self.args.update({key.encode(): [str(value).encode()] for key, value in fields.items()})
            self.args[b"csrf"] = [csrf.encode()]
        self.content = io.BytesIO()
        self.headers = {b"host": b"localhost:8080", b"cookie": ("ose_csrf=" + TOKEN).encode()}
        self.response_headers = {}
        self.response_code = 200

    def getClientIP(self):
        return self._ip

    def getAllHeaders(self):
        return self.headers

    def setHeader(self, key, value):
        self.response_headers[key.lower()] = value

    def setResponseCode(self, code):
        self.response_code = code

    def isSecure(self):
        return False


class WebFeatureTests(unittest.TestCase):
    def setUp(self):
        self.stack = ExitStack()
        self.addCleanup(self.stack.close)
        self.root = Path(self.stack.enter_context(TemporaryDirectory()))
        self.stack.enter_context(patch.dict("os.environ", {"REXIMEMO_CONFIG": str(self.root / "settings.json")}))
        self.creators = self.root / "Creators"
        self.comments = self.root / "Comments"
        self.creators.mkdir()
        self.comments.mkdir()
        self.stack.enter_context(patch.object(database, "CREATORS_DIR", self.creators))
        self.stack.enter_context(patch.object(database, "COMMENTS_DIR", self.comments))
        self.db = DatabaseBackend(self.root / "test.sqlite3")
        self.stack.callback(self.db.close)
        self.stack.enter_context(patch.object(webapp, "Database", self.db))
        self.stack.enter_context(patch.object(web_features, "Database", self.db))
        self.router = webapp.WebRoot()
        self.alice = self.db.create_account("Alice", "test password")
        self.bob = self.db.create_account("Bob", "test password")
        self.admin = self.db.get_account_by_username("reximemo")["id"]
        for account, address in ((self.alice, "192.0.2.1"), (self.bob, "192.0.2.2"), (self.admin, "192.0.2.3")):
            self.db.login_ip(address, account)
        room = self.db.get_or_create_creator("192.0.2.1", self.alice, fsid="A000000000000001")
        with self.db.conn:
            cur = self.db.conn.execute(
                "INSERT INTO flipnotes(public_id,creator_room_id,creator_fsid,filename,title,channel_id,posted_at) VALUES(?,?,?,?,?,?,?)",
                ("f_test", room["id"], room["fsid"], "sample", "Original title", 1, 100),
            )
        self.note_id = cur.lastrowid
        path = self.db.FlipnotePath(room["fsid"], "sample")
        path.parent.mkdir(parents=True)
        path.write_bytes(b"Original PPM bytes")

    def get(self, path, *, ip="192.0.2.9", query=None):
        request = Request(path, ip=ip, query=query)
        body = self.router.render_GET(request)
        return request, body

    def post(self, path, fields=None, *, ip="192.0.2.3", csrf=TOKEN):
        request = Request(path, ip=ip, fields=fields or {}, csrf=csrf)
        body = self.router.render_POST(request)
        return request, body

    def test_creator_can_edit_but_another_account_cannot(self):
        fields = {"title": "Edited title", "description": "A description", "channel_id": 1}
        denied, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.2")
        self.assertEqual(denied.response_code, 403)
        self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Original title")
        allowed, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.1")
        self.assertEqual(allowed.response_code, 303)
        self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Edited title")
        self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["description"], "A description")
        self.assertEqual(self.get("/watch/f_test/manage", ip="192.0.2.2")[0].response_code, 403)
        self.assertEqual(self.get("/watch/f_test/manage", ip="192.0.2.1")[0].response_code, 200)

    def test_only_commenter_or_administrator_can_remove_comment(self):
        comment = self.db.add_text_comment(self.note_id, "Bob's comment", account_id=self.bob)
        denied, _ = self.post(f"/comments/{comment}/delete", ip="192.0.2.1")
        self.assertEqual(denied.response_code, 403)
        self.assertIsNotNone(self.db.get_comment(comment))
        allowed, _ = self.post(f"/comments/{comment}/delete", ip="192.0.2.2")
        self.assertEqual(allowed.response_code, 303)
        self.assertIsNone(self.db.get_comment(comment))

    def test_moderation_hides_public_pages_and_media_but_staff_can_review(self):
        note = self.db.get_flipnote_by_id(self.note_id)
        request, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 1, "reason": "Review"})
        self.assertEqual(request.response_code, 303)
        with patch.object(webapp, "ppm_thumbnail_png", return_value=b"PNG preview"):
            for path in ("/watch/f_test", "/media/f_test.ppm", "/flipnote/f_test.ppm", "/thumb/f_test.png"):
                with self.subTest(path=path):
                    public, _ = self.get(path)
                    owner, _ = self.get(path, ip="192.0.2.1")
                    reviewer, body = self.get(path, ip="192.0.2.3")
                    self.assertEqual(public.response_code, 404)
                    self.assertEqual(owner.response_code, 404)
                    self.assertEqual(reviewer.response_code, 200)
                    self.assertIn(b"no-store", reviewer.response_headers[b"cache-control"])
                    self.assertTrue(body)
        public, body = self.get("/browse")
        self.assertNotIn(note["public_id"].encode(), body)
        admin, body = self.get("/browse", ip="192.0.2.3")
        self.assertIn(b"hidden-content", body)
        self.assertIn(b"f_test", body)
        request, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 0})
        self.assertEqual(request.response_code, 303)
        self.assertEqual(self.get("/media/f_test.ppm")[1], b"Original PPM bytes")

    def test_administrator_tabs_render_and_regular_accounts_are_denied(self):
        self.db.add_text_comment(self.note_id, "Example comment", account_id=self.bob)
        for tab, heading in (("flipnotes", "Flipnotes"), ("comments", "Example comment"),
                             ("channels", "New channel"), ("server", "Server appearance"), ("bans", "Block an IP")):
            with self.subTest(tab=tab):
                admin, body = self.get("/admin", ip="192.0.2.3", query={"tab": tab})
                self.assertEqual(admin.response_code, 200)
                self.assertIn(heading.encode(), body)
                self.assertIn(b'name="csrf"', body)
                regular, _ = self.get("/admin", ip="192.0.2.1", query={"tab": tab})
                self.assertEqual(regular.response_code, 403)

    def test_channel_removal_moves_notes_and_preserves_default_channel(self):
        channel = self.db.create_channel("Animations", "Drawn stories")
        self.db.update_flipnote(self.note_id, "Film", channel_id=channel)
        response, _ = self.post("/admin/channel-delete", {"id": channel, "target_channel_id": 1})
        self.assertEqual(response.response_code, 303)
        self.assertIsNone(self.db.get_channel(channel))
        self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["channel_id"], 1)
        protected, _ = self.post("/admin/channel-delete", {"id": 1, "target_channel_id": 1})
        self.assertEqual(protected.response_code, 400)
        self.assertIsNotNone(self.db.get_channel(1))

    def test_server_settings_save_and_guest_posting_setting_is_enforced(self):
        fields = {"site_name": "My Flipnote Server", "tagline": "Our drawings", "welcome_message": "Welcome!",
                  "accent_colour": "#123456", "max_upload_kib": 2048, "registration_enabled": 1}
        request, _ = self.post("/admin/server", fields)
        self.assertEqual(request.response_code, 303)
        config = server_config.load_config()
        self.assertEqual(config["site_name"], "My Flipnote Server")
        self.assertFalse(config["guest_posting"])
        public, body = self.get("/")
        self.assertIn(b"My Flipnote Server", body)
        rejected, _ = self.post("/watch/f_test/comment", {"text": "Guest comment"}, ip="192.0.2.9")
        self.assertEqual(rejected.response_code, 403)
        allowed, _ = self.post("/watch/f_test/comment", {"text": "Signed in comment"}, ip="192.0.2.1")
        self.assertEqual(allowed.response_code, 303)
        self.assertEqual(self.db.count_comments(self.note_id), 1)

    def test_invalid_csrf_rejects_creator_and_staff_mutations(self):
        fields = {"title": "Bad change", "channel_id": 1}
        rejected, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.1", csrf="b" * 64)
        self.assertEqual(rejected.response_code, 403)
        rejected, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 1}, csrf="")
        self.assertEqual(rejected.response_code, 403)
        self.assertFalse(self.db.get_flipnote_by_id(self.note_id)["hidden"])
        self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Original title")


if __name__ == "__main__":
    unittest.main()