docker-ep-web-server/test_account_lifecycle.py

import tempfile
import unittest
from pathlib import Path

import app as application
from werkzeug.security import generate_password_hash


class AccountLifecycleTestCase(unittest.TestCase):
    def setUp(self):
        self.temp_directory = tempfile.TemporaryDirectory()
        self.database_path = Path(self.temp_directory.name) / "test.db"
        self.upload_directory = Path(self.temp_directory.name) / "uploads"
        self.original_config = application.app.config.copy()
        application.app.config.update(
            DATABASE=self.database_path,
            UPLOAD_FOLDER=self.upload_directory,
            TESTING=True,
            WTF_CSRF_ENABLED=False,
        )
        with application.app.app_context():
            application.init_db()
            db = application.get_db()
            self.admin_id = db.execute(
                """INSERT INTO users (username, email, email_verified, password_hash, created_at, is_approved, role, mfa_enabled)
                   VALUES (?, ?, 1, ?, ?, 1, 'admin', 1)""",
                ("administrator", "admin@example.com", generate_password_hash("admin-password"), "2026-09-01T00:00:00+00:00"),
            ).lastrowid
            self.member_id = db.execute(
                """INSERT INTO users (username, email, email_verified, password_hash, created_at, is_approved, role, mfa_enabled)
                   VALUES (?, ?, 1, ?, ?, 1, 'member', 1)""",
                ("member", "member@example.com", generate_password_hash("member-password"), "2026-09-01T00:00:00+00:00"),
            ).lastrowid
            db.commit()
        self.client = application.app.test_client()

    def tearDown(self):
        application.app.config.update(self.original_config)
        self.temp_directory.cleanup()

    def sign_in_as_admin(self):
        with self.client.session_transaction() as session:
            session["user_id"] = self.admin_id
            session["mfa_verified"] = True

    def test_freeze_blocks_login_and_active_session(self):
        self.sign_in_as_admin()
        response = self.client.post(f"/admin/users/{self.member_id}/freeze")
        self.assertEqual(response.status_code, 302)
        with application.app.app_context():
            self.assertEqual(application.get_db().execute("SELECT is_frozen FROM users WHERE id = ?", (self.member_id,)).fetchone()[0], 1)

        response = self.client.post("/login", data={"username": "member", "password": "member-password"})
        self.assertIn(b"This account has been frozen by an administrator.", response.data)
        with self.client.session_transaction() as session:
            session["user_id"] = self.member_id
            session["mfa_verified"] = True
        response = self.client.get("/write")
        self.assertEqual(response.status_code, 302)
        self.assertIn("/login", response.location)
        with self.client.session_transaction() as session:
            self.assertNotIn("user_id", session)

    def test_delete_member_removes_related_data_and_images(self):
        self.upload_directory.mkdir()
        primary_image = self.upload_directory / "primary.webp"
        inline_image = self.upload_directory / "inline.webp"
        primary_image.write_bytes(b"primary")
        inline_image.write_bytes(b"inline")
        with application.app.app_context():
            db = application.get_db()
            post_id = db.execute(
                """INSERT INTO posts (title, slug, excerpt, body, category, image_filename, author_id, created_at, updated_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                ("Member post", "member-post", "Excerpt", "Body", "Code", primary_image.name, self.member_id, "2026-09-01T00:00:00+00:00", "2026-09-01T00:00:00+00:00"),
            ).lastrowid
            db.execute("INSERT INTO post_images (post_id, filename, caption, position, created_at) VALUES (?, ?, ?, ?, ?)", (post_id, inline_image.name, "", 1, "2026-09-01T00:00:00+00:00"))
            db.execute("INSERT INTO password_reset_tokens (user_id, token_hash, expires_at, created_at) VALUES (?, ?, ?, ?)", (self.member_id, "reset-token", "2026-10-01T00:00:00+00:00", "2026-09-01T00:00:00+00:00"))
            db.execute("INSERT INTO email_verification_tokens (user_id, email_address, token_hash, token_type, expires_at, created_at) VALUES (?, ?, ?, ?, ?, ?)", (self.member_id, "member@example.com", "verification-token", "registration", "2026-10-01T00:00:00+00:00", "2026-09-01T00:00:00+00:00"))
            db.commit()

        self.sign_in_as_admin()
        self.assertEqual(self.client.post(f"/admin/users/{self.member_id}/delete").status_code, 302)
        with application.app.app_context():
            db = application.get_db()
            self.assertIsNone(db.execute("SELECT id FROM users WHERE id = ?", (self.member_id,)).fetchone())
            self.assertIsNone(db.execute("SELECT id FROM posts WHERE id = ?", (post_id,)).fetchone())
            self.assertEqual(db.execute("SELECT COUNT(*) FROM post_images WHERE post_id = ?", (post_id,)).fetchone()[0], 0)
            self.assertEqual(db.execute("SELECT COUNT(*) FROM password_reset_tokens WHERE user_id = ?", (self.member_id,)).fetchone()[0], 0)
            self.assertEqual(db.execute("SELECT COUNT(*) FROM email_verification_tokens WHERE user_id = ?", (self.member_id,)).fetchone()[0], 0)
        self.assertFalse(primary_image.exists())
        self.assertFalse(inline_image.exists())

    def test_administrator_accounts_cannot_be_frozen_or_deleted(self):
        self.sign_in_as_admin()
        self.assertEqual(self.client.post(f"/admin/users/{self.admin_id}/freeze").status_code, 302)
        self.assertEqual(self.client.post(f"/admin/users/{self.admin_id}/delete").status_code, 404)
        with application.app.app_context():
            self.assertEqual(application.get_db().execute("SELECT is_frozen FROM users WHERE id = ?", (self.admin_id,)).fetchone()[0], 0)


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