cal/paper-dynasty-database
1
0
Fork 0
Code Issues 7 Pull Requests 4 Actions Packages Projects Releases Wiki Activity
3e84a06b23
paper-dynasty-database/tests/test_refractor_boost_integration.py
Cal Corum 7f17c9b9f2 fix: address PR #177 review — move import os to top-level, add audit idempotency guard 
- Move `import os` from inside evaluate_game() to module top-level imports
  (lazy imports are only for circular dependency avoidance)
- Add get_or_none idempotency guard before RefractorBoostAudit.create()
  inside db.atomic() to prevent IntegrityError on UNIQUE(card_state, tier)
  constraint in PostgreSQL when apply_tier_boost is called twice for the
  same tier
- Update atomicity test stub to provide card_state/tier attributes for
  the new Peewee expression in the idempotency guard

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-30 13:16:27 -05:00

1318 lines
49 KiB
Python
Raw Blame History

"""Integration tests for apply_tier_boost() orchestration.
Tests the full flow: create base card + ratings -> apply boost -> verify
variant card created with correct ratings, audit record written, card state
updated, and Card instances propagated.
Uses a named shared-memory SQLite database (same pattern as
test_postgame_refractor.py) so that db.atomic() inside apply_tier_boost()
and test assertions operate on the same underlying connection. The
refractor_boost module's 'db' reference is patched to point at this
shared-memory database.
The conftest autouse setup_test_db fixture is overridden by the module-level
setup_boost_int_db fixture (autouse=True) so each test gets a fresh schema.
BattingCard, BattingCardRatings, PitchingCard, PitchingCardRatings, and
RefractorBoostAudit are included in the model list so apply_tier_boost() can
create rows during tests.
All pitcher sum assertions use the full 108-sum (18 variable + 9 x-check
columns), not just the 79 variable-column subset, because that is the
card-level invariant apply_tier_boost() must preserve.
"""
import os
# Must set before any app imports so SKIP_TABLE_CREATION is True in db_engine.
os.environ["DATABASE_TYPE"] = "postgresql"
os.environ.setdefault("POSTGRES_PASSWORD", "test-dummy")
os.environ.setdefault("API_TOKEN", "test-token")
import app.services.refractor_boost as _refractor_boost_module
import pytest
from peewee import SqliteDatabase
from app.db_engine import (
BattingCard,
BattingCardRatings,
BattingSeasonStats,
Card,
Cardset,
Decision,
Event,
MlbPlayer,
Pack,
PackType,
PitchingCard,
PitchingCardRatings,
PitchingSeasonStats,
Player,
ProcessedGame,
Rarity,
RefractorBoostAudit,
RefractorCardState,
RefractorCosmetic,
RefractorTierBoost,
RefractorTrack,
Roster,
RosterSlot,
ScoutClaim,
ScoutOpportunity,
StratGame,
StratPlay,
Team,
)
from app.services.refractor_boost import (
BATTER_OUTCOME_COLUMNS,
PITCHER_OUTCOME_COLUMNS,
PITCHER_XCHECK_COLUMNS,
apply_tier_boost,
compute_variant_hash,
)
from app.services.refractor_evaluator import evaluate_card
# ---------------------------------------------------------------------------
# Named shared-memory SQLite for integration tests.
# A shared-cache URI allows all threads (e.g. TestClient routes and pytest
# fixtures) to share the same in-memory DB. Required because SQLite
# :memory: is per-connection.
# ---------------------------------------------------------------------------
_boost_int_db = SqliteDatabase(
"file:boost_int_test?mode=memory&cache=shared",
uri=True,
pragmas={"foreign_keys": 1},
)
# All models in dependency order (parents before children).
_BOOST_INT_MODELS = [
Rarity,
Event,
Cardset,
MlbPlayer,
Player,
Team,
PackType,
Pack,
Card,
Roster,
RosterSlot,
StratGame,
StratPlay,
Decision,
ScoutOpportunity,
ScoutClaim,
BattingSeasonStats,
PitchingSeasonStats,
ProcessedGame,
BattingCard,
BattingCardRatings,
PitchingCard,
PitchingCardRatings,
RefractorTrack,
RefractorCardState,
RefractorTierBoost,
RefractorCosmetic,
RefractorBoostAudit,
]
# Patch the service-layer 'db' reference so that db.atomic() inside
# apply_tier_boost() operates on the shared-memory SQLite connection.
_refractor_boost_module.db = _boost_int_db
# ---------------------------------------------------------------------------
# Database fixture — binds all models to the shared-memory SQLite db.
# autouse=True so every test automatically gets a fresh schema.
# This overrides the conftest autouse setup_test_db fixture for this module.
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def setup_boost_int_db():
"""Bind integration test models to the shared-memory SQLite db.
Creates tables before each test and drops them in reverse order after.
The autouse=True ensures every test in this module gets an isolated schema
without explicitly requesting the fixture.
"""
_boost_int_db.bind(_BOOST_INT_MODELS)
_boost_int_db.connect(reuse_if_open=True)
_boost_int_db.create_tables(_BOOST_INT_MODELS)
yield _boost_int_db
_boost_int_db.drop_tables(list(reversed(_BOOST_INT_MODELS)), safe=True)
# ---------------------------------------------------------------------------
# Shared factory helpers
# ---------------------------------------------------------------------------
def _make_rarity():
r, _ = Rarity.get_or_create(value=1, name="Common", defaults={"color": "#ffffff"})
return r
def _make_player(name="Test Player", pos="1B"):
cs, _ = Cardset.get_or_create(
name="BI Test Set",
defaults={"description": "boost int test", "total_cards": 100},
)
return Player.create(
p_name=name,
rarity=_make_rarity(),
cardset=cs,
set_num=1,
pos_1=pos,
image="https://example.com/img.png",
mlbclub="TST",
franchise="TST",
description=f"boost int test: {name}",
)
def _make_team(abbrev="TST", gmid=99001):
return Team.create(
abbrev=abbrev,
sname=abbrev,
lname=f"Team {abbrev}",
gmid=gmid,
gmname=f"gm_{abbrev.lower()}",
gsheet="https://docs.google.com/spreadsheets/bi",
wallet=500,
team_value=1000,
collection_value=1000,
season=11,
is_ai=False,
)
def _make_track(name="Batter Track", card_type="batter"):
track, _ = RefractorTrack.get_or_create(
name=name,
defaults=dict(
card_type=card_type,
formula="pa + tb * 2",
t1_threshold=37,
t2_threshold=149,
t3_threshold=448,
t4_threshold=896,
),
)
return track
def _make_state(player, team, track, current_tier=0, current_value=0.0):
return RefractorCardState.create(
player=player,
team=team,
track=track,
current_tier=current_tier,
current_value=current_value,
fully_evolved=False,
last_evaluated_at=None,
)
# Representative batter ratings that sum to exactly 108.
_BASE_BATTER_RATINGS = {
"homerun": 3.0,
"bp_homerun": 1.0,
"triple": 0.5,
"double_three": 2.0,
"double_two": 2.0,
"double_pull": 6.0,
"single_two": 4.0,
"single_one": 12.0,
"single_center": 5.0,
"bp_single": 2.0,
"hbp": 3.0,
"walk": 7.0,
"strikeout": 15.0,
"lineout": 3.0,
"popout": 2.0,
"flyout_a": 5.0,
"flyout_bq": 4.0,
"flyout_lf_b": 3.0,
"flyout_rf_b": 9.0,
"groundout_a": 6.0,
"groundout_b": 8.0,
"groundout_c": 5.5,
}
# Representative pitcher ratings: 18 variable columns sum to 79,
# 9 x-check columns sum to 29, full card sums to 108.
_BASE_PITCHER_RATINGS = {
# Variable columns (sum=79)
"homerun": 3.3,
"bp_homerun": 2.0,
"triple": 0.75,
"double_three": 0.0,
"double_two": 0.0,
"double_cf": 2.95,
"single_two": 5.7,
"single_one": 0.0,
"single_center": 5.0,
"bp_single": 5.0,
"hbp": 3.0,
"walk": 5.0,
"strikeout": 10.0,
"flyout_lf_b": 15.1,
"flyout_cf_b": 0.9,
"flyout_rf_b": 0.0,
"groundout_a": 15.1,
"groundout_b": 5.2,
# X-check columns (sum=29)
"xcheck_p": 1.0,
"xcheck_c": 3.0,
"xcheck_1b": 2.0,
"xcheck_2b": 6.0,
"xcheck_3b": 3.0,
"xcheck_ss": 7.0,
"xcheck_lf": 2.0,
"xcheck_cf": 3.0,
"xcheck_rf": 2.0,
}
def _create_base_batter(player, variant=0):
"""Create a BattingCard with variant=0 and two ratings rows (vL, vR)."""
card = BattingCard.create(
player=player,
variant=variant,
steal_low=1,
steal_high=6,
steal_auto=False,
steal_jump=0.5,
bunting="C",
hit_and_run="B",
running=3,
offense_col=2,
hand="R",
)
for vs_hand in ("L", "R"):
BattingCardRatings.create(
battingcard=card,
vs_hand=vs_hand,
pull_rate=0.4,
center_rate=0.35,
slap_rate=0.25,
avg=0.300,
obp=0.370,
slg=0.450,
**_BASE_BATTER_RATINGS,
)
return card
def _create_base_pitcher(player, variant=0):
"""Create a PitchingCard with variant=0 and two ratings rows (vL, vR)."""
card = PitchingCard.create(
player=player,
variant=variant,
balk=1,
wild_pitch=2,
hold=3,
starter_rating=60,
relief_rating=50,
closer_rating=None,
batting=None,
offense_col=1,
hand="R",
)
for vs_hand in ("L", "R"):
PitchingCardRatings.create(
pitchingcard=card,
vs_hand=vs_hand,
avg=0.250,
obp=0.310,
slg=0.380,
**_BASE_PITCHER_RATINGS,
)
return card
def _injectable_kwargs(model_map: dict) -> dict:
"""Build apply_tier_boost() injectable kwargs from a name->model mapping."""
return {
"_batting_card_model": model_map.get("BattingCard", BattingCard),
"_batting_ratings_model": model_map.get(
"BattingCardRatings", BattingCardRatings
),
"_pitching_card_model": model_map.get("PitchingCard", PitchingCard),
"_pitching_ratings_model": model_map.get(
"PitchingCardRatings", PitchingCardRatings
),
"_card_model": model_map.get("Card", Card),
"_state_model": model_map.get("RefractorCardState", RefractorCardState),
"_audit_model": model_map.get("RefractorBoostAudit", RefractorBoostAudit),
}
# Default injectable kwargs point at the real models (which are now bound to
# the shared-memory DB via the fixture).
_DEFAULT_KWARGS = _injectable_kwargs({})
# ---------------------------------------------------------------------------
# TestBatterTierUpFlow
# ---------------------------------------------------------------------------
class TestBatterTierUpFlow:
"""Full batter T1 boost flow: card + ratings created, state updated, base unchanged."""
def test_creates_variant_card(self):
"""T1 boost creates a new BattingCard row with the correct variant hash.
What: Set up a player with a base BattingCard (variant=0) and a
RefractorCardState. Call apply_tier_boost for T1. Assert that a new
BattingCard row exists with the expected variant hash.
Why: The variant card is the persistent record of the boosted card.
If it is not created, the tier-up has no lasting effect on the card's
identity in the database.
"""
player = _make_player()
team = _make_team()
track = _make_track()
_make_state(player, team, track)
_create_base_batter(player)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
variant_card = BattingCard.get_or_none(
(BattingCard.player == player) & (BattingCard.variant == expected_variant)
)
assert variant_card is not None, (
f"Expected BattingCard with variant={expected_variant} to be created"
)
def test_creates_boosted_ratings(self):
"""T1 boost creates BattingCardRatings rows with positive deltas applied.
What: After T1 boost, the variant card's vR ratings must have
homerun > base_homerun (the primary positive delta column is homerun
+0.5 per tier, funded by strikeout -1.5 and groundout_a -0.5).
Why: If the ratings rows are not created, or if the boost formula is
not applied, the variant card has the same outcomes as the base card
and offers no gameplay advantage.
"""
player = _make_player()
team = _make_team()
track = _make_track()
_make_state(player, team, track)
_create_base_batter(player)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
new_variant = compute_variant_hash(player.player_id, 1)
new_card = BattingCard.get(
(BattingCard.player == player) & (BattingCard.variant == new_variant)
)
vr_ratings = BattingCardRatings.get(
(BattingCardRatings.battingcard == new_card)
& (BattingCardRatings.vs_hand == "R")
)
# homerun must be higher than base (base + 0.5 delta)
assert vr_ratings.homerun > _BASE_BATTER_RATINGS["homerun"], (
f"Expected homerun > {_BASE_BATTER_RATINGS['homerun']}, "
f"got {vr_ratings.homerun}"
)
# strikeout must be lower (reduced by 1.5)
assert vr_ratings.strikeout < _BASE_BATTER_RATINGS["strikeout"], (
f"Expected strikeout < {_BASE_BATTER_RATINGS['strikeout']}, "
f"got {vr_ratings.strikeout}"
)
def test_ratings_sum_108(self):
"""Boosted batter ratings rows sum to exactly 108.
What: After T1 boost, sum all 22 BATTER_OUTCOME_COLUMNS for both the
vL and vR ratings rows on the variant card. Each must be 108.0.
Why: The 108-sum is the card-level invariant. Peewee bypasses Pydantic
validators, so apply_tier_boost() must explicitly assert and preserve
this invariant. A sum other than 108 would corrupt game simulation.
"""
player = _make_player()
team = _make_team()
track = _make_track()
_make_state(player, team, track)
_create_base_batter(player)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
new_variant = compute_variant_hash(player.player_id, 1)
new_card = BattingCard.get(
(BattingCard.player == player) & (BattingCard.variant == new_variant)
)
for ratings_row in BattingCardRatings.select().where(
BattingCardRatings.battingcard == new_card
):
total = sum(getattr(ratings_row, col) for col in BATTER_OUTCOME_COLUMNS)
assert abs(total - 108.0) < 0.01, (
f"Batter 108-sum violated for vs_hand={ratings_row.vs_hand}: "
f"sum={total:.6f}"
)
def test_audit_record_created(self):
"""RefractorBoostAudit row is created with correct tier and variant.
What: After T1 boost, a RefractorBoostAudit row must exist for the
card state with tier=1 and variant_created matching the expected hash.
Why: The audit record is the permanent log of when and how each tier
boost was applied. Without it, there is no way to debug tier-up
regressions or replay boost history.
"""
player = _make_player()
team = _make_team()
track = _make_track()
state = _make_state(player, team, track)
_create_base_batter(player)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
audit = RefractorBoostAudit.get_or_none(
(RefractorBoostAudit.card_state == state) & (RefractorBoostAudit.tier == 1)
)
assert audit is not None, "Expected RefractorBoostAudit row to be created"
assert audit.variant_created == expected_variant
def test_card_state_variant_updated(self):
"""RefractorCardState.variant and current_tier are updated after boost.
What: After T1 boost, RefractorCardState.variant must equal the new
variant hash and current_tier must be 1.
Why: apply_tier_boost() is the sole writer of current_tier on tier-up.
If either field is not updated, the card state is inconsistent with
the variant card that was created.
"""
player = _make_player()
team = _make_team()
track = _make_track()
state = _make_state(player, team, track)
_create_base_batter(player)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier == 1
assert state.variant == expected_variant
def test_base_card_unchanged(self):
"""variant=0 BattingCard and BattingCardRatings are not modified.
What: After T1 boost, the base card (variant=0) and its ratings rows
must be byte-for-byte identical to what they were before the boost.
Why: The base card is the source of truth for all variant calculations.
If it is modified, subsequent tier-ups will compute incorrect boosts and
the original card identity is lost.
"""
player = _make_player()
team = _make_team()
track = _make_track()
_make_state(player, team, track)
base_card = _create_base_batter(player)
# Capture base card homerun before boost
base_vr = BattingCardRatings.get(
(BattingCardRatings.battingcard == base_card)
& (BattingCardRatings.vs_hand == "R")
)
base_homerun_before = base_vr.homerun
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
# Re-fetch the base card ratings and verify unchanged
base_vr_after = BattingCardRatings.get(
(BattingCardRatings.battingcard == base_card)
& (BattingCardRatings.vs_hand == "R")
)
assert base_vr_after.homerun == base_homerun_before, (
f"Base card homerun was modified: "
f"{base_homerun_before} -> {base_vr_after.homerun}"
)
# Verify there is still only one base card
base_cards = list(
BattingCard.select().where(
(BattingCard.player == player) & (BattingCard.variant == 0)
)
)
assert len(base_cards) == 1, f"Expected 1 base card, found {len(base_cards)}"
# ---------------------------------------------------------------------------
# TestPitcherTierUpFlow
# ---------------------------------------------------------------------------
class TestPitcherTierUpFlow:
"""Full pitcher T1 boost flow: x-check columns unchanged, 108-sum preserved."""
def test_creates_variant_card(self):
"""T1 boost creates a PitchingCard with the correct variant hash.
What: Set up a player with a base PitchingCard (variant=0) and a
RefractorCardState with card_type='sp'. Call apply_tier_boost for T1.
Assert that a new PitchingCard exists with the expected variant hash.
Why: Pitcher cards follow a different boost algorithm (TB budget
priority) but must still produce a variant card row.
"""
player = _make_player(pos="SP")
team = _make_team(abbrev="PT1", gmid=99010)
track = _make_track(name="SP Track", card_type="sp")
_make_state(player, team, track)
_create_base_pitcher(player)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "sp", **_DEFAULT_KWARGS)
variant_card = PitchingCard.get_or_none(
(PitchingCard.player == player) & (PitchingCard.variant == expected_variant)
)
assert variant_card is not None
def test_xcheck_columns_unchanged(self):
"""X-check columns on boosted pitcher ratings are identical to the base card.
What: After T1 boost, every xcheck_* column on the variant card's
ratings rows must exactly match the corresponding value on the base card.
Why: X-check columns encode defensive routing weights. The pitcher
boost algorithm is explicitly designed to never touch them. If any
x-check column is modified, game simulation defensive logic breaks and
the 108-sum invariant (variable 79 + x-check 29) is violated.
"""
player = _make_player(pos="SP")
team = _make_team(abbrev="PT2", gmid=99011)
track = _make_track(name="SP Track2", card_type="sp")
_make_state(player, team, track)
base_card = _create_base_pitcher(player)
apply_tier_boost(player.player_id, team.id, 1, "sp", **_DEFAULT_KWARGS)
new_variant = compute_variant_hash(player.player_id, 1)
new_card = PitchingCard.get(
(PitchingCard.player == player) & (PitchingCard.variant == new_variant)
)
for vs_hand in ("L", "R"):
base_row = PitchingCardRatings.get(
(PitchingCardRatings.pitchingcard == base_card)
& (PitchingCardRatings.vs_hand == vs_hand)
)
new_row = PitchingCardRatings.get(
(PitchingCardRatings.pitchingcard == new_card)
& (PitchingCardRatings.vs_hand == vs_hand)
)
for col in PITCHER_XCHECK_COLUMNS:
assert getattr(new_row, col) == pytest.approx(
getattr(base_row, col), abs=1e-9
), (
f"X-check column '{col}' was modified for vs_hand={vs_hand}: "
f"{getattr(base_row, col)} -> {getattr(new_row, col)}"
)
def test_108_sum_pitcher(self):
"""Boosted pitcher ratings sum to exactly 108 (variable 79 + x-check 29).
What: After T1 boost, sum all 18 PITCHER_OUTCOME_COLUMNS and all 9
PITCHER_XCHECK_COLUMNS for each vs_hand split on the variant card.
The combined total must be 108.0 for each split.
Why: The card-level invariant is the full 108 total — not just the 79
variable-column subset. If the x-check columns are not preserved or
the variable columns drift, the invariant is broken.
"""
player = _make_player(pos="SP")
team = _make_team(abbrev="PT3", gmid=99012)
track = _make_track(name="SP Track3", card_type="sp")
_make_state(player, team, track)
_create_base_pitcher(player)
apply_tier_boost(player.player_id, team.id, 1, "sp", **_DEFAULT_KWARGS)
new_variant = compute_variant_hash(player.player_id, 1)
new_card = PitchingCard.get(
(PitchingCard.player == player) & (PitchingCard.variant == new_variant)
)
for row in PitchingCardRatings.select().where(
PitchingCardRatings.pitchingcard == new_card
):
var_sum = sum(getattr(row, col) for col in PITCHER_OUTCOME_COLUMNS)
xcheck_sum = sum(getattr(row, col) for col in PITCHER_XCHECK_COLUMNS)
total = var_sum + xcheck_sum
assert abs(total - 108.0) < 0.01, (
f"Pitcher 108-sum (variable + x-check) violated for "
f"vs_hand={row.vs_hand}: var={var_sum:.4f} xcheck={xcheck_sum:.4f} "
f"total={total:.6f}"
)
def test_correct_priority_columns_reduced(self):
"""T1 boost reduces the highest-priority non-zero column (double_cf).
What: The base pitcher has double_cf=2.95. After T1 boost the TB
budget algorithm starts with double_cf (cost=2 TB per chance) and
should reduce it by 0.75 chances (1.5 TB / 2 TB-per-chance).
Why: Validates that the pitcher priority algorithm is applied correctly
in the orchestration layer and that the outcome columns propagate to
the ratings row.
"""
player = _make_player(pos="SP")
team = _make_team(abbrev="PT4", gmid=99013)
track = _make_track(name="SP Track4", card_type="sp")
_make_state(player, team, track)
_create_base_pitcher(player)
apply_tier_boost(player.player_id, team.id, 1, "sp", **_DEFAULT_KWARGS)
new_variant = compute_variant_hash(player.player_id, 1)
new_card = PitchingCard.get(
(PitchingCard.player == player) & (PitchingCard.variant == new_variant)
)
vr_row = PitchingCardRatings.get(
(PitchingCardRatings.pitchingcard == new_card)
& (PitchingCardRatings.vs_hand == "R")
)
# Budget=1.5, double_cf cost=2 -> takes 0.75 chances
expected_double_cf = _BASE_PITCHER_RATINGS["double_cf"] - 0.75
expected_strikeout = _BASE_PITCHER_RATINGS["strikeout"] + 0.75
assert vr_row.double_cf == pytest.approx(expected_double_cf, abs=1e-4)
assert vr_row.strikeout == pytest.approx(expected_strikeout, abs=1e-4)
# ---------------------------------------------------------------------------
# TestCumulativeT1ToT4
# ---------------------------------------------------------------------------
class TestCumulativeT1ToT4:
"""Apply all 4 tiers sequentially to a batter; verify cumulative state."""
def test_four_variant_cards_exist(self):
"""After T1 through T4, four distinct BattingCard rows exist (plus variant=0).
What: Call apply_tier_boost for tiers 1, 2, 3, 4 sequentially.
Assert that there are exactly 5 BattingCard rows for the player
(1 base + 4 variants).
Why: Each tier must create exactly one new variant card. If tiers
share a card or skip creating one, the tier progression is broken.
"""
player = _make_player()
team = _make_team(abbrev="CT1", gmid=99020)
track = _make_track(name="Cumulative Track")
_make_state(player, team, track)
_create_base_batter(player)
for tier in range(1, 5):
apply_tier_boost(
player.player_id, team.id, tier, "batter", **_DEFAULT_KWARGS
)
all_cards = list(BattingCard.select().where(BattingCard.player == player))
assert len(all_cards) == 5, (
f"Expected 5 BattingCard rows (1 base + 4 variants), got {len(all_cards)}"
)
variants = {c.variant for c in all_cards}
assert 0 in variants, "Base card (variant=0) should still exist"
def test_each_tier_sums_to_108(self):
"""After each of the four sequential boosts, ratings sum to 108.
What: Apply T1 through T4 sequentially and after each tier verify
that all ratings rows for the new variant card sum to 108.
Why: Each boost sources from the previous tier's variant. Any
drift introduced by one tier compounds into subsequent tiers.
Checking after each tier catches drift early rather than only at T4.
"""
player = _make_player()
team = _make_team(abbrev="CT2", gmid=99021)
track = _make_track(name="Cumulative Track2")
_make_state(player, team, track)
_create_base_batter(player)
for tier in range(1, 5):
apply_tier_boost(
player.player_id, team.id, tier, "batter", **_DEFAULT_KWARGS
)
new_variant = compute_variant_hash(player.player_id, tier)
new_card = BattingCard.get(
(BattingCard.player == player) & (BattingCard.variant == new_variant)
)
for row in BattingCardRatings.select().where(
BattingCardRatings.battingcard == new_card
):
total = sum(getattr(row, col) for col in BATTER_OUTCOME_COLUMNS)
assert abs(total - 108.0) < 0.01, (
f"108-sum violated at tier {tier} vs_hand={row.vs_hand}: "
f"sum={total:.6f}"
)
def test_t4_has_cumulative_deltas(self):
"""T4 ratings equal base + 4 cumulative boost deltas.
What: After applying T1 through T4, verify that the T4 variant's
homerun column is approximately base_homerun + 4 * 0.5 = base + 2.0.
Why: Each tier applies +0.5 to homerun. Four sequential tiers must
produce a cumulative +2.0 delta. If the algorithm sources from the
wrong base (e.g. always from variant=0 instead of the previous tier's
variant), the cumulative delta would be wrong.
"""
player = _make_player()
team = _make_team(abbrev="CT3", gmid=99022)
track = _make_track(name="Cumulative Track3")
_make_state(player, team, track)
_create_base_batter(player)
for tier in range(1, 5):
apply_tier_boost(
player.player_id, team.id, tier, "batter", **_DEFAULT_KWARGS
)
t4_variant = compute_variant_hash(player.player_id, 4)
t4_card = BattingCard.get(
(BattingCard.player == player) & (BattingCard.variant == t4_variant)
)
vr_row = BattingCardRatings.get(
(BattingCardRatings.battingcard == t4_card)
& (BattingCardRatings.vs_hand == "R")
)
expected_homerun = _BASE_BATTER_RATINGS["homerun"] + 4 * 0.5
assert vr_row.homerun == pytest.approx(expected_homerun, abs=0.01), (
f"T4 homerun expected {expected_homerun}, got {vr_row.homerun}"
)
# ---------------------------------------------------------------------------
# TestIdempotency
# ---------------------------------------------------------------------------
class TestIdempotency:
"""Calling apply_tier_boost twice for the same tier produces no duplicates."""
def test_no_duplicate_card(self):
"""Second call for the same tier reuses the existing variant card.
What: Call apply_tier_boost for T1 twice. Assert that there is still
exactly one BattingCard with the T1 variant hash (not two).
Why: Idempotency is required because evaluate-game may be called
multiple times for the same game if the bot retries. Duplicate cards
would corrupt the inventory and break the unique-variant DB constraint.
"""
player = _make_player()
team = _make_team(abbrev="ID1", gmid=99030)
track = _make_track(name="Idempotency Track")
_make_state(player, team, track)
_create_base_batter(player)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
t1_variant = compute_variant_hash(player.player_id, 1)
cards = list(
BattingCard.select().where(
(BattingCard.player == player) & (BattingCard.variant == t1_variant)
)
)
assert len(cards) == 1, (
f"Expected exactly 1 T1 variant card, found {len(cards)}"
)
def test_no_duplicate_ratings(self):
"""Second call for the same tier creates only one ratings row per split.
What: Call apply_tier_boost for T1 twice. Assert that each vs_hand
split has exactly one BattingCardRatings row on the variant card.
Why: Duplicate ratings rows for the same (card, vs_hand) combination
would cause the game engine to pick an arbitrary row, producing
non-deterministic outcomes.
"""
player = _make_player()
team = _make_team(abbrev="ID2", gmid=99031)
track = _make_track(name="Idempotency Track2")
_make_state(player, team, track)
_create_base_batter(player)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
t1_variant = compute_variant_hash(player.player_id, 1)
t1_card = BattingCard.get(
(BattingCard.player == player) & (BattingCard.variant == t1_variant)
)
ratings_count = (
BattingCardRatings.select()
.where(BattingCardRatings.battingcard == t1_card)
.count()
)
assert ratings_count == 2, (
f"Expected 2 ratings rows (vL + vR), found {ratings_count}"
)
# ---------------------------------------------------------------------------
# TestCardVariantPropagation
# ---------------------------------------------------------------------------
class TestCardVariantPropagation:
"""Card.variant is propagated to all matching (player, team) Card rows."""
def test_multiple_cards_updated(self):
"""Three Card rows for same player/team all get variant updated after boost.
What: Create 3 Card rows for the same (player, team) pair with
variant=0. Call apply_tier_boost for T1. Assert that all 3 Card
rows now have variant equal to the T1 hash.
Why: A player may have multiple Card instances (e.g. from different
cardsets or pack types). All must reflect the current tier's variant
so that any display pathway shows the boosted art.
"""
player = _make_player()
team = _make_team(abbrev="VP1", gmid=99040)
track = _make_track(name="Propagation Track")
_make_state(player, team, track)
_create_base_batter(player)
cs, _ = Cardset.get_or_create(
name="Prop Test Set", defaults={"description": "prop", "total_cards": 10}
)
rarity = _make_rarity()
card_ids = []
for _ in range(3):
c = Card.create(
player=player,
team=team,
variant=0,
cardset=cs,
rarity=rarity,
price=10,
)
card_ids.append(c.id)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
for cid in card_ids:
updated = Card.get_by_id(cid)
assert updated.variant == expected_variant, (
f"Card id={cid} variant not updated: "
f"expected {expected_variant}, got {updated.variant}"
)
def test_other_teams_unaffected(self):
"""Card rows for the same player on a different team are NOT updated.
What: Create a Card for (player, team_a) and a Card for (player, team_b).
Call apply_tier_boost for T1 on team_a only. Assert that the team_b
card's variant is still 0.
Why: Variant propagation is scoped to (player, team) — applying a
boost for one team must not bleed into another team's card instances.
"""
player = _make_player()
team_a = _make_team(abbrev="VA1", gmid=99041)
team_b = _make_team(abbrev="VA2", gmid=99042)
track = _make_track(name="Propagation Track2")
_make_state(player, team_a, track)
_create_base_batter(player)
cs, _ = Cardset.get_or_create(
name="Prop Test Set2", defaults={"description": "prop2", "total_cards": 10}
)
rarity = _make_rarity()
card_a = Card.create(
player=player, team=team_a, variant=0, cardset=cs, rarity=rarity, price=10
)
card_b = Card.create(
player=player, team=team_b, variant=0, cardset=cs, rarity=rarity, price=10
)
apply_tier_boost(player.player_id, team_a.id, 1, "batter", **_DEFAULT_KWARGS)
# team_a card should be updated
updated_a = Card.get_by_id(card_a.id)
assert updated_a.variant != 0
# team_b card should still be 0
updated_b = Card.get_by_id(card_b.id)
assert updated_b.variant == 0, (
f"team_b card was unexpectedly updated to variant={updated_b.variant}"
)
# ---------------------------------------------------------------------------
# TestCardTypeValidation
# ---------------------------------------------------------------------------
class TestCardTypeValidation:
"""apply_tier_boost rejects invalid card_type values."""
def test_invalid_card_type_raises_value_error(self):
"""card_type='dh' is not valid — must be 'batter', 'sp', or 'rp'.
What: Call apply_tier_boost() with card_type='dh'. Assert that a
ValueError is raised before any DB interaction occurs.
Why: The card_type guard runs at the top of apply_tier_boost(), before
any model lookup. Passing an unsupported type would silently use the
wrong boost formula; an early ValueError prevents corrupted data.
No DB setup is needed because the raise happens before any model is
touched.
"""
with pytest.raises(ValueError, match=r"Invalid card_type"):
apply_tier_boost(1, 1, 1, "dh", **_DEFAULT_KWARGS)
def test_empty_card_type_raises_value_error(self):
"""Empty string card_type is rejected before any DB interaction.
What: Call apply_tier_boost() with card_type=''. Assert that a
ValueError is raised before any DB interaction occurs.
Why: An empty string is not one of the three valid types and must be
caught by the same guard that rejects 'dh'. Ensures the validation
uses an allowlist check rather than a partial-string check that might
accidentally pass an empty value.
"""
with pytest.raises(ValueError, match=r"Invalid card_type"):
apply_tier_boost(1, 1, 1, "", **_DEFAULT_KWARGS)
# ---------------------------------------------------------------------------
# TestCrossPlayerIsolation
# ---------------------------------------------------------------------------
class TestCrossPlayerIsolation:
"""Boosting one player must not affect another player's cards on the same team."""
def test_boost_does_not_update_other_players_cards_on_same_team(self):
"""Two players on the same team. Boost player_a. Player_b's Card.variant stays 0.
What: Create two players on the same team, each with a base BattingCard,
a RefractorCardState, and a Card inventory row (variant=0). Call
apply_tier_boost for player_a only. Assert that player_a's Card row
is updated to the new variant hash while player_b's Card row remains
at variant=0.
Why: The variant propagation query inside apply_tier_boost() filters on
both player_id AND team_id. This test guards against a regression where
only the team_id filter is applied, which would update every player's
Card row on that team.
"""
player_a = _make_player(name="Player A Cross", pos="1B")
player_b = _make_player(name="Player B Cross", pos="2B")
team = _make_team(abbrev="CP1", gmid=99080)
track = _make_track(name="CrossPlayer Track")
_make_state(player_a, team, track)
_make_state(player_b, team, track)
_create_base_batter(player_a)
_create_base_batter(player_b)
Card.create(player=player_a, team=team, variant=0)
Card.create(player=player_b, team=team, variant=0)
apply_tier_boost(player_a.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
card_a = Card.get((Card.player == player_a) & (Card.team == team))
assert card_a.variant != 0, (
f"Player A's card variant should have been updated from 0, "
f"got {card_a.variant}"
)
card_b = Card.get((Card.player == player_b) & (Card.team == team))
assert card_b.variant == 0, (
f"Player B's card variant should still be 0, got {card_b.variant}"
)
# ---------------------------------------------------------------------------
# TestAtomicity
# ---------------------------------------------------------------------------
class TestAtomicity:
"""Tier increment and variant writes are guarded so a mid-boost failure
does not leave RefractorCardState partially updated.
One test verifies that a pre-atomic failure (missing source card) does
not touch the state row at all. A second test verifies that a failure
INSIDE the db.atomic() block rolls back the state mutations even though
the card and ratings rows (created before the block) persist.
"""
def test_missing_source_card_does_not_modify_current_tier(self):
"""Failed boost due to missing source card does not modify current_tier.
What: Set up a RefractorCardState but do NOT create a BattingCard.
Call apply_tier_boost for T1 — it should raise ValueError because
the source card is missing. After the failure, current_tier must
still be 0.
Why: The ValueError is raised at the source-card fetch step, which
happens BEFORE the db.atomic() block. No write is ever attempted,
so the state row should be completely untouched. This guards against
regressions where early validation logic is accidentally removed and
writes are attempted with bad data.
"""
player = _make_player()
team = _make_team(abbrev="AT1", gmid=99050)
track = _make_track(name="Atomicity Track")
state = _make_state(player, team, track, current_tier=0)
# Intentionally no base card created
with pytest.raises(ValueError, match=r"No battingcard.*player="):
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier == 0, (
f"current_tier should still be 0 after failed boost, got {state.current_tier}"
)
def test_audit_failure_inside_atomic_rolls_back_state_mutations(self):
"""Failure inside db.atomic() rolls back card_state tier and variant writes.
What: Create a valid base BattingCard with ratings so the function
progresses past source-card validation and card/ratings creation.
Override _audit_model with a stub whose .create() method raises
RuntimeError — this triggers inside the db.atomic() block (step 8a),
after the new card and ratings rows have already been written (steps
6-7, which are outside the atomic block).
After the exception propagates out, verify that RefractorCardState
still has current_tier==0 and variant==None (the pre-boost values).
The atomic rollback must prevent the card_state.save() and
Card.update() writes from being committed even though card/ratings
rows persist (they were written before the atomic block began).
Why: db.atomic() guarantees that either ALL writes inside the block
are committed together or NONE are. If this guarantee breaks, a
partially-committed state would show a tier advance without a
corresponding audit record, leaving the card in an inconsistent state
that is invisible to the evaluator's retry logic.
"""
class _FailingAuditModel:
"""Stub that raises on .create() to simulate audit write failure.
Provides card_state/tier attributes for the Peewee expression in the
idempotency guard, and get_or_none returns None so it proceeds to
create(), which then raises to simulate the failure.
"""
card_state = RefractorBoostAudit.card_state
tier = RefractorBoostAudit.tier
@staticmethod
def get_or_none(*args, **kwargs):
return None
@staticmethod
def create(**kwargs):
raise RuntimeError("Simulated audit write failure inside atomic block")
player = _make_player(name="Atomic Rollback Player", pos="CF")
team = _make_team(abbrev="RB1", gmid=99052)
track = _make_track(name="Atomicity Track3")
state = _make_state(player, team, track, current_tier=0)
_create_base_batter(player)
failing_kwargs = dict(_DEFAULT_KWARGS)
failing_kwargs["_audit_model"] = _FailingAuditModel
with pytest.raises(RuntimeError, match="Simulated audit write failure"):
apply_tier_boost(player.player_id, team.id, 1, "batter", **failing_kwargs)
# The atomic block should have rolled back — state row must be unchanged.
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier == 0, (
f"current_tier should still be 0 after atomic rollback, got {state.current_tier}"
)
assert state.variant is None or state.variant == 0, (
f"variant should not have been updated after atomic rollback, got {state.variant}"
)
def test_successful_boost_writes_tier_and_variant_together(self):
"""Successful boost atomically writes current_tier and variant.
What: After a successful T1 boost, both current_tier == 1 AND
variant == compute_variant_hash(player_id, 1) must be true on the
same RefractorCardState row.
Why: If the writes were not atomic, a read between the tier write
and the variant write could see an inconsistent state. The atomic
block ensures both are committed together.
"""
player = _make_player()
team = _make_team(abbrev="AT2", gmid=99051)
track = _make_track(name="Atomicity Track2")
state = _make_state(player, team, track)
_create_base_batter(player)
expected_variant = compute_variant_hash(player.player_id, 1)
apply_tier_boost(player.player_id, team.id, 1, "batter", **_DEFAULT_KWARGS)
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier == 1
assert state.variant == expected_variant
# ---------------------------------------------------------------------------
# TestEvaluateCardDryRun
# ---------------------------------------------------------------------------
def _make_stats_for_t1(player, team):
"""Create a BattingSeasonStats row to give the player enough value for T1.
The default track formula is 'pa + tb * 2'. T1 threshold is 37.
We seed 40 PA with 0 extra bases so value = 40 > 37.
"""
return BattingSeasonStats.create(
player=player,
team=team,
season=11,
pa=40,
)
class TestEvaluateCardDryRun:
"""evaluate_card(dry_run=True) computes the new tier without writing it."""
def test_dry_run_does_not_write_current_tier(self):
"""dry_run=True evaluation leaves current_tier at 0 even when formula says tier=1.
What: Seed enough stats for T1 (pa=40 > threshold 37). Call
evaluate_card with dry_run=True. Assert current_tier is still 0.
Why: The dry_run=True path must not write current_tier so that
apply_tier_boost() can write it atomically with the variant card.
If current_tier were written here, a subsequent boost failure would
leave the tier advanced but no variant created.
"""
player = _make_player()
team = _make_team(abbrev="DR1", gmid=99060)
track = _make_track(name="DryRun Track")
state = _make_state(player, team, track, current_tier=0)
_make_stats_for_t1(player, team)
evaluate_card(
player.player_id,
team.id,
dry_run=True,
_state_model=RefractorCardState,
)
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier == 0, (
f"dry_run=True must not write current_tier; got {state.current_tier}"
)
def test_dry_run_writes_current_value(self):
"""dry_run=True DOES update current_value even though current_tier is skipped.
What: After evaluate_card(dry_run=True) with pa=40 stats, current_value
must be > 0 (formula: pa + tb*2 = 40).
Why: current_value must be updated so that progress display (progress_pct)
reflects the latest stats even when boost is pending.
"""
player = _make_player()
team = _make_team(abbrev="DR2", gmid=99061)
track = _make_track(name="DryRun Track2")
state = _make_state(player, team, track, current_tier=0)
_make_stats_for_t1(player, team)
evaluate_card(
player.player_id,
team.id,
dry_run=True,
_state_model=RefractorCardState,
)
state = RefractorCardState.get_by_id(state.id)
assert state.current_value > 0, (
f"dry_run=True should still update current_value; got {state.current_value}"
)
def test_dry_run_returns_computed_tier(self):
"""Return dict includes 'computed_tier' reflecting the formula result.
What: With pa=40 (value=40 > T1 threshold=37), evaluate_card with
dry_run=True must return {'computed_tier': 1, 'current_tier': 0, ...}.
Why: The evaluate-game endpoint uses computed_tier to detect tier-ups.
If it were absent or equal to current_tier, the endpoint would not
call apply_tier_boost() and no boost would be applied.
"""
player = _make_player()
team = _make_team(abbrev="DR3", gmid=99062)
track = _make_track(name="DryRun Track3")
_make_state(player, team, track, current_tier=0)
_make_stats_for_t1(player, team)
result = evaluate_card(
player.player_id,
team.id,
dry_run=True,
_state_model=RefractorCardState,
)
assert "computed_tier" in result, "Return dict must include 'computed_tier'"
assert result["computed_tier"] >= 1, (
f"computed_tier should be >= 1 with pa=40, got {result['computed_tier']}"
)
assert result["current_tier"] == 0, (
f"current_tier must remain 0 in dry_run mode, got {result['current_tier']}"
)
def test_non_dry_run_preserves_existing_behaviour(self):
"""dry_run=False (default) writes current_tier as before.
What: With pa=40 (value=40 > T1 threshold=37), evaluate_card with
dry_run=False must write current_tier=1 to the database.
Why: The manual /evaluate endpoint uses dry_run=False. Existing
behaviour must be preserved so that cards can still be manually
re-evaluated without a boost cycle.
"""
player = _make_player()
team = _make_team(abbrev="DR4", gmid=99063)
track = _make_track(name="DryRun Track4")
state = _make_state(player, team, track, current_tier=0)
_make_stats_for_t1(player, team)
evaluate_card(
player.player_id,
team.id,
dry_run=False,
_state_model=RefractorCardState,
)
state = RefractorCardState.get_by_id(state.id)
assert state.current_tier >= 1, (
f"dry_run=False must write current_tier; got {state.current_tier}"
)
Reference in New Issue View Git Blame Copy Permalink