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

262 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-19 23:55 +0000

1""" 

2Hybrid properties are two implementations of one truth: a python body that runs on a loaded instance, 

3and a SQL expression that runs in the database. Nothing in the ORM checks that they agree, so they can 

4drift apart silently, which is how the lite_users strong verification bug survived for 21 months. 

5 

6This module discovers every hybrid on every model, builds a deliberately diverse population for it, 

7and asserts that the python value equals the value postgres computes, for every row. A hybrid that 

8cannot run in python at all (its body is written in terms of `func.now()` and friends, so evaluating 

9it on an instance yields a SQL expression rather than a value) is listed in SQL_ONLY, and we assert 

10that it really is inert in python rather than quietly returning something wrong. 

11 

12Adding a hybrid to a model without adding it here fails test_every_hybrid_is_covered. 

13""" 

14 

15from collections.abc import Callable 

16from datetime import date, timedelta 

17from typing import Any 

18 

19import pytest 

20from psycopg.types.range import TimestamptzRange 

21from sqlalchemy import inspect, select, update 

22from sqlalchemy.ext.hybrid import HybridExtensionType 

23from sqlalchemy.orm import Session 

24from sqlalchemy.sql.elements import ClauseElement 

25 

26from couchers.constants import GUIDELINES_VERSION, PHONE_VERIFICATION_LIFETIME, TOS_VERSION 

27from couchers.crypto import random_hex 

28from couchers.db import session_scope 

29from couchers.models import ( 

30 AccountDeletionToken, 

31 ActivenessProbe, 

32 ActivenessProbeStatus, 

33 BackgroundJob, 

34 BackgroundJobState, 

35 Base, 

36 ContributorForm, 

37 Conversation, 

38 Event, 

39 EventOccurrence, 

40 GroupChat, 

41 GroupChatRole, 

42 GroupChatSubscription, 

43 HostingStatus, 

44 HostRequest, 

45 HostRequestStatus, 

46 InitiatedUpload, 

47 LoginToken, 

48 ModerationObjectType, 

49 ModNote, 

50 Node, 

51 NodeType, 

52 PassportSex, 

53 PasswordResetToken, 

54 PostalVerificationAttempt, 

55 PostalVerificationStatus, 

56 SignupFlow, 

57 SleepingArrangement, 

58 StrongVerificationAttempt, 

59 StrongVerificationAttemptStatus, 

60 Thread, 

61 User, 

62 UserSession, 

63) 

64from couchers.moderation.utils import create_moderation 

65from couchers.utils import create_coordinate, create_polygon_lat_lng, now, to_multi 

66from tests.fixtures.db import generate_user, make_user_invisible 

67 

68# Hybrids whose body only makes sense in SQL: evaluating them on an instance yields a SQLAlchemy 

69# expression, which blows up the moment anything treats it as a value. They have no python 

70# implementation to disagree with, so there is nothing to compare -- but see 

71# test_sql_only_hybrids_are_inert_in_python, which holds them to being loudly, not quietly, unusable. 

72SQL_ONLY = { 

73 "BackgroundJob.ready_for_retry": "compares next_attempt_after against func.now()", 

74 "GroupChatSubscription.is_muted": "compares muted_until against func.now()", 

75 "InitiatedUpload.is_valid": "compares created/expiry against func.now()", 

76 "UserSession.is_valid": "compares created/expiry/last_seen against func.now() and a SQL interval", 

77} 

78 

79 

80def _label(model: type[Base], name: str) -> str: 

81 return f"{model.__name__}.{name}" 

82 

83 

84def _hybrids(extension_type: HybridExtensionType) -> list[tuple[type[Base], str]]: 

85 # `@x.inplace.expression` binds one hybrid to two names, the public one and the private one holding 

86 # the SQL expression, so dedupe on the descriptor itself and keep the name people write in queries 

87 found: dict[tuple[type[Base], int], str] = {} 

88 for mapper in Base.registry.mappers: 

