guanghu-ice-heart/server-tools/hololake-code-channel/jd-candidate/test_owner_migration.py

#!/usr/bin/env python3

from __future__ import annotations

import importlib.util
import json
import pathlib
import sqlite3
import subprocess
import tempfile
import unittest


MODULE_PATH = pathlib.Path(__file__).with_name("hlcc-bootstrap.py")
SPEC = importlib.util.spec_from_file_location("hlcc_bootstrap", MODULE_PATH)
assert SPEC and SPEC.loader
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)


SCHEMA = """
create table user (
  id integer primary key autoincrement,
  lower_name text not null,
  name text not null,
  email text not null,
  passwd text not null,
  salt text,
  passwd_hash_algo text,
  avatar text not null,
  avatar_email text not null,
  type integer default 0,
  is_active integer default 1,
  is_admin integer default 0,
  num_repos integer default 0,
  num_stars integer default 0,
  num_followers integer default 0,
  num_following integer default 0,
  use_custom_avatar integer default 0,
  prohibit_login integer default 0
);
create table repository (
  id integer primary key autoincrement,
  owner_id integer not null,
  name text not null
);
create table access_token (
  id integer primary key autoincrement,
  uid integer not null,
  name text not null,
  token_hash text not null
);
"""


class OwnerMigrationTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        root = pathlib.Path(self.temporary.name)
        self.old = root / "old.db"
        self.new = root / "new.db"
        self.receipt = root / "receipt.json"
        for database in (self.old, self.new):
            connection = sqlite3.connect(database)
            connection.executescript(SCHEMA)
            connection.commit()
            connection.close()

        connection = sqlite3.connect(self.old)
        connection.execute(
            """
            insert into user (
              lower_name, name, email, passwd, salt, passwd_hash_algo,
              avatar, avatar_email, is_active, is_admin, num_repos,
              num_stars, num_followers, num_following, use_custom_avatar
            ) values (?, ?, ?, ?, ?, ?, ?, ?, 1, 1, 12, 4, 3, 2, 1)
            """,
            (
                "bingshuo",
                "bingshuo",
                "owner@example.invalid",
                "preserved-password-hash",
                "preserved-salt",
                "pbkdf2$50000$50",
                "legacy-avatar",
                "avatar@example.invalid",
            ),
        )
        connection.execute(
            "insert into repository (owner_id, name) values (1, 'legacy-repo')"
        )
        connection.execute(
            "insert into access_token (uid, name, token_hash) values (1, 'legacy-token', 'secret-hash')"
        )
        connection.commit()
        connection.close()

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def test_migrates_only_owner_identity_and_preserves_password_hash(self) -> None:
        result = MODULE.migrate_owner_identity(self.old, self.new, self.receipt)
        self.assertEqual(result, "migrated")
        connection = sqlite3.connect(self.new)
        owner = connection.execute(
            """
            select lower_name, passwd, salt, passwd_hash_algo, is_active,
                   is_admin, num_repos, num_stars, num_followers,
                   num_following, use_custom_avatar
            from user
            """
        ).fetchone()
        self.assertEqual(
            owner,
            (
                "bingshuo",
                "preserved-password-hash",
                "preserved-salt",
                "pbkdf2$50000$50",
                1,
                1,
                0,
                0,
                0,
                0,
                0,
            ),
        )
        self.assertEqual(connection.execute("select count(*) from repository").fetchone()[0], 0)
        self.assertEqual(connection.execute("select count(*) from access_token").fetchone()[0], 0)
        connection.close()
        receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
        self.assertTrue(receipt["identity_only"])
        self.assertFalse(receipt["access_tokens_migrated"])
        self.assertFalse(receipt["repositories_migrated"])

    def test_is_idempotent(self) -> None:
        self.assertEqual(MODULE.migrate_owner_identity(self.old, self.new, self.receipt), "migrated")
        self.assertEqual(MODULE.migrate_owner_identity(self.old, self.new, self.receipt), "already-present")


class SharedRepositoryTests(unittest.TestCase):
    def test_configures_native_group_sharing_and_removes_only_known_hook(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            repository = pathlib.Path(temporary) / "channel.git"
            subprocess.run(["git", "init", "--bare", str(repository)], check=True, capture_output=True)
            hook = repository / "hooks" / "post-receive.d" / "guanghu-ice-heart-share"
            hook.parent.mkdir(parents=True)
            hook.write_text(
                "#!/bin/sh\nprintf '%s\\n' post_receive_permissions_reconciled\n",
                encoding="utf-8",
            )

            self.assertEqual(
                MODULE.configure_shared_channel_repository(repository),
                "configured",
            )
            configured = subprocess.run(
                [
                    "git",
                    "--git-dir",
                    str(repository),
                    "config",
                    "--get",
                    "core.sharedRepository",
                ],
                check=True,
                capture_output=True,
                text=True,
            ).stdout.strip()
            self.assertIn(configured, {"1", "group"})
            self.assertFalse(hook.exists())

    def test_refuses_unknown_hook_content(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            repository = pathlib.Path(temporary) / "channel.git"
            subprocess.run(["git", "init", "--bare", str(repository)], check=True, capture_output=True)
            hook = repository / "hooks" / "post-receive.d" / "guanghu-ice-heart-share"
            hook.parent.mkdir(parents=True)
            hook.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")

            with self.assertRaisesRegex(RuntimeError, "identity mismatch"):
                MODULE.configure_shared_channel_repository(repository)
            self.assertTrue(hook.exists())


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