Spaces:
Sleeping
Sleeping
File size: 5,444 Bytes
b7a5bab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | import random
import pytest
from store import OrderRequest, PriceUpdate, StoreAction, StoreEnv
from store.models import dump_model
from store.server.store_environment import StoreEnvironment
def test_reset_returns_full_plan_like_observation() -> None:
env = StoreEnvironment()
observation = env.reset()
assert observation.store.day == 1
assert len(observation.stock) == 30
assert len(observation.market) == 100
assert len(observation.marketing_actions) == 15
assert len(observation.seller_profiles) == 4
assert observation.pending_orders == []
assert observation.available_action.pass_action == StoreAction()
assert observation.available_action.example_action.action_ids == [7]
def test_pass_step_advances_three_days_but_reward_uses_six_day_forecast() -> None:
env = StoreEnvironment()
env.reset()
observation = env.step(StoreAction())
assert observation.store.day == 4
assert observation.metadata["reward_breakdown"]["commit_window_days"] == 3
assert observation.metadata["reward_breakdown"]["reward_horizon_days"] == 6
def test_eta_sampling_is_deterministic_for_a_seed() -> None:
env_a = StoreEnvironment()
env_b = StoreEnvironment()
env_a.reset()
env_b.reset()
env_a._rng = random.Random(7)
env_b._rng = random.Random(7)
assert env_a._sample_actual_eta("Seller C", 5) == env_b._sample_actual_eta("Seller C", 5)
def test_commit_state_only_keeps_three_days_of_progress() -> None:
env = StoreEnvironment()
env.reset()
env._sample_actual_eta = lambda seller, quoted: 4 # type: ignore[method-assign]
observation = env.step(
StoreAction(
orders=[OrderRequest(product="Rice", seller="Seller B", quantity=10)],
)
)
assert observation.store.day == 4
assert len(observation.pending_orders) == 1
assert observation.pending_orders[0].days_remaining == 1
def test_price_update_changes_customer_satisfaction() -> None:
env = StoreEnvironment()
reset_observation = env.reset()
milk_row = next(row for row in reset_observation.stock if row.product_name == "Milk")
observation = env.step(
StoreAction(
prices=[
PriceUpdate(
product="Milk",
price=max(1, int(milk_row.selling_price_inr) - 5),
)
]
)
)
assert observation.store.customer_satisfaction > reset_observation.store.customer_satisfaction
def test_marketing_actions_are_durable_and_cannot_be_reactivated() -> None:
env = StoreEnvironment()
env.reset()
observation = env.step(StoreAction(action_ids=[7]))
assert observation.store.active_action_ids == [7]
assert observation.store.active_recurring_cost_inr > 0
assert any(
action.action_id == 7 and action.is_active for action in observation.marketing_actions
)
with pytest.raises(ValueError):
env.step(StoreAction(action_ids=[7]))
def test_recurring_costs_are_deducted_each_simulated_day() -> None:
env = StoreEnvironment()
env.reset()
runtime = env._build_initial_runtime()
runtime["store"]["bank_balance_inr"] = 10_000.0
runtime["store"]["cumulative_profit_inr"] = 0.0
runtime["store"]["current_daily_expense_inr"] = 0.0
runtime["store"]["customer_satisfaction"] = 0.0
runtime["active_action_ids"] = {7}
metrics = env._simulate_days(runtime, 3)
assert metrics["recurring_costs_paid_inr"] == 1_800.0
assert runtime["store"]["bank_balance_inr"] == 8_200.0
assert runtime["store"]["cumulative_profit_inr"] == -1_800.0
def test_large_capital_loss_reduces_reward_and_ends_episode() -> None:
baseline = StoreEnvironment()
baseline.reset()
baseline_reward = baseline.step(StoreAction()).reward
env = StoreEnvironment()
env.reset()
observation = env.step(StoreAction(action_ids=[11]))
assert observation.reward < baseline_reward
assert observation.done is True
assert observation.metadata["done_reason"] == "bank_balance_below_zero"
def test_episode_ends_after_tenth_decision_window() -> None:
env = StoreEnvironment()
env.reset()
observation = None
for _ in range(9):
observation = env.step(StoreAction())
assert observation.done is False
observation = env.step(StoreAction())
assert observation is not None
assert observation.done is True
assert observation.store.day == 31
assert observation.metadata["done_reason"] == "episode_horizon_reached"
def test_client_payload_and_result_parsing_support_nested_models() -> None:
client = StoreEnv(base_url="http://localhost:8000")
action = StoreAction(
orders=[OrderRequest(product="Rice", seller="Seller A", quantity=5)],
prices=[PriceUpdate(product="Milk", price=33)],
action_ids=[7],
)
payload = client._step_payload(action)
assert payload == {
"orders": [{"product": "Rice", "seller": "Seller A", "quantity": 5}],
"prices": [{"product": "Milk", "price": 33}],
"action_ids": [7],
}
env = StoreEnvironment()
observation = env.reset()
result = client._parse_result(
{
"observation": dump_model(observation, exclude_none=True),
"reward": 0.0,
"done": False,
}
)
assert result.observation.store.day == 1
assert len(result.observation.market) == 100
|