mi
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import random
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app.domain import (
|
|
ContactCandidate,
|
|
WorkflowState,
|
|
assert_transition,
|
|
choose_newest,
|
|
full_jitter_delay,
|
|
select_email,
|
|
validate_phone,
|
|
)
|
|
from app.mapping import CitizenshipDictionary, UnknownCitizenship
|
|
|
|
|
|
def test_contact_choice_has_numeric_id_tie_breaker() -> None:
|
|
created = datetime(2026, 8, 6, tzinfo=UTC)
|
|
selected = choose_newest(
|
|
[
|
|
ContactCandidate("9", created, None),
|
|
ContactCandidate("10", created, None),
|
|
]
|
|
)
|
|
assert selected and selected.b24_id == "10"
|
|
|
|
|
|
def test_phone_email_and_citizenship_mapping() -> None:
|
|
assert validate_phone("+79001234567") == "+79001234567"
|
|
with pytest.raises(ValueError):
|
|
validate_phone("8 900 123-45-67")
|
|
assert (
|
|
select_email(
|
|
[
|
|
{"VALUE": "home@example.test", "VALUE_TYPE": "HOME"},
|
|
{"VALUE": "work@example.test", "VALUE_TYPE": "WORK"},
|
|
]
|
|
)
|
|
== "work@example.test"
|
|
)
|
|
dictionary = CitizenshipDictionary(ttl_seconds=60)
|
|
now = datetime(2026, 8, 6, tzinfo=UTC)
|
|
dictionary.load([{"ID": "7", "VALUE": "Казахстан"}], now)
|
|
assert dictionary.resolve("7", now + timedelta(seconds=30)) == "Казахстан"
|
|
with pytest.raises(UnknownCitizenship):
|
|
dictionary.resolve("8", now)
|
|
|
|
|
|
def test_state_machine_and_retry_bounds() -> None:
|
|
assert_transition(WorkflowState.CREATED, WorkflowState.RUNNING)
|
|
with pytest.raises(ValueError):
|
|
assert_transition(WorkflowState.SUCCEEDED, WorkflowState.RUNNING)
|
|
delay = full_jitter_delay(4, 1, 5, rng=random.Random(1))
|
|
assert 0 <= delay <= 5
|