Coverage for slidge/db/alembic/versions/39ce1636571a_store_only_local_parts_of_jids.py: 92%

1"""Store only local parts of JIDs 

2 

3Revision ID: 39ce1636571a 

4Revises: bdf1c8b85ea3 

5Create Date: 2026-07-15 21:20:54.149210 

6 

7""" 

8 

9from collections.abc import Sequence 

10 

11import sqlalchemy as sa 

12from alembic import op 

13 

14# revision identifiers, used by Alembic. 

15revision: str = "39ce1636571a" 

16down_revision: str | None = "bdf1c8b85ea3" 

17branch_labels: str | Sequence[str] | None = None 

18depends_on: str | Sequence[str] | None = None 

19 

20 

21def upgrade() -> None: 

22 _update("contact") 

23 _update("room") 

24 

25 

26def _update(table: str) -> None: 

27 with op.batch_alter_table(table, schema=None) as batch_op: 

28 batch_op.add_column(sa.Column("jid_localpart", sa.String(), nullable=True)) 

29 

30 if op.get_bind().dialect.name == "postgresql": 

31 expr = "SUBSTRING(jid FROM 1 FOR POSITION('@' IN jid) - 1)" 

32 else: 

33 expr = "SUBSTR(jid, 1, INSTR(jid, '@') - 1)" 

34 op.execute(sa.text(f"UPDATE {table} SET jid_localpart = {expr}")) 

35 

36 with op.batch_alter_table(table, schema=None) as batch_op: 

37 batch_op.alter_column("jid_localpart", nullable=False) 

38 batch_op.create_unique_constraint( 

39 f"uq_{table}_user_account_id_jid_localpart", 

40 ["user_account_id", "jid_localpart"], 

41 ) 

42 with op.batch_alter_table(table, schema=None) as batch_op: 

43 batch_op.drop_constraint( 

44 batch_op.f(f"uq_{table}_user_account_id_jid"), type_="unique" 

45 ) 

46 batch_op.drop_column("jid") 

47 

48 

49def downgrade() -> None: 

50 raise RuntimeError("Downgrades are not permitted. Clear slidge's home dir.")