fix: migrate legacy network_capability operation types on startup

George Papathanailrequested to merge
fix/operation-type-migration into develop

Merge request reports

src/open_exposure_gateway/adapters/database/core.py

+8 −3
Changes for src/open_exposure_gateway/adapters/database/core.py: 8 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -24,8 +24,9 @@ async def build_engine_and_session_maker( 
    return engine, session_maker





# create_all never alters existing tables; columns added after first release go here.

_ADDED_COLUMNS = (

# create_all never alters existing tables; columns added and values renamed after

# first release go here.

_STARTUP_STATEMENTS = (

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS started_at TIMESTAMP WITH TIME ZONE",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS ended_at TIMESTAMP WITH TIME ZONE",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS status_info VARCHAR(32)",
@@ -33,11 +34,15 @@ _ADDED_COLUMNS = ( 
    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS application_server JSONB",

    "ALTER TABLE app_registrations ADD COLUMN IF NOT EXISTS app_provider_id VARCHAR(255)",

    "ALTER TABLE app_registrations ADD COLUMN IF NOT EXISTS app_provider VARCHAR(64)",

    "UPDATE operations SET operation_type = 'qod_session' "

    "WHERE operation_type = 'network_capability'",

    "UPDATE operations SET operation_type = 'qod_session_deactivate' "

    "WHERE operation_type = 'network_capability_deactivate'",

)





async def schema_initialization(engine: AsyncEngine) -> None:

    async with engine.begin() as conn:

        await conn.run_sync(get_metadata().create_all)

        for statement in _ADDED_COLUMNS:

        for statement in _STARTUP_STATEMENTS:

            await conn.execute(text(statement))

tests/integration/test_postgres.py

+32 −0
Changes for tests/integration/test_postgres.py: 32 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -8,6 +8,7 @@ import pytest 
from sqlalchemy import text

from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker



from open_exposure_gateway.adapters.database.core import schema_initialization

from open_exposure_gateway.adapters.database.repos.app_instances import (

    SqlAppInstanceRepository,

)
@@ -210,6 +211,37 @@ async def test_operation_repo_updates_updated_at_on_second_save(db_engine: Async 
    assert updated.status == OperationStatus.COMPLETED





@pytest.mark.parametrize(

    ("legacy_value", "expected_type"),

    [

        ("network_capability", OperationType.QOD_SESSION),

        ("network_capability_deactivate", OperationType.QOD_SESSION_DEACTIVATE),

    ],

)

async def test_schema_initialization_renames_legacy_qod_operation_types(

    db_engine: AsyncEngine, legacy_value: str, expected_type: OperationType

) -> None:

    session_factory = async_sessionmaker(bind=db_engine, expire_on_commit=False)

    operation = _operation()

    operation.operation_type = expected_type



    async with session_factory() as session:

        await SqlOperationRepository(session).save(operation)

        await session.execute(

            text("UPDATE operations SET operation_type = :legacy WHERE operation_id = :id"),

            {"legacy": legacy_value, "id": operation.operation_id},

        )

        await session.commit()



    await schema_initialization(db_engine)



    async with session_factory() as session:

        loaded = await SqlOperationRepository(session).get_by_id(operation.operation_id)



    assert loaded is not None

    assert loaded.operation_type == expected_type





async def test_operation_repo_list_stale_filters_on_status_type_and_age(

    db_session: AsyncSession,

) -> None: