142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
#!/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()
|