#!/usr/bin/env python3

from __future__ import annotations

import importlib.util
import json
import pathlib
import sqlite3
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")


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