Coverage for app/backend/src/tests/test_public_trip_migration.py: 100%

1from datetime import timedelta 

2from importlib import import_module 

3 

4import pytest 

5from alembic.migration import MigrationContext 

6from alembic.operations import Operations 

7from sqlalchemy import select, text, update 

8from sqlalchemy.exc import IntegrityError 

9 

10from couchers.models.public_trips import PublicTrip 

11from couchers.utils import today 

12from tests.fixtures.db import generate_user 

13from tests.test_public_trips import _create_trip_directly, _make_node 

14 

15minimum_stay_migration = import_module("couchers.migrations.versions.0200_require_public_trip_minimum_stay") 

16 

17 

18@pytest.mark.parametrize("nights", [0, 1]) 

19def test_public_trip_minimum_stay_migration(db, testdb_engine, nights): 

20 user, _ = generate_user() 

21 node_id = _make_node() 

22 start = today() + timedelta(days=5) 

23 end = start + timedelta(days=nights) 

24 trip_id = _create_trip_directly(user.id, node_id, start, start + timedelta(days=1)) 

25 

26 with ( 

27 testdb_engine.connect().execution_options(isolation_level="READ COMMITTED") as connection, 

28 connection.begin() as transaction, 

29 ): 

30 connection.execute( 

31 text( 

32 "ALTER TABLE public_trips DROP CONSTRAINT ck_public_trips_valid_date_range, " 

33 "ADD CONSTRAINT ck_public_trips_valid_date_range CHECK (from_date <= to_date)" 

34 ) 

35 ) 

36 connection.execute(update(PublicTrip).where(PublicTrip.id == trip_id).values(to_date=end)) 

37 with Operations.context(MigrationContext.configure(connection)): 

38 if nights == 0: 

39 with pytest.raises(RuntimeError, match="without an overnight stay"): 

40 minimum_stay_migration.upgrade() 

41 assert ( 

42 "<=" 

43 in connection.execute( 

44 text( 

45 "SELECT pg_get_constraintdef(oid) FROM pg_constraint " 

46 "WHERE conrelid = 'public_trips'::regclass AND conname = 'ck_public_trips_valid_date_range'" 

47 ) 

48 ).scalar_one() 

49 ) 

50 else: 

51 minimum_stay_migration.upgrade() 

52 with pytest.raises(IntegrityError, match="ck_public_trips_valid_date_range"): 

53 with connection.begin_nested(): 

54 connection.execute(update(PublicTrip).where(PublicTrip.id == trip_id).values(to_date=start)) 

55 minimum_stay_migration.downgrade() 

56 with connection.begin_nested() as savepoint: 

57 connection.execute(update(PublicTrip).where(PublicTrip.id == trip_id).values(to_date=start)) 

58 savepoint.rollback() 

59 

60 assert connection.execute( 

61 select(PublicTrip.from_date, PublicTrip.to_date).where(PublicTrip.id == trip_id) 

62 ).one() == (start, end) 

63 transaction.rollback()