89 for name, descriptor in mapper.all_orm_descriptors.items(): 

90 if descriptor.extension_type != extension_type: 

91 continue 

92 key = (mapper.class_, id(descriptor)) 

93 if key not in found or found[key].startswith("_"): 

94 found[key] = name 

95 return sorted(((model, name) for (model, _), name in found.items()), key=lambda pair: _label(*pair)) 

96 

97 

98HYBRID_PROPERTIES = _hybrids(HybridExtensionType.HYBRID_PROPERTY) 

99HYBRID_METHODS = _hybrids(HybridExtensionType.HYBRID_METHOD) 

100 

101Population = Callable[[], None] 

102POPULATIONS: dict[type[Base], Population] = {} 

103 

104 

105def _populates(model: type[Base]) -> Callable[[Population], Population]: 

106 def decorator(population: Population) -> Population: 

107 POPULATIONS[model] = population 

108 return population 

109 

110 return decorator 

111 

112 

113@pytest.fixture(autouse=True) 

114def _(testconfig): 

115 pass 

116 

117 

118## Populations: one per model, each diverse enough that every hybrid on the model takes at least two 

119## different values across the rows (test_hybrid_agrees_with_sql asserts that). 

120 

121 

122@_populates(User) 

123def _populate_users() -> None: 

124 generate_user() 

125 generate_user(accepted_tos=TOS_VERSION - 1) 

126 generate_user(accepted_community_guidelines=GUIDELINES_VERSION - 1) 

127 generate_user(max_guests=3, sleeping_arrangement=SleepingArrangement.private) 

128 generate_user(max_guests=3, sleeping_arrangement=None) 

129 banned, _ = generate_user() 

130 make_user_invisible(banned.id) 

131 generate_user(delete_user=True) 

132 shadowed, _ = generate_user() 

133 relocating, _ = generate_user() 

134 noted, _ = generate_user() 

135 acknowledged, _ = generate_user() 

136 probed, _ = generate_user() 

137 responded, _ = generate_user() 

138 phone_verified, _ = generate_user() 

139 phone_stale, _ = generate_user() 

140 code_sent, _ = generate_user() 

141 moderator, _ = generate_user(is_superuser=True) 

142 

143 with session_scope() as session: 

144 session.execute(update(User).where(User.id == shadowed.id).values(shadowed_at=now() - timedelta(days=1))) 

145 session.execute(update(User).where(User.id == relocating.id).values(needs_to_update_location=True)) 

146 session.add( 

147 ModNote(user_id=noted.id, creator_user_id=moderator.id, internal_id="pending", note_content="Be nice") 

148 ) 

149 session.add( 

150 ModNote( 

151 user_id=acknowledged.id, 

152 creator_user_id=moderator.id, 

153 internal_id="acknowledged", 

154 note_content="Be nice", 

155 acknowledged=now() - timedelta(days=1), 

156 ) 

157 ) 

158 session.add(ActivenessProbe(user_id=probed.id)) 

159 session.add( 

160 ActivenessProbe( 

161 user_id=responded.id, 

162 responded=now() - timedelta(days=1), 

163 response=ActivenessProbeStatus.still_active, 

164 ) 

165 ) 

166 # a phone number is required whenever the verification is: see the phone_verified_conditions constraint 

167 session.execute( 

168 update(User) 

169 .where(User.id == phone_verified.id) 

170 .values(phone="+46701740601", phone_verification_verified=now() - timedelta(days=1)) 

171 ) 

172 session.execute( 

173 update(User) 

174 .where(User.id == phone_stale.id) 

175 .values( 

176 phone="+46701740602", 

177 phone_verification_verified=now() - PHONE_VERIFICATION_LIFETIME - timedelta(days=1), 

178 ) 

179 ) 

180 session.execute( 

181 update(User).where(User.id == code_sent.id).values(phone_verification_sent=now() - timedelta(hours=1)) 

182 ) 

183 

184 

185@_populates(ModNote) 

186@_populates(ActivenessProbe) 

