107 lines
4.4 KiB
Python
107 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from alembic import command
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account
|
|
from govoplan_access.backend.security.passwords import hash_password
|
|
from govoplan_core.db.migrations import alembic_config
|
|
|
|
|
|
class PasswordRecoveryMigrationTests(unittest.TestCase):
|
|
def test_release_and_development_upgrade_preserve_existing_credentials(self):
|
|
for track in ("release", "dev"):
|
|
with (
|
|
self.subTest(track=track),
|
|
tempfile.TemporaryDirectory(
|
|
prefix="govoplan-password-migration-"
|
|
) as directory,
|
|
):
|
|
url = f"sqlite:///{Path(directory) / 'isolated-upgrade.db'}"
|
|
config = alembic_config(
|
|
database_url=url, enabled_modules=("access",), migration_track=track
|
|
)
|
|
command.upgrade(config, "4f2a9c8e7b6d")
|
|
command.upgrade(config, "d8f1b4e7a0c3")
|
|
engine = create_engine(url)
|
|
try:
|
|
with Session(engine) as session:
|
|
session.add(
|
|
Account(
|
|
id="existing",
|
|
email="existing@example.test",
|
|
normalized_email="existing@example.test",
|
|
password_hash=hash_password("Existing-password"),
|
|
password_reset_required=True,
|
|
)
|
|
)
|
|
session.commit()
|
|
with engine.connect() as connection:
|
|
before = list(
|
|
connection.execute(
|
|
text("SELECT * FROM access_accounts")
|
|
).mappings()
|
|
)
|
|
tables = set(inspect(connection).get_table_names())
|
|
command.upgrade(config, "e9a2c5f8b1d4")
|
|
command.upgrade(config, "e9a2c5f8b1d4")
|
|
with engine.connect() as connection:
|
|
inspector = inspect(connection)
|
|
self.assertEqual(
|
|
tables | {"access_password_recoveries"},
|
|
set(inspector.get_table_names()),
|
|
)
|
|
self.assertEqual(
|
|
before,
|
|
list(
|
|
connection.execute(
|
|
text("SELECT * FROM access_accounts")
|
|
).mappings()
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
"id",
|
|
"account_id",
|
|
"issuer_account_id",
|
|
"issuer_membership_id",
|
|
"code_hash",
|
|
"expires_at",
|
|
"consumed_at",
|
|
"created_at",
|
|
"updated_at",
|
|
},
|
|
{
|
|
column["name"]
|
|
for column in inspector.get_columns(
|
|
"access_password_recoveries"
|
|
)
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
3,
|
|
len(
|
|
inspector.get_foreign_keys("access_password_recoveries")
|
|
),
|
|
)
|
|
self.assertIn(
|
|
["code_hash"],
|
|
[
|
|
constraint["column_names"]
|
|
for constraint in inspector.get_unique_constraints(
|
|
"access_password_recoveries"
|
|
)
|
|
],
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|