692 lines
24 KiB
Python
692 lines
24 KiB
Python
# tests/repositories/test_member.py
|
||
import datetime as dt
|
||
from typing import List, Any
|
||
|
||
import pytest
|
||
|
||
from backend.models import Member as MemberModel, ScheduleStatus
|
||
from backend.repositories import MemberRepository
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Helper: a few sample members we can reuse across tests
|
||
# ----------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def sample_members() -> List[MemberModel]:
|
||
"""Return a list of MemberModel objects (not yet persisted)."""
|
||
return [
|
||
MemberModel(
|
||
MemberId=-1,
|
||
FirstName="Alice",
|
||
LastName="Anderson",
|
||
Email="alice@example.com",
|
||
PhoneNumber="555‑1111",
|
||
ClassificationId=1,
|
||
Notes=None,
|
||
IsActive=1,
|
||
LastScheduledAt=None,
|
||
LastAcceptedAt=None,
|
||
LastDeclinedAt=None,
|
||
DeclineStreak=0,
|
||
),
|
||
MemberModel(
|
||
MemberId=-1,
|
||
FirstName="Bob",
|
||
LastName="Baker",
|
||
Email="bob@example.com",
|
||
PhoneNumber="555‑2222",
|
||
ClassificationId=2,
|
||
Notes="VIP",
|
||
IsActive=1,
|
||
LastScheduledAt=dt.datetime(2025, 8, 20, 10, 0, 0),
|
||
LastAcceptedAt=dt.datetime(2025, 8, 19, 9, 30, 0),
|
||
LastDeclinedAt=None,
|
||
DeclineStreak=0,
|
||
),
|
||
MemberModel(
|
||
MemberId=-1,
|
||
FirstName="Carol",
|
||
LastName="Carter",
|
||
Email=None,
|
||
PhoneNumber=None,
|
||
ClassificationId=1,
|
||
Notes=None,
|
||
IsActive=0, # inactive – useful for filter tests
|
||
LastScheduledAt=None,
|
||
LastAcceptedAt=None,
|
||
LastDeclinedAt=None,
|
||
DeclineStreak=0,
|
||
),
|
||
]
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Fixture to wipe the Members table (used by tests that need a clean slate)
|
||
# ----------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def clean_members(member_repo: MemberRepository):
|
||
"""
|
||
Delete *all* rows from the Members table **and** any rows that
|
||
reference it (ServiceAvailability). The service‑availability tests
|
||
rely on the seeded Alice/Bob rows, so we only invoke this fixture
|
||
in the member‑repo tests that need isolation.
|
||
"""
|
||
# 1️⃣ Remove dependent rows first – otherwise the FK constraint blocks us.
|
||
member_repo.db.execute(
|
||
f"DELETE FROM ServiceAvailability"
|
||
) # commit happens inside `execute`
|
||
|
||
# 2️⃣ Now we can safely delete the members themselves.
|
||
member_repo.db.execute(
|
||
f"DELETE FROM {member_repo._TABLE}"
|
||
)
|
||
member_repo.db._conn.commit()
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Helper to build a MemberModel with explicit timestamps.
|
||
# ----------------------------------------------------------------------
|
||
def make_member(
|
||
repo: MemberRepository,
|
||
first_name: str,
|
||
last_name: str,
|
||
*,
|
||
classification_id: int = 1,
|
||
is_active: int = 1,
|
||
accepted_at: str | None = None,
|
||
scheduled_at: str | None = None,
|
||
declined_at: str | None = None,
|
||
decline_streak: int = 0,
|
||
) -> MemberModel:
|
||
"""Insert a member and then manually set the optional timestamp columns."""
|
||
m = repo.create(
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
email=None,
|
||
phone_number=None,
|
||
classification_id=classification_id,
|
||
notes=None,
|
||
is_active=is_active,
|
||
)
|
||
# Directly update the row so we can control the timestamps without
|
||
# invoking the repository’s higher‑level helpers (which would reset
|
||
# other fields).
|
||
sql = f"""
|
||
UPDATE {repo._TABLE}
|
||
SET
|
||
LastAcceptedAt = ?,
|
||
LastScheduledAt = ?,
|
||
LastDeclinedAt = ?,
|
||
DeclineStreak = ?
|
||
WHERE {repo._PK} = ?
|
||
"""
|
||
repo.db.execute(
|
||
sql,
|
||
(
|
||
accepted_at,
|
||
scheduled_at,
|
||
declined_at,
|
||
decline_streak,
|
||
m.MemberId,
|
||
),
|
||
)
|
||
# Refresh the model from the DB so the attributes reflect the changes.
|
||
return repo.get_by_id(m.MemberId) # type: ignore[return-value]
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 1️⃣ Basic CRUD – create & get_by_id
|
||
# ----------------------------------------------------------------------
|
||
def test_create_and_get_by_id(member_repo: MemberRepository):
|
||
member = member_repo.create(
|
||
first_name="Diana",
|
||
last_name="Doe",
|
||
email="diana@example.com",
|
||
phone_number="555‑3333",
|
||
classification_id=3,
|
||
notes="New recruit",
|
||
is_active=1,
|
||
)
|
||
|
||
# Primary key should be a positive integer (AUTOINCREMENT starts at 1)
|
||
assert isinstance(member.MemberId, int) and member.MemberId > 0
|
||
|
||
# Retrieve the same row
|
||
fetched = member_repo.get_by_id(member.MemberId)
|
||
assert fetched is not None
|
||
assert fetched.FirstName == "Diana"
|
||
assert fetched.LastName == "Doe"
|
||
assert fetched.Email == "diana@example.com"
|
||
assert fetched.ClassificationId == 3
|
||
assert fetched.IsActive == 1
|
||
assert fetched.Notes == "New recruit"
|
||
|
||
|
||
def test_get_by_id_returns_none_when_missing(member_repo: MemberRepository):
|
||
"""A PK that does not exist must return ``None`` (no exception)."""
|
||
assert member_repo.get_by_id(9999) is None
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 2️⃣ list_all – bulk insertion + retrieval
|
||
# ----------------------------------------------------------------------
|
||
def test_list_all(
|
||
member_repo: MemberRepository,
|
||
sample_members: List[MemberModel],
|
||
clean_members, # ensure we start from an empty table
|
||
):
|
||
for m in sample_members:
|
||
member_repo.create(
|
||
first_name=m.FirstName,
|
||
last_name=m.LastName,
|
||
email=m.Email,
|
||
phone_number=m.PhoneNumber,
|
||
classification_id=m.ClassificationId,
|
||
notes=m.Notes,
|
||
is_active=m.IsActive,
|
||
)
|
||
|
||
all_members = member_repo.list_all()
|
||
# Because we cleared the table first, we expect exactly the three we added.
|
||
assert len(all_members) == 3
|
||
|
||
# Spot‑check that each name appears
|
||
names = {(m.FirstName, m.LastName) for m in all_members}
|
||
assert ("Alice", "Anderson") in names
|
||
assert ("Bob", "Baker") in names
|
||
assert ("Carol", "Carter") in names
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 3️⃣ get_by_classification_ids – filter by classification list
|
||
# ----------------------------------------------------------------------
|
||
def test_get_by_classification_ids(
|
||
member_repo: MemberRepository,
|
||
sample_members: List[MemberModel],
|
||
clean_members,
|
||
):
|
||
for m in sample_members:
|
||
member_repo.create(
|
||
first_name=m.FirstName,
|
||
last_name=m.LastName,
|
||
email=m.Email,
|
||
phone_number=m.PhoneNumber,
|
||
classification_id=m.ClassificationId,
|
||
notes=m.Notes,
|
||
is_active=m.IsActive,
|
||
)
|
||
|
||
# Classification 1 → Alice + Carol (2 rows)
|
||
result = member_repo.get_by_classification_ids([1])
|
||
assert len(result) == 2
|
||
assert {r.FirstName for r in result} == {"Alice", "Carol"}
|
||
|
||
# Classification 2 → only Bob
|
||
result = member_repo.get_by_classification_ids([2])
|
||
assert len(result) == 1
|
||
assert result[0].FirstName == "Bob"
|
||
|
||
# Both classifications → all three
|
||
result = member_repo.get_by_classification_ids([1, 2])
|
||
assert len(result) == 3
|
||
|
||
|
||
def test_candidate_queue_obeys_boost_and_timestamp_sorting(
|
||
member_repo: MemberRepository,
|
||
):
|
||
"""
|
||
Verify that ``candidate_queue`` respects:
|
||
|
||
1️⃣ The boost clause (low ``DeclineStreak`` + recent ``LastDeclinedAt``).
|
||
2️⃣ ``LastAcceptedAt`` ASC (oldest first, ``NULL`` → far‑past).
|
||
3️⃣ ``LastScheduledAt`` ASC (same handling).
|
||
|
||
The default boost window is 2 days (172 800 seconds).
|
||
|
||
Additional rule (as stated in the doc‑string):
|
||
*Members whose ``LastAcceptedAt`` is NULL should appear **before** members
|
||
that have a non‑NULL acceptance date.*
|
||
"""
|
||
# --------------------------------------------------------------
|
||
# 0️⃣ Remove any ServiceAvailability rows that reference the seeded
|
||
# members, then delete the seeded members themselves.
|
||
# --------------------------------------------------------------
|
||
member_repo.db.execute("DELETE FROM ServiceAvailability")
|
||
member_repo.db.execute(
|
||
f"DELETE FROM {member_repo._TABLE} WHERE MemberId IN (1, 2)"
|
||
)
|
||
member_repo.db._conn.commit()
|
||
|
||
# --------------------------------------------------------------
|
||
# 1️⃣ Build a diverse set of members.
|
||
# --------------------------------------------------------------
|
||
# ── A – active, no timestamps (baseline, NULL acceptance)
|
||
a = make_member(member_repo, "Alice", "Anderson")
|
||
|
||
# ── B – active, accepted yesterday (non‑NULL acceptance)
|
||
yesterday = (dt.datetime.utcnow() - dt.timedelta(days=1)).strftime(
|
||
"%Y-%m-%d %H:%M:%S"
|
||
)
|
||
b = make_member(
|
||
member_repo,
|
||
"Bob",
|
||
"Baker",
|
||
accepted_at=yesterday,
|
||
)
|
||
|
||
# ── C – active, declined **today** with a low streak (boost candidate)
|
||
today_iso = dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
|
||
c = make_member(
|
||
member_repo,
|
||
"Carol",
|
||
"Clark",
|
||
declined_at=today_iso,
|
||
decline_streak=1, # < 2 → qualifies for boost
|
||
)
|
||
|
||
# ── D – active, declined **3 days ago** (outside the 2‑day boost window,
|
||
# still NULL acceptance)
|
||
three_days_ago = (dt.datetime.utcnow() - dt.timedelta(days=3)).strftime(
|
||
"%Y-%m-%d %H:%M:%S"
|
||
)
|
||
d = make_member(
|
||
member_repo,
|
||
"Dave",
|
||
"Davis",
|
||
declined_at=three_days_ago,
|
||
decline_streak=1,
|
||
)
|
||
|
||
# ── E – **inactive** member – should never appear when only_active=True.
|
||
e = make_member(
|
||
member_repo,
|
||
"Eve",
|
||
"Evans",
|
||
is_active=0,
|
||
)
|
||
|
||
# --------------------------------------------------------------
|
||
# 2️⃣ Pull the queue (default: only_active=True, boost_seconds=2 days)
|
||
# --------------------------------------------------------------
|
||
q = member_repo.candidate_queue(classification_ids=[1])
|
||
|
||
# --------------------------------------------------------------
|
||
# 3️⃣ Expected order (explain each step):
|
||
# --------------------------------------------------------------
|
||
# • Boosted members first → Carol (recent decline, streak < 2)
|
||
# • Then all members whose ``LastAcceptedAt`` is NULL,
|
||
# ordered by ``LastScheduledAt`` (both are NULL, so fallback to PK order):
|
||
# → Alice, then Dave
|
||
# • Finally members with a non‑NULL acceptance date → Bob
|
||
# • Eve is inactive → omitted.
|
||
expected_first_names = ["Carol", "Alice", "Dave", "Bob"]
|
||
assert [m.FirstName for m in q] == expected_first_names
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 5️⃣ touch_last_scheduled – updates the timestamp column
|
||
# ----------------------------------------------------------------------
|
||
def test_touch_last_scheduled_updates_timestamp(member_repo: MemberRepository):
|
||
member = member_repo.create(
|
||
first_name="Eve",
|
||
last_name="Evans",
|
||
email=None,
|
||
phone_number=None,
|
||
classification_id=4,
|
||
notes=None,
|
||
is_active=1,
|
||
)
|
||
assert member.LastScheduledAt is None
|
||
|
||
# Call the helper – it should set LastScheduledAt to the current UTC time.
|
||
member_repo.touch_last_scheduled(member.MemberId)
|
||
|
||
refreshed = member_repo.get_by_id(member.MemberId)
|
||
assert refreshed is not None
|
||
assert refreshed.LastScheduledAt is not None
|
||
|
||
# SQLite stores timestamps as ISO‑8601 strings; parsing should succeed.
|
||
dt.datetime.fromisoformat(refreshed.LastScheduledAt)
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 6️⃣ set_last_declined – records decline date and increments streak
|
||
# ----------------------------------------------------------------------
|
||
def test_set_last_declined_resets_streak_and_records_date(member_repo: MemberRepository):
|
||
member = member_repo.create(
|
||
first_name="Frank",
|
||
last_name="Foster",
|
||
email=None,
|
||
phone_number=None,
|
||
classification_id=4,
|
||
notes=None,
|
||
is_active=1,
|
||
)
|
||
# Initial state
|
||
assert member.DeclineStreak == 0
|
||
assert member.LastDeclinedAt is None
|
||
|
||
# Simulate a decline today.
|
||
today_iso = dt.date.today().isoformat()
|
||
member_repo.set_last_declined(member.MemberId, today_iso)
|
||
|
||
refreshed = member_repo.get_by_id(member.MemberId)
|
||
assert refreshed.DeclineStreak == 1
|
||
assert refreshed.LastDeclinedAt == today_iso
|
||
|
||
# Simulate a second decline tomorrow – streak should increase again.
|
||
tomorrow_iso = (dt.date.today() + dt.timedelta(days=1)).isoformat()
|
||
member_repo.set_last_declined(member.MemberId, tomorrow_iso)
|
||
|
||
refreshed2 = member_repo.get_by_id(member.MemberId)
|
||
assert refreshed2.DeclineStreak == 2
|
||
assert refreshed2.LastDeclinedAt == tomorrow_iso
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 7️⃣ get_active – filter active members only
|
||
# ----------------------------------------------------------------------
|
||
def test_get_active_filters_correctly(member_repo: MemberRepository, clean_members):
|
||
"""Test that get_active returns only active members."""
|
||
# Create active member
|
||
active_member = member_repo.create(
|
||
first_name="Active",
|
||
last_name="User",
|
||
email="active@example.com",
|
||
phone_number="555-1234",
|
||
classification_id=1,
|
||
is_active=1,
|
||
)
|
||
|
||
# Create inactive member
|
||
inactive_member = member_repo.create(
|
||
first_name="Inactive",
|
||
last_name="User",
|
||
email="inactive@example.com",
|
||
phone_number="555-5678",
|
||
classification_id=1,
|
||
is_active=0,
|
||
)
|
||
|
||
active_members = member_repo.get_active()
|
||
|
||
# Should only return the active member
|
||
assert len(active_members) == 1
|
||
assert active_members[0].MemberId == active_member.MemberId
|
||
assert active_members[0].FirstName == "Active"
|
||
assert active_members[0].IsActive == 1
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 8️⃣ set_last_accepted – resets decline data and sets acceptance date
|
||
# ----------------------------------------------------------------------
|
||
def test_set_last_accepted_resets_decline_data(member_repo: MemberRepository):
|
||
"""Test that set_last_accepted clears decline data and sets acceptance timestamp."""
|
||
member = member_repo.create(
|
||
first_name="Test",
|
||
last_name="Member",
|
||
email="test@example.com",
|
||
phone_number=None,
|
||
classification_id=1,
|
||
is_active=1,
|
||
)
|
||
|
||
# First decline the member to set up decline data
|
||
yesterday_iso = (dt.date.today() - dt.timedelta(days=1)).isoformat()
|
||
member_repo.set_last_declined(member.MemberId, yesterday_iso)
|
||
|
||
# Verify decline data is set
|
||
declined_member = member_repo.get_by_id(member.MemberId)
|
||
assert declined_member.DeclineStreak == 1
|
||
assert declined_member.LastDeclinedAt == yesterday_iso
|
||
assert declined_member.LastAcceptedAt is None
|
||
|
||
# Now accept
|
||
member_repo.set_last_accepted(member.MemberId)
|
||
|
||
# Verify acceptance resets decline data
|
||
accepted_member = member_repo.get_by_id(member.MemberId)
|
||
assert accepted_member.DeclineStreak == 0
|
||
assert accepted_member.LastDeclinedAt is None
|
||
assert accepted_member.LastAcceptedAt is not None
|
||
|
||
# Verify timestamp is recent (within last 5 seconds)
|
||
accepted_time = dt.datetime.fromisoformat(accepted_member.LastAcceptedAt.replace('f', '000'))
|
||
time_diff = dt.datetime.utcnow() - accepted_time
|
||
assert time_diff.total_seconds() < 5
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 9️⃣ reset_to_queue_front – moves member to front of scheduling queue
|
||
# ----------------------------------------------------------------------
|
||
def test_reset_to_queue_front_moves_member_to_front(member_repo: MemberRepository, clean_members):
|
||
"""Test that reset_to_queue_front properly resets timestamps to move member to front."""
|
||
# Create two members with different timestamps
|
||
older_member = make_member(
|
||
member_repo,
|
||
"Older",
|
||
"Member",
|
||
accepted_at="2025-01-01 10:00:00",
|
||
scheduled_at="2025-01-01 10:00:00",
|
||
declined_at="2025-01-01 10:00:00",
|
||
decline_streak=2,
|
||
)
|
||
|
||
newer_member = make_member(
|
||
member_repo,
|
||
"Newer",
|
||
"Member",
|
||
accepted_at="2025-08-01 10:00:00",
|
||
scheduled_at="2025-08-01 10:00:00",
|
||
)
|
||
|
||
# Verify initial queue order (newer should come first due to older accepted date)
|
||
initial_queue = member_repo.candidate_queue([1])
|
||
assert len(initial_queue) == 2
|
||
assert initial_queue[0].FirstName == "Older" # older accepted date comes first
|
||
assert initial_queue[1].FirstName == "Newer"
|
||
|
||
# Reset newer member to queue front
|
||
member_repo.reset_to_queue_front(newer_member.MemberId)
|
||
|
||
# Verify newer member is now at front
|
||
updated_queue = member_repo.candidate_queue([1])
|
||
assert len(updated_queue) == 2
|
||
assert updated_queue[0].FirstName == "Newer" # should now be first
|
||
assert updated_queue[1].FirstName == "Older"
|
||
|
||
# Verify the reset member has expected timestamp values
|
||
reset_member = member_repo.get_by_id(newer_member.MemberId)
|
||
assert reset_member.LastAcceptedAt == '1970-01-01 00:00:00'
|
||
assert reset_member.LastScheduledAt == '1970-01-01 00:00:00'
|
||
assert reset_member.LastDeclinedAt is None
|
||
assert reset_member.DeclineStreak == 0
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 🔟 Edge cases and error conditions
|
||
# ----------------------------------------------------------------------
|
||
def test_create_with_minimal_data(member_repo: MemberRepository):
|
||
"""Test creating a member with only required fields."""
|
||
member = member_repo.create(
|
||
first_name="Min",
|
||
last_name="Member"
|
||
)
|
||
|
||
assert member.FirstName == "Min"
|
||
assert member.LastName == "Member"
|
||
assert member.Email is None
|
||
assert member.PhoneNumber is None
|
||
assert member.ClassificationId is None
|
||
assert member.Notes is None
|
||
assert member.IsActive == 1 # default value
|
||
assert member.DeclineStreak == 0 # default value
|
||
|
||
|
||
def test_get_by_classification_ids_empty_list(member_repo: MemberRepository):
|
||
"""Test that empty classification list returns empty result without DB query."""
|
||
result = member_repo.get_by_classification_ids([])
|
||
assert result == []
|
||
|
||
|
||
def test_get_by_classification_ids_nonexistent_classification(member_repo: MemberRepository):
|
||
"""Test querying for nonexistent classification IDs."""
|
||
result = member_repo.get_by_classification_ids([999, 1000])
|
||
assert result == []
|
||
|
||
|
||
def test_candidate_queue_with_inactive_members(member_repo: MemberRepository, clean_members):
|
||
"""Test that candidate_queue respects only_active parameter."""
|
||
# Create active and inactive members
|
||
active = make_member(member_repo, "Active", "Member", is_active=1)
|
||
inactive = make_member(member_repo, "Inactive", "Member", is_active=0)
|
||
|
||
# Test with only_active=True (default)
|
||
queue_active_only = member_repo.candidate_queue([1], only_active=True)
|
||
assert len(queue_active_only) == 1
|
||
assert queue_active_only[0].FirstName == "Active"
|
||
|
||
# Test with only_active=False
|
||
queue_all = member_repo.candidate_queue([1], only_active=False)
|
||
assert len(queue_all) == 2
|
||
names = {m.FirstName for m in queue_all}
|
||
assert names == {"Active", "Inactive"}
|
||
|
||
|
||
def test_candidate_queue_boost_window_edge_cases(member_repo: MemberRepository, clean_members):
|
||
"""Test boost logic with edge cases around the boost window."""
|
||
now = dt.datetime.utcnow()
|
||
|
||
# Member declined well outside boost window (3 days ago)
|
||
outside_window = (now - dt.timedelta(days=3)).strftime("%Y-%m-%d %H:%M:%S")
|
||
outside_member = make_member(
|
||
member_repo,
|
||
"Outside",
|
||
"Member",
|
||
declined_at=outside_window,
|
||
decline_streak=1,
|
||
)
|
||
|
||
# Member declined well inside boost window (6 hours ago)
|
||
well_inside = (now - dt.timedelta(hours=6)).strftime("%Y-%m-%d %H:%M:%S")
|
||
inside_member = make_member(
|
||
member_repo,
|
||
"Inside",
|
||
"Member",
|
||
declined_at=well_inside,
|
||
decline_streak=1,
|
||
)
|
||
|
||
# Member with high decline streak (should not get boost)
|
||
high_streak_member = make_member(
|
||
member_repo,
|
||
"HighStreak",
|
||
"Member",
|
||
declined_at=well_inside,
|
||
decline_streak=5, # >= 2, so no boost
|
||
)
|
||
|
||
queue = member_repo.candidate_queue([1])
|
||
|
||
# Inside member should be boosted to front
|
||
first_names = [m.FirstName for m in queue]
|
||
assert "Inside" == first_names[0] # should be boosted
|
||
assert "HighStreak" in first_names[1:] # should not be boosted
|
||
assert "Outside" in first_names[1:] # should not be boosted
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"decline_streak,should_boost",
|
||
[
|
||
(0, True), # streak < 2
|
||
(1, True), # streak < 2
|
||
(2, False), # streak >= 2
|
||
(5, False), # streak >= 2
|
||
]
|
||
)
|
||
def test_candidate_queue_decline_streak_boost_logic(
|
||
member_repo: MemberRepository,
|
||
clean_members,
|
||
decline_streak: int,
|
||
should_boost: bool
|
||
):
|
||
"""Test boost logic for different decline streak values."""
|
||
now = dt.datetime.utcnow()
|
||
recent_decline = (now - dt.timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# Create a baseline member (no recent declines)
|
||
baseline = make_member(member_repo, "Baseline", "Member")
|
||
|
||
# Create test member with specific decline streak
|
||
test_member = make_member(
|
||
member_repo,
|
||
"Test",
|
||
"Member",
|
||
declined_at=recent_decline,
|
||
decline_streak=decline_streak,
|
||
)
|
||
|
||
queue = member_repo.candidate_queue([1])
|
||
first_names = [m.FirstName for m in queue]
|
||
|
||
if should_boost:
|
||
assert first_names[0] == "Test" # boosted to front
|
||
assert first_names[1] == "Baseline"
|
||
else:
|
||
# Order depends on other factors, but Test should not be boosted
|
||
# Both have NULL LastAcceptedAt, so order by LastScheduledAt (both NULL)
|
||
# then likely by primary key order
|
||
assert "Test" in first_names
|
||
assert "Baseline" in first_names
|
||
|
||
|
||
def test_touch_last_scheduled_with_nonexistent_member(member_repo: MemberRepository):
|
||
"""Test touch_last_scheduled with nonexistent member ID (should not raise error)."""
|
||
# This should not raise an exception, just silently do nothing
|
||
member_repo.touch_last_scheduled(99999)
|
||
|
||
|
||
def test_set_operations_with_nonexistent_member(member_repo: MemberRepository):
|
||
"""Test set operations with nonexistent member IDs."""
|
||
# These should not raise exceptions
|
||
member_repo.set_last_accepted(99999)
|
||
member_repo.set_last_declined(99999, "2025-08-29")
|
||
member_repo.reset_to_queue_front(99999)
|
||
|
||
|
||
def test_member_data_integrity_after_operations(member_repo: MemberRepository):
|
||
"""Test that member data remains consistent after various operations."""
|
||
member = member_repo.create(
|
||
first_name="Integrity",
|
||
last_name="Test",
|
||
email="integrity@example.com",
|
||
phone_number="555-0000",
|
||
classification_id=2,
|
||
notes="Test member",
|
||
is_active=1,
|
||
)
|
||
|
||
original_id = member.MemberId
|
||
original_email = member.Email
|
||
original_classification = member.ClassificationId
|
||
|
||
# Perform various timestamp operations
|
||
member_repo.touch_last_scheduled(member.MemberId)
|
||
member_repo.set_last_declined(member.MemberId, "2025-08-29")
|
||
member_repo.set_last_accepted(member.MemberId)
|
||
member_repo.reset_to_queue_front(member.MemberId)
|
||
|
||
# Verify core data is unchanged
|
||
final_member = member_repo.get_by_id(member.MemberId)
|
||
assert final_member.MemberId == original_id
|
||
assert final_member.FirstName == "Integrity"
|
||
assert final_member.LastName == "Test"
|
||
assert final_member.Email == original_email
|
||
assert final_member.ClassificationId == original_classification
|
||
assert final_member.IsActive == 1
|
||
|
||
# Verify reset operation results
|
||
assert final_member.LastAcceptedAt == '1970-01-01 00:00:00'
|
||
assert final_member.LastScheduledAt == '1970-01-01 00:00:00'
|
||
assert final_member.LastDeclinedAt is None
|
||
assert final_member.DeclineStreak == 0 |