Coverage for app/backend/src/tests/test_hybrid

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.

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.

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

13"""

15from collections.abc import Callable

16from datetime import date, timedelta

17from typing import Any

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

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

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}

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

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

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))

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## Populations: one per model, each diverse enough that every hybrid on the model takes at least two

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

115

116

117@_populates(User)

118def _populate_users() -> None:

119 generate_user()

120 generate_user(accepted_tos=TOS_VERSION - 1)

121 generate_user(accepted_community_guidelines=GUIDELINES_VERSION - 1)

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

123 generate_user(max_guests=3, sleeping_arrangement=None)

124 banned, _ = generate_user()

125 make_user_invisible(banned.id)

126 generate_user(delete_user=True)

127 shadowed, _ = generate_user()

128 relocating, _ = generate_user()

129 noted, _ = generate_user()

130 acknowledged, _ = generate_user()

131 probed, _ = generate_user()

132 responded, _ = generate_user()

133 phone_verified, _ = generate_user()

134 phone_stale, _ = generate_user()

135 code_sent, _ = generate_user()

136 moderator, _ = generate_user(is_superuser=True)

137

138 with session_scope() as session:

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

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

141 session.add(

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

143 )

144 session.add(

145 ModNote(

146 user_id=acknowledged.id,

147 creator_user_id=moderator.id,

148 internal_id="acknowledged",

149 note_content="Be nice",

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

151 )

152 )

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

154 session.add(

155 ActivenessProbe(

156 user_id=responded.id,

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

158 response=ActivenessProbeStatus.still_active,

159 )

160 )

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

162 session.execute(

163 update(User)

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

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

166 )

167 session.execute(

168 update(User)

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

170 .values(

171 phone="+46701740602",

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

173 )

174 )

175 session.execute(

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

177 )

178

179

180@_populates(ModNote)

181@_populates(ActivenessProbe)

182def _populate_user_flags() -> None:

183 _populate_users()

184

185

186WOMAN_BIRTHDATE = date(1990, 3, 4)

187MAN_BIRTHDATE = date(1985, 11, 22)

188

189

190@_populates(StrongVerificationAttempt)

191def _populate_strong_verification_attempts() -> None:

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

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

194

195 with session_scope() as session:

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

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

198 # succeeded but the passport has expired

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

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

201 session.add(

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

203 )

204 # never got any data at all

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

206

207

208def _attempt(

209 user_id: int,

210 n: int,

211 status: StrongVerificationAttemptStatus,

212 *,

213 expiry_days: int | None,

214 has_full_data: bool = True,

215) -> StrongVerificationAttempt:

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

217 has_minimal_data = expiry_days is not None

218 # full data implies minimal data

219 has_full_data = has_full_data and has_minimal_data

220 return StrongVerificationAttempt(

221 verification_attempt_token=f"verification_attempt_token_{n}",

222 user_id=user_id,

223 status=status,

224 has_full_data=has_full_data,

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

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

227 passport_date_of_birth=WOMAN_BIRTHDATE if has_full_data else None,

228 passport_sex=PassportSex.female if has_full_data else None,

229 has_minimal_data=has_minimal_data,

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

231 passport_nationality="UTO" if has_minimal_data else None,

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

233 iris_token=f"iris_token_{n}",

234 iris_session_id=n,

235 )

236

237

238@_populates(PostalVerificationAttempt)

239def _populate_postal_verification_attempts() -> None:

240 verified, _ = generate_user()

241 cancelled, _ = generate_user()

242 pending, _ = generate_user()

243

244 with session_scope() as session:

245 session.add(

246 PostalVerificationAttempt(

247 user_id=verified.id,

248 status=PostalVerificationStatus.succeeded,

249 address_line_1="1 Test Street",

250 city="Testing city",

251 country_code="US",

252 verification_code="ABC123",

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

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

255 )

256 )

257 session.add(

258 PostalVerificationAttempt(

259 user_id=cancelled.id,

260 status=PostalVerificationStatus.cancelled,

261 address_line_1="2 Test Street",

262 city="Testing city",

263 country_code="US",

264 )

265 )

266 session.add(

267 PostalVerificationAttempt(

268 user_id=pending.id,

269 status=PostalVerificationStatus.pending_address_confirmation,

270 address_line_1="3 Test Street",

271 city="Testing city",

272 country_code="US",

273 )

274 )

275

276

277@_populates(HostRequest)

278def _populate_host_requests() -> None:

279 surfer, _ = generate_user()

280 host, _ = generate_user()

281 today = date.today()

282

283 with session_scope() as session:

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

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

286 # the reference window closed 14 days after the stay

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

288 # the stay hasn't happened yet

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

290 # never went ahead

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

292

293

294def _host_request(

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

296) -> HostRequest:

297 conversation = Conversation()

298 session.add(conversation)

299 session.flush()

300 moderation_state = create_moderation(

301 session=session,

302 object_type=ModerationObjectType.host_request,

303 object_id=conversation.id,

304 creator_user_id=surfer_id,

305 )

306 host_request = HostRequest(

307 conversation_id=conversation.id,

308 initiator_user_id=surfer_id,

309 recipient_user_id=host_id,

310 moderation_state_id=moderation_state.id,

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

312 to_date=to_date,

313 status=status,

314 hosting_city="Testing city",

315 hosting_location=create_coordinate(40.7108, -73.9740),

316 hosting_radius=100,

317 )

318 session.add(host_request)

319 session.flush()

320 return host_request

321

322

323@_populates(EventOccurrence)

324def _populate_event_occurrences() -> None:

325 creator, _ = generate_user()

326

327 with session_scope() as session:

328 node = Node(

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

330 node_type=NodeType.world,

331 )

332 session.add(node)

333 session.flush()

334 event = Event(

335 parent_node_id=node.id,

336 title="Testing event",

337 creator_user_id=creator.id,

338 owner_user_id=creator.id,

339 )

340 session.add(event)

341 session.flush()

342

343 # occurrences may not overlap within an event

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

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

346

347 thread = Thread()

348 session.add(thread)

349 session.flush()

350

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

352 occurrence = EventOccurrence(

353 event_id=event.id,

354 moderation_state_id=moderation_state_id,

355 creator_user_id=creator.id,

356 content="Testing event occurrence",

357 geom=create_coordinate(1, 1),

358 address="Somewhere",

359 timezone="Etc/UTC",

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

361 thread_id=thread.id,

362 )

363 session.add(occurrence)

364 session.flush()

365 return occurrence.id

366

367 create_moderation(

368 session=session,

369 object_type=ModerationObjectType.event_occurrence,

370 object_id=create_occurrence,

371 creator_user_id=creator.id,

372 )

373

374

375@_populates(GroupChatSubscription)

376def _populate_group_chat_subscriptions() -> None:

377 creator, _ = generate_user()

378 other, _ = generate_user()

379

380 with session_scope() as session:

381 conversation = Conversation()

382 session.add(conversation)

383 session.flush()

384 moderation_state = create_moderation(

385 session=session,

386 object_type=ModerationObjectType.group_chat,

387 object_id=conversation.id,

388 creator_user_id=creator.id,

389 )

390 session.add(

391 GroupChat(

392 conversation_id=conversation.id,

393 creator_id=creator.id,

394 is_dm=True,

395 moderation_state_id=moderation_state.id,

396 )

397 )

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

399 session.add(muted)

400 session.add(

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

402 )

403 session.flush()

404 session.execute(

405 update(GroupChatSubscription)

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

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

408 )

409

410

411@_populates(UserSession)

412def _populate_user_sessions() -> None:

413 user, _ = generate_user()

414 with session_scope() as session:

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

416 session.add(

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

418 )

419

420

421@_populates(LoginToken)

422def _populate_login_tokens() -> None:

423 user, _ = generate_user()

424 with session_scope() as session:

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

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

427

428

429@_populates(PasswordResetToken)

430def _populate_password_reset_tokens() -> None:

431 user, _ = generate_user()

432 with session_scope() as session:

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

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

435

436

437@_populates(AccountDeletionToken)

438def _populate_account_deletion_tokens() -> None:

439 user, _ = generate_user()

440 with session_scope() as session:

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

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

443

444

445@_populates(InitiatedUpload)

446def _populate_initiated_uploads() -> None:

447 user, _ = generate_user()

448 with session_scope() as session:

449 session.add(

450 InitiatedUpload(

451 key=random_hex(32),

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

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

454 initiator_user_id=user.id,

455 )

456 )

457 session.add(

458 InitiatedUpload(

459 key=random_hex(32),

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

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

462 initiator_user_id=user.id,

463 )

464 )

465

466

467@_populates(ContributorForm)

468def _populate_contributor_forms() -> None:

469 user, _ = generate_user()

470 with session_scope() as session:

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

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

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

474

475

476@_populates(SignupFlow)

477def _populate_signup_flows() -> None:

478 with session_scope() as session:

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

480 session.add(

481 SignupFlow(

482 name="Completed",

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

484 flow_token=random_hex(32),

485 email_verified=True,

486 email_token=random_hex(32),

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

488 username="completed",

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

490 gender="Woman",

491 hosting_status=HostingStatus.cant_host,

492 city="Testing city",

493 geom=create_coordinate(40.7108, -73.9740),

494 geom_radius=100,

495 accepted_tos=TOS_VERSION,

496 opt_out_of_newsletter=False,

497 filled_motivations=True,

498 )

499 )

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

501 session.add(

502 SignupFlow(

503 name="Expired",

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

505 flow_token=random_hex(32),

506 email_token=random_hex(32),

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

508 )

509 )

510 # never got as far as being sent an email

511 session.add(

512 SignupFlow(

513 name="Fresh",

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

515 flow_token=random_hex(32),

516 )

517 )

518

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

520 if flow.name == "Completed":

521 flow.accepted_community_guidelines = GUIDELINES_VERSION

522

523

524@_populates(BackgroundJob)

525def _populate_background_jobs() -> None:

526 with session_scope() as session:

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

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

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

532## The tests

535def test_every_hybrid_is_covered() -> None:

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

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

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

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

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

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

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

543 )

544

545

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

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

548

549

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

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

552 POPULATIONS[model]()

553

554 with session_scope() as session:

555 mapper = inspect(model)

556 sql_values = _sql_values(session, model, name)

557

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

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

560 python_value = getattr(instance, name)

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

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

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

564 )

565

566

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

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

569 """

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

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

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

573 """

574 POPULATIONS[model]()

575

576 with session_scope() as session:

577 _sql_values(session, model, name)

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

579 assert instances

580 for instance in instances:

581 with pytest.raises(TypeError):

582 value = getattr(instance, name)

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

584 bool(value)

585

586

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

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

589 """

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

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

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

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

594 """

595 POPULATIONS[model]()

596

597 with session_scope() as session:

598 mapper = inspect(model)

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

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

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

602

603 seen = set()

604 for instance in instances:

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

606 for user in users:

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

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

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

610 sql_value = session.execute(

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

612 .select_from(model)

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

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

615 ).scalar_one()

616 assert _agree(python_value, sql_value), (

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

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

619 )

620 seen.add(bool(sql_value))

621

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

623

624

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

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

627 mapper = inspect(model)

628 values = {

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

630 }

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

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

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

634 )

635 return values

636

637

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

639 """

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

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

642 """

643 if sql_value is None:

644 return python_value is False

645 return bool(python_value == sql_value)

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

260 statements