Coverage for app/backend/src/couchers/models/ota.py: 100%

1import enum 

2from datetime import datetime 

3from typing import TYPE_CHECKING 

4 

5from sqlalchemy import ( 

6 BigInteger, 

7 CheckConstraint, 

8 DateTime, 

9 Enum, 

10 ForeignKey, 

11 Index, 

12 String, 

13 UniqueConstraint, 

14 func, 

15) 

16from sqlalchemy.orm import Mapped, mapped_column, relationship 

17 

18from couchers.models.base import Base 

19 

20if TYPE_CHECKING: 

21 from couchers.models.users import User 

22 

23 

24class OTAPlatform(enum.Enum): 

25 ios = enum.auto() 

26 android = enum.auto() 

27 

28 

29class OTAPackage(Base, kw_only=True): 

30 # The signed manifest bytes live on the CDN under `version`; this row only records which bundle is 

31 # available and how recent it is, so the backend can resolve a request and fetch the bytes verbatim. 

32 __tablename__ = "ota_packages" 

33 

34 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

35 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

36 

37 creator_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

38 

39 platform: Mapped[OTAPlatform] = mapped_column(Enum(OTAPlatform)) 

40 # The manifest's runtimeVersion / build's expo-runtime-version. A build only accepts a manifest whose 

41 # runtimeVersion equals its own, so (platform, fingerprint) is the compatibility key. 

42 fingerprint: Mapped[str] = mapped_column(String) 

43 # The CDN path component the signed manifest is published under, e.g. v1.3.<commit>.<sha>. 

44 version: Mapped[str] = mapped_column(String) 

45 # The manifest's createdAt: the publish/stamp time used to order rollouts. A rollback rolls forward by 

46 # republishing the good bundle re-stamped with a newer createdAt so it sorts newest. 

47 manifest_created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) 

48 # NOT unique: a re-stamped rollback reuses the same bundle content and so can repeat an id. 

49 manifest_id: Mapped[str] = mapped_column(String) 

50 

51 # Stops handing this bundle to new check-ins; can't reclaim devices already on it (they only move 

52 # forward in createdAt), so it's a stop-gap until a re-stamped rollback is published. 

53 banned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None) 

54 banned_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None) 

55 banned_reason: Mapped[str | None] = mapped_column(String, default=None) 

56 

57 creator_user: Mapped[User] = relationship(init=False, foreign_keys="OTAPackage.creator_user_id") 

58 banned_by_user: Mapped[User | None] = relationship(init=False, foreign_keys="OTAPackage.banned_by_user_id") 

59 

60 __table_args__ = ( 

61 UniqueConstraint("platform", "version", name="uq_ota_packages_platform_version"), 

62 Index("ix_ota_packages_resolve", "platform", "fingerprint", "manifest_created_at"), 

63 # All three ban columns move together: either the package isn't banned, or every audit field 

64 # is filled in. Bans are irreversible (rolled forward by republishing) so the reason is 

65 # required. 

66 CheckConstraint( 

67 "(banned_at IS NULL AND banned_by_user_id IS NULL AND banned_reason IS NULL) " 

68 "OR (banned_at IS NOT NULL AND banned_by_user_id IS NOT NULL AND banned_reason IS NOT NULL)", 

69 name="ck_ota_packages_ban_columns_consistent", 

70 ), 

71 )