187def _populate_user_flags() -> None: 

188 _populate_users() 

189 

190 

191WOMAN_BIRTHDATE = date(1990, 3, 4) 

192MAN_BIRTHDATE = date(1985, 11, 22) 

193 

194 

195@_populates(StrongVerificationAttempt) 

196def _populate_strong_verification_attempts() -> None: 

197 woman, _ = generate_user(gender="Woman", birthdate=WOMAN_BIRTHDATE) 

198 man, _ = generate_user(gender="Man", birthdate=MAN_BIRTHDATE) 

199 

200 with session_scope() as session: 

201 # succeeded and unexpired: the only shape that verifies anyone 

202 session.add(_attempt(woman.id, 1, StrongVerificationAttemptStatus.succeeded, expiry_days=365)) 

203 # succeeded but the passport has expired 

204 session.add(_attempt(man.id, 2, StrongVerificationAttemptStatus.succeeded, expiry_days=-1)) 

205 # same passport data as the first attempt, but the data has since been deleted 

206 session.add( 

207 _attempt(woman.id, 3, StrongVerificationAttemptStatus.deleted, expiry_days=365, has_full_data=False) 

208 ) 

209 # never got any data at all 

210 session.add(_attempt(man.id, 4, StrongVerificationAttemptStatus.failed, expiry_days=None)) 

211 

212 

213def _attempt( 

214 user_id: int, 

215 n: int, 

216 status: StrongVerificationAttemptStatus, 

217 *, 

218 expiry_days: int | None, 

219 has_full_data: bool = True, 

220) -> StrongVerificationAttempt: 

221 """A strong verification attempt in one of the shapes the check constraints allow.""" 

222 has_minimal_data = expiry_days is not None 

223 # full data implies minimal data 

224 has_full_data = has_full_data and has_minimal_data 

225 return StrongVerificationAttempt( 

226 verification_attempt_token=f"verification_attempt_token_{n}", 

227 user_id=user_id, 

228 status=status, 

229 has_full_data=has_full_data, 

230 passport_encrypted_data=b"not real" if has_full_data else None, 

231 # the passport always describes the first user, so pairing it with the second is a real mismatch 

232 passport_date_of_birth=WOMAN_BIRTHDATE if has_full_data else None, 

233 passport_sex=PassportSex.female if has_full_data else None, 

234 has_minimal_data=has_minimal_data, 

235 passport_expiry_date=date.today() + timedelta(days=expiry_days) if expiry_days is not None else None, 

236 passport_nationality="UTO" if has_minimal_data else None, 

237 passport_last_three_document_chars=f"{n:03}" if has_minimal_data else None, 

238 iris_token=f"iris_token_{n}", 

239 iris_session_id=n, 

240 ) 

241 

242 

243@_populates(PostalVerificationAttempt) 

244def _populate_postal_verification_attempts() -> None: 

245 verified, _ = generate_user() 

246 cancelled, _ = generate_user() 

247 pending, _ = generate_user() 

248 

249 with session_scope() as session: 

250 session.add( 

251 PostalVerificationAttempt( 

252 user_id=verified.id, 

253 status=PostalVerificationStatus.succeeded, 

254 address_line_1="1 Test Street", 

255 city="Testing city", 

256 country_code="US", 

257 verification_code="ABC123", 

258 postcard_sent_at=now() - timedelta(days=10), 

259 verified_at=now() - timedelta(days=1), 

260 ) 

261 ) 

262 session.add( 

263 PostalVerificationAttempt( 

264 user_id=cancelled.id, 

265 status=PostalVerificationStatus.cancelled, 

266 address_line_1="2 Test Street", 

267 city="Testing city", 

268 country_code="US", 

269 ) 

270 ) 

271 session.add( 

272 PostalVerificationAttempt( 

273 user_id=pending.id, 

274 status=PostalVerificationStatus.pending_address_confirmation, 

275 address_line_1="3 Test Street", 

276 city="Testing city", 

277 country_code="US", 

278 ) 

279 ) 

280 

281 

282@_populates(HostRequest) 

283def _populate_host_requests() -> None: 

284 surfer, _ = generate_user() 

285 host, _ = generate_user() 

286 today = date.today() 

287 

288 with session_scope() as session: 

289 # the stay just ended, so the reference window is open 

290 _host_request(session, surfer.id, host.id, HostRequestStatus.accepted, today - timedelta(days=1)) 

291 # the reference window closed 14 days after the stay 

292 _host_request(session, surfer.id, host.id, HostRequestStatus.confirmed, today - timedelta(days=30)) 

293 # the stay hasn't happened yet 

294 _host_request(session, surfer.id, host.id, HostRequestStatus.confirmed, today + timedelta(days=30)) 

295 # never went ahead 

296 _host_request(session, surfer.id, host.id, HostRequestStatus.rejected, today - timedelta(days=1)) 

297 

298 

299def _host_request( 

300 session: Session, surfer_id: int, host_id: int, status: HostRequestStatus, to_date: date 

301) -> HostRequest: 

302 conversation = Conversation() 

303 session.add(conversation) 

304 session.flush() 

305 moderation_state = create_moderation( 

306 session=session, 

307 object_type=ModerationObjectType.host_request, 

308 object_id=conversation.id, 

309 creator_user_id=surfer_id, 

310 ) 

311 host_request = HostRequest( 

312 conversation_id=conversation.id, 

313 initiator_user_id=surfer_id, 

314 recipient_user_id=host_id, 

315 moderation_state_id=moderation_state.id, 

316 from_date=to_date - timedelta(days=2), 

317 to_date=to_date, 

318 status=status, 

319 hosting_city="Testing city", 

320 hosting_location=create_coordinate(40.7108, -73.9740), 

321 hosting_radius=100, 

322 ) 

323 session.add(host_request) 

324 session.flush() 

325 return host_request 

326 

327 

328@_populates(EventOccurrence) 

329def _populate_event_occurrences() -> None: 

330 creator, _ = generate_user() 

331 

332 with session_scope() as session: 

333 node = Node( 

334 geom=to_multi(create_polygon_lat_lng([[0, 0], [0, 2], [2, 2], [2, 0], [0, 0]])), 

335 node_type=NodeType.world, 

336 ) 

337 session.add(node) 

338 thread = Thread() 

339 session.add(thread) 

340 session.flush() 

341 event = Event( 

342 parent_node_id=node.id, 

343 title="Testing event", 

344 creator_user_id=creator.id, 

345 owner_user_id=creator.id, 

346 thread_id=thread.id, 

347 ) 

348 session.add(event) 

349 session.flush() 

350 

351 # occurrences may not overlap within an event 

352 for days, hours in [(1, 2), (10, 3)]: 

353 start = now() + timedelta(days=days) 

354 

355 def create_occurrence(moderation_state_id: int, start=start, hours=hours) -> int: 

356 occurrence = EventOccurrence( 

357 event_id=event.id, 

358 moderation_state_id=moderation_state_id, 

359 creator_user_id=creator.id, 

360 content="Testing event occurrence", 

361 geom=create_coordinate(1, 1), 

362 address="Somewhere", 

363 timezone="Etc/UTC", 

364 during=TimestamptzRange(start, start + timedelta(hours=hours)), 

365 ) 

366 session.add(occurrence) 

367 session.flush() 

368 return occurrence.id 

369 

370 create_moderation( 

371 session=session, 

372 object_type=ModerationObjectType.event_occurrence, 

373 object_id=create_occurrence, 

374 creator_user_id=creator.id, 

375 ) 

376 

377 

378@_populates(GroupChatSubscription) 

379def _populate_group_chat_subscriptions() -> None: 

380 creator, _ = generate_user() 

381 other, _ = generate_user() 

382 

383 with session_scope() as session: 

384 conversation = Conversation() 

385 session.add(conversation) 

386 session.flush() 

387 moderation_state = create_moderation( 

388 session=session, 

389 object_type=ModerationObjectType.group_chat, 

390 object_id=conversation.id, 

391 creator_user_id=creator.id, 

392 ) 

393 session.add( 

394 GroupChat( 

395 conversation_id=conversation.id, 

396 creator_id=creator.id, 

397 is_dm=True, 

398 moderation_state_id=moderation_state.id, 

399 ) 

400 ) 

401 muted = GroupChatSubscription(user_id=creator.id, group_chat_id=conversation.id, role=GroupChatRole.admin) 

402 session.add(muted) 

403 session.add( 

404 GroupChatSubscription(user_id=other.id, group_chat_id=conversation.id, role=GroupChatRole.participant) 

405 ) 

406 session.flush() 

407 session.execute( 

408 update(GroupChatSubscription) 

409 .where(GroupChatSubscription.id == muted.id) 

410 .values(muted_until=now() + timedelta(days=7)) 

411 ) 

412 

413 

414@_populates(UserSession) 

415def _populate_user_sessions() -> None: 

416 user, _ = generate_user() 

417 with session_scope() as session: 

418 session.add(UserSession(token=random_hex(32), user_id=user.id, long_lived=True, is_api_key=True)) 

419 session.add( 

420 UserSession(token=random_hex(32), user_id=user.id, long_lived=False, is_api_key=False, deleted=now()) 

421 ) 

422 

423 

424@_populates(LoginToken) 

425def _populate_login_tokens() -> None: 

426 user, _ = generate_user() 

427 with session_scope() as session: 

428 session.add(LoginToken(token=random_hex(32), user_id=user.id, expiry=now() + timedelta(hours=1))) 

429 session.add(LoginToken(token=random_hex(32), user_id=user.id, expiry=now() - timedelta(hours=1))) 

430 

431 

432@_populates(PasswordResetToken) 

433def _populate_password_reset_tokens() -> None: 

434 user, _ = generate_user() 

435 with session_scope() as session: 

436 session.add(PasswordResetToken(token=random_hex(32), user_id=user.id, expiry=now() + timedelta(hours=1))) 

437 session.add(PasswordResetToken(token=random_hex(32), user_id=user.id, expiry=now() - timedelta(hours=1))) 

438 

439 

440@_populates(AccountDeletionToken) 

441def _populate_account_deletion_tokens() -> None: 

442 user, _ = generate_user() 

443 with session_scope() as session: 

444 session.add(AccountDeletionToken(token=random_hex(32), user_id=user.id, expiry=now() + timedelta(hours=1))) 

445 session.add(AccountDeletionToken(token=random_hex(32), user_id=user.id, expiry=now() - timedelta(hours=1))) 

446 

447 

448@_populates(InitiatedUpload) 

449def _populate_initiated_uploads() -> None: 

450 user, _ = generate_user() 

451 with session_scope() as session: 

452 session.add( 

453 InitiatedUpload( 

454 key=random_hex(32), 

455 created=now() - timedelta(hours=1), 

456 expiry=now() + timedelta(hours=1), 

457 initiator_user_id=user.id, 

458 ) 

459 ) 

460 session.add( 

461 InitiatedUpload( 

462 key=random_hex(32), 

463 created=now() - timedelta(hours=2), 

464 expiry=now() - timedelta(hours=1), 

465 initiator_user_id=user.id, 

466 ) 

467 ) 

468 

469 

470@_populates(ContributorForm) 

471def _populate_contributor_forms() -> None: 

472 user, _ = generate_user() 

473 with session_scope() as session: 

474 session.add(ContributorForm(user_id=user.id, contribute_ways=[])) 

475 session.add(ContributorForm(user_id=user.id, contribute_ways=["community"])) 

476 session.add(ContributorForm(user_id=user.id, contribute_ways=[], ideas="I have one")) 

477 

478 

479@_populates(SignupFlow) 

480def _populate_signup_flows() -> None: 

481 with session_scope() as session: 

482 # a flow that has been completed all the way through 

483 session.add( 

484 SignupFlow( 

485 name="Completed", 

486 email="completed@couchers.org.invalid", 

487 flow_token=random_hex(32), 

488 email_verified=True, 

489 email_token=random_hex(32), 

490 email_token_expiry=now() + timedelta(hours=1), 

491 username="completed", 

492 birthdate=date(1990, 1, 1), 

493 gender="Woman", 

494 hosting_status=HostingStatus.cant_host, 

495 city="Testing city", 

496 geom=create_coordinate(40.7108, -73.9740), 

497 geom_radius=100, 

498 accepted_tos=TOS_VERSION, 

499 opt_out_of_newsletter=False, 

500 filled_motivations=True, 

501 ) 

502 ) 

503 # the email token has expired, and the account details were never filled in 

504 session.add( 

505 SignupFlow( 

506 name="Expired", 

507 email="expired@couchers.org.invalid", 

508 flow_token=random_hex(32), 

509 email_token=random_hex(32), 

510 email_token_expiry=now() - timedelta(hours=1), 

511 ) 

512 ) 

513 # never got as far as being sent an email 

514 session.add( 

515 SignupFlow( 

516 name="Fresh", 

517 email="fresh@couchers.org.invalid", 

518 flow_token=random_hex(32), 

519 ) 

520 ) 

521 

522 for flow in session.execute(select(SignupFlow)).scalars().all(): 

523 if flow.name == "Completed": 

524 flow.accepted_community_guidelines = GUIDELINES_VERSION 

525 

526 

527@_populates(BackgroundJob) 

528def _populate_background_jobs() -> None: 

529 with session_scope() as session: 

530 session.add(BackgroundJob(job_type="dummy_job", payload=b"")) 

531 session.add(BackgroundJob(job_type="dummy_job", payload=b"", state=BackgroundJobState.completed)) 

532 session.add(BackgroundJob(job_type="dummy_job", payload=b"", state=BackgroundJobState.error, try_count=5)) 

533 

534 

535## The tests 

536 

537 

538def test_every_hybrid_is_covered() -> None: 

539 """A new hybrid on a model has to bring a population with it, or it goes untested.""" 

540 models = {model for model, _ in HYBRID_PROPERTIES + HYBRID_METHODS} 

541 assert models - POPULATIONS.keys() == set(), "these models have hybrids but no population" 

542 assert POPULATIONS.keys() - models == set(), "these populations are for models without hybrids" 

543 assert SQL_ONLY.keys() <= {_label(model, name) for model, name in HYBRID_PROPERTIES}, "stale SQL_ONLY entries" 

544 assert {model for model, _ in HYBRID_METHODS} == {StrongVerificationAttempt}, ( 

545 "test_hybrid_method_agrees_with_sql only knows how to bind a User as the subject" 

546 ) 

547 

548 

549COMPARABLE = [pair for pair in HYBRID_PROPERTIES if _label(*pair) not in SQL_ONLY] 

550SQL_ONLY_PROPERTIES = [pair for pair in HYBRID_PROPERTIES if _label(*pair) in SQL_ONLY] 

551 

552 

553@pytest.mark.parametrize(("model", "name"), COMPARABLE, ids=[_label(*pair) for pair in COMPARABLE]) 

554def test_hybrid_agrees_with_sql(db, model: type[Base], name: str) -> None: 

555 POPULATIONS[model]() 

556 

557 with session_scope() as session: 

558 mapper = inspect(model) 

559 sql_values = _sql_values(session, model, name) 

560 

561 for instance in session.execute(select(model)).scalars(): 

562 key = tuple(mapper.primary_key_from_instance(instance)) 

563 python_value = getattr(instance, name) 

564 assert _agree(python_value, sql_values[key]), ( 

565 f"{_label(model, name)} disagrees on {key}: python says {python_value!r}, " 

566 f"postgres says {sql_values[key]!r}" 

567 ) 

568 

569 

570@pytest.mark.parametrize(("model", "name"), SQL_ONLY_PROPERTIES, ids=[_label(*pair) for pair in SQL_ONLY_PROPERTIES]) 

571def test_sql_only_hybrids_are_inert_in_python(db, model: type[Base], name: str) -> None: 

572 """ 

573 A hybrid with no python implementation must fail loudly rather than answer wrongly: reading it off 

574 an instance either raises, or hands back a SQL expression that raises the moment anything reads it 

575 as a boolean. If one of these ever starts returning a value, it needs comparing, not listing here. 

576 """ 

577 POPULATIONS[model]() 

578 

579 with session_scope() as session: 

580 _sql_values(session, model, name) 

581 instances = session.execute(select(model)).scalars().all() 

582 assert instances 

583 for instance in instances: 

584 with pytest.raises(TypeError): 

585 value = getattr(instance, name) 

586 assert isinstance(value, ClauseElement), f"{_label(model, name)} returns a python value now" 

587 bool(value) 

588 

589 

590@pytest.mark.parametrize(("model", "name"), HYBRID_METHODS, ids=[_label(model, name) for model, name in HYBRID_METHODS]) 

591def test_hybrid_method_agrees_with_sql(db, model: type[Base], name: str) -> None: 

592 """ 

593 These take a subject, so they have two forms that have to agree: evaluated in python on a pair of 

594 instances, and evaluated in SQL over the subject's table. Every pair is checked, not just the 

595 matching ones: the lite_users bug was a query that reported the right answer for the pairs it was 

596 meant to cover and a wrong one for everybody else. 

597 """ 

598 POPULATIONS[model]() 

599 

600 with session_scope() as session: 

601 mapper = inspect(model) 

602 instances = session.execute(select(model)).scalars().all() 

603 users = session.execute(select(User).order_by(User.id)).scalars().all() 

604 assert len(instances) >= 2 and len(users) >= 2 

605 

606 seen = set() 

607 for instance in instances: 

608 key = tuple(mapper.primary_key_from_instance(instance)) 

609 for user in users: 

610 python_value = getattr(instance, name)(user) 

611 # the subject comes from the users table, exactly as it does in a real query; the join 

612 # pins it to this one user so the row is the pair under test 

613 sql_value = session.execute( 

614 select(getattr(model, name)(User)) 

615 .select_from(model) 

616 .join(User, User.id == user.id) 

617 .where(*(c == v for c, v in zip(mapper.primary_key, key))) 

618 ).scalar_one() 

619 assert _agree(python_value, sql_value), ( 

620 f"{_label(model, name)} on {key} against user {user.id}: python says " 

621 f"{python_value!r}, postgres says {sql_value!r}" 

622 ) 

623 seen.add(bool(sql_value)) 

624 

625 assert seen == {True, False}, f"{_label(model, name)} takes the same value on every pair" 

626 

627 

628def _sql_values(session: Session, model: type[Base], name: str) -> dict[tuple[Any, ...], Any]: 

629 """The hybrid as postgres computes it, per row, and a check that the population actually varies it.""" 

630 mapper = inspect(model) 

631 values = { 

632 tuple(row[:-1]): row[-1] for row in session.execute(select(*mapper.primary_key, getattr(model, name))).all() 

633 } 

634 assert len(values) >= 2, "the population needs at least two rows to be worth comparing" 

635 assert len(set(values.values())) >= 2, ( 

636 f"{_label(model, name)} takes the same value on every row: the population doesn't exercise it" 

637 ) 

638 return values 

639 

640 

641def _agree(python_value: Any, sql_value: Any) -> bool: 

642 """ 

643 Postgres computes a predicate over a NULL column as NULL, which is falsy everywhere these hybrids 

644 are used (WHERE, AND, OR), so python's False agrees with it. A python True against a NULL does not. 

645 """ 

646 if sql_value is None: 

647 return python_value is False 

648 return bool(python_value == sql_value)