opencompletion.com/tests/unit/test_random_buckets.py
Claude 22db7a9a8a
Run black formatter on all Python files
Format code according to black style guidelines for consistency
2025-11-10 19:37:05 +00:00

637 lines
24 KiB
Python

#!/usr/bin/env python3
"""
Unit tests for random bucket rolling feature
Tests the random bucket system:
- Random bucket probability rolling
- Multi-bucket triggering and processing
- String concatenation in metadata (n+,value)
- Navigation resolution with multiple buckets
- Attempt counting with multiple buckets
"""
import unittest
import random
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
class TestRandomBucketRolling(unittest.TestCase):
"""Test cases for random bucket probability rolling"""
def test_random_bucket_triggers_when_roll_below_probability(self):
"""Test that random bucket triggers when roll < probability"""
step = {"random_buckets": {"emergency": {"probability": 0.5}}}
with patch("random.random", return_value=0.3): # 0.3 < 0.5
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
self.assertIn("emergency", triggered_buckets)
self.assertEqual(len(triggered_buckets), 1)
def test_random_bucket_does_not_trigger_when_roll_above_probability(self):
"""Test that random bucket doesn't trigger when roll >= probability"""
step = {"random_buckets": {"emergency": {"probability": 0.5}}}
with patch("random.random", return_value=0.7): # 0.7 >= 0.5
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
self.assertEqual(len(triggered_buckets), 0)
def test_multiple_random_buckets_can_trigger_simultaneously(self):
"""Test that multiple random buckets can trigger on same turn"""
step = {
"random_buckets": {
"emergency": {"probability": 0.5},
"task": {"probability": 0.5},
}
}
# Mock random to always return low values
with patch("random.random", return_value=0.2): # 0.2 < 0.5 for both
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
self.assertEqual(len(triggered_buckets), 2)
self.assertIn("emergency", triggered_buckets)
self.assertIn("task", triggered_buckets)
def test_double_trigger_with_20_iterations(self):
"""Test that double-triggering happens within 20 iterations"""
step = {
"random_buckets": {
"emergency": {"probability": 0.15},
"task": {"probability": 0.15},
}
}
double_trigger_found = False
iterations = 0
# Try up to 20 times to find a double trigger
for i in range(20):
iterations += 1
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
if len(triggered_buckets) == 2:
double_trigger_found = True
print(
f"✓ Double trigger found on iteration {iterations}: {triggered_buckets}"
)
break
# With 15% probability each, chance of both triggering = 0.15 * 0.15 = 0.0225 (2.25%)
# Over 20 trials, probability of at least one double = 1 - (1 - 0.0225)^20 ≈ 36%
# This test may occasionally fail due to randomness, but should pass most of the time
if not double_trigger_found:
print(
f"⚠️ Warning: No double trigger found in {iterations} iterations (expected ~36% success rate)"
)
# We don't assert here because random tests can fail
# Instead we just report the result
self.assertLessEqual(iterations, 20)
def test_triple_trigger_with_20_iterations(self):
"""Test that triple-triggering happens within 20 iterations"""
step = {
"random_buckets": {
"emergency": {"probability": 1.0}, # 100% to prevent flaky tests
"task": {"probability": 1.0}, # 100% to prevent flaky tests
"challenge": {"probability": 1.0}, # 100% to prevent flaky tests
}
}
triple_trigger_found = False
iterations = 0
# Try up to 20 times to find a triple trigger (should succeed on first try with 100%)
for i in range(20):
iterations += 1
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
if len(triggered_buckets) == 3:
triple_trigger_found = True
print(
f"✓ Triple trigger found on iteration {iterations}: {triggered_buckets}"
)
break
# With 100% probability each, all three should trigger on first iteration
self.assertTrue(
triple_trigger_found,
"Triple trigger should have been found with 100% probabilities",
)
self.assertEqual(
iterations,
1,
"Triple trigger should happen on first iteration with 100% probabilities",
)
def test_zero_probability_never_triggers(self):
"""Test that 0% probability never triggers"""
step = {"random_buckets": {"impossible": {"probability": 0.0}}}
# Try 100 times - should never trigger
for _ in range(100):
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
self.assertEqual(len(triggered_buckets), 0)
def test_100_percent_probability_always_triggers(self):
"""Test that 100% probability always triggers"""
step = {"random_buckets": {"guaranteed": {"probability": 1.0}}}
# Try 10 times - should always trigger
for _ in range(10):
triggered_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_buckets.append(bucket_name)
self.assertEqual(len(triggered_buckets), 1)
self.assertIn("guaranteed", triggered_buckets)
class TestMultiBucketProcessing(unittest.TestCase):
"""Test cases for processing multiple active buckets"""
def test_user_bucket_processed_first(self):
"""Test that user's response bucket is processed before random events"""
user_category = "navigation"
triggered_random_buckets = ["emergency", "task"]
all_active_buckets = [user_category] + triggered_random_buckets
self.assertEqual(all_active_buckets[0], "navigation")
self.assertEqual(all_active_buckets[1], "emergency")
self.assertEqual(all_active_buckets[2], "task")
def test_last_bucket_navigation_wins(self):
"""Test that LAST bucket's next_section_and_step wins"""
transitions = [
("navigation", {"next_section_and_step": "section_1:step_1"}),
("emergency", {"next_section_and_step": "section_2:step_2"}),
("task", {"next_section_and_step": "section_3:step_3"}),
]
final_next_section_and_step = None
for bucket_name, transition in transitions:
if "next_section_and_step" in transition:
final_next_section_and_step = transition["next_section_and_step"]
self.assertEqual(final_next_section_and_step, "section_3:step_3")
def test_any_bucket_counts_as_attempt(self):
"""Test that if ANY bucket counts, the turn counts"""
transitions = [
("navigation", {"counts_as_attempt": False}),
("emergency", {"counts_as_attempt": True}),
("task", {"counts_as_attempt": False}),
]
any_counts_as_attempt = False
for bucket_name, transition in transitions:
if transition.get("counts_as_attempt", True):
any_counts_as_attempt = True
self.assertTrue(any_counts_as_attempt)
def test_no_bucket_counts_when_all_false(self):
"""Test that turn doesn't count when all buckets have counts_as_attempt: false"""
transitions = [
("navigation", {"counts_as_attempt": False}),
("hint", {"counts_as_attempt": False}),
]
any_counts_as_attempt = False
for bucket_name, transition in transitions:
if transition.get("counts_as_attempt", True):
any_counts_as_attempt = True
self.assertFalse(any_counts_as_attempt)
def test_metadata_accumulates_across_buckets(self):
"""Test that metadata accumulates from all active buckets"""
metadata = {"score": 0}
transitions = [
("navigation", {"metadata_add": {"score": "n+10"}}),
("emergency", {"metadata_add": {"emergency_count": "n+1"}}),
("task", {"metadata_add": {"task_count": "n+1"}}),
]
# Simulate processing all transitions
for bucket_name, transition in transitions:
if "metadata_add" in transition:
for key, value in transition["metadata_add"].items():
if isinstance(value, str) and value.startswith("n+"):
# Numeric increment
increment = int(value[2:])
metadata[key] = metadata.get(key, 0) + increment
else:
metadata[key] = value
self.assertEqual(metadata["score"], 10)
self.assertEqual(metadata["emergency_count"], 1)
self.assertEqual(metadata["task_count"], 1)
class TestStringConcatenationMetadata(unittest.TestCase):
"""Test cases for string concatenation in metadata operations"""
def test_string_append_to_empty(self):
"""Test appending to empty metadata value"""
metadata = {}
key = "visited_sections"
value = "n+,torpedo_room"
if value.startswith("n+,"):
suffix = value[3:]
existing_value = metadata.get(key, "")
if existing_value:
metadata[key] = f"{existing_value},{suffix}"
else:
metadata[key] = suffix
self.assertEqual(metadata["visited_sections"], "torpedo_room")
def test_string_append_to_existing(self):
"""Test appending to existing comma-separated value"""
metadata = {"visited_sections": "forward_escape_trunk"}
key = "visited_sections"
value = "n+,torpedo_room"
if value.startswith("n+,"):
suffix = value[3:]
existing_value = metadata.get(key, "")
if existing_value:
metadata[key] = f"{existing_value},{suffix}"
else:
metadata[key] = suffix
self.assertEqual(
metadata["visited_sections"], "forward_escape_trunk,torpedo_room"
)
def test_string_append_multiple_times(self):
"""Test multiple append operations"""
metadata = {}
values = ["n+,room1", "n+,room2", "n+,room3"]
for value in values:
if value.startswith("n+,"):
suffix = value[3:]
existing_value = metadata.get("visited_sections", "")
if existing_value:
metadata["visited_sections"] = f"{existing_value},{suffix}"
else:
metadata["visited_sections"] = suffix
self.assertEqual(metadata["visited_sections"], "room1,room2,room3")
def test_string_remove_from_list(self):
"""Test removing value from comma-separated list"""
metadata = {"visited_sections": "room1,room2,room3"}
key = "visited_sections"
value = "n-,room2"
if value.startswith("n-,"):
suffix = value[3:]
existing_value = metadata.get(key, "")
if existing_value:
parts = existing_value.split(",")
parts = [p for p in parts if p != suffix]
metadata[key] = ",".join(parts)
self.assertEqual(metadata["visited_sections"], "room1,room3")
def test_numeric_increment_still_works(self):
"""Test that numeric operations still work (n+5, not n+,5)"""
metadata = {"score": 10}
key = "score"
value = "n+5"
if value.startswith("n+") and not value.startswith("n+,"):
# Numeric operation
increment = int(value[2:])
metadata[key] = metadata.get(key, 0) + increment
self.assertEqual(metadata["score"], 15)
def test_numeric_decrement_still_works(self):
"""Test that numeric decrement works (n-5)"""
metadata = {"health": 100}
key = "health"
value = "n-20"
if value.startswith("n-") and not value.startswith("n-,"):
# Numeric operation
decrement = int(value[2:])
metadata[key] = metadata.get(key, 0) - decrement
self.assertEqual(metadata["health"], 80)
def test_distinguish_string_vs_numeric_operations(self):
"""Test that we correctly distinguish n+,value vs n+5"""
metadata = {}
# String concatenation
value1 = "n+,room1"
if value1.startswith("n+,"):
suffix = value1[3:]
metadata["rooms"] = suffix
# Numeric increment
value2 = "n+10"
if value2.startswith("n+") and not value2.startswith("n+,"):
increment = int(value2[2:])
metadata["score"] = metadata.get("score", 0) + increment
self.assertEqual(metadata["rooms"], "room1")
self.assertEqual(metadata["score"], 10)
class TestRandomBucketIntegration(unittest.TestCase):
"""Integration tests for complete random bucket workflow"""
def test_complete_workflow_single_trigger(self):
"""Test complete workflow with one random event"""
# Setup
metadata = {"visited_sections": ""}
user_response = "forward"
category = "torpedo_room"
step = {
"random_buckets": {
"emergency": {"probability": 0.05},
"daily_task": {"probability": 0.15},
},
"transitions": {
"torpedo_room": {
"metadata_add": {
"current_section": "torpedo_room",
"visited_sections": "n+,torpedo_room",
},
"next_section_and_step": "navigation_hub:torpedo_room",
},
"emergency": {
"metadata_add": {"emergency_active": "true"},
"next_section_and_step": "emergency:handle",
},
"daily_task": {
"metadata_add": {"task_active": "true"},
"next_section_and_step": "task:handle",
},
},
}
# Simulate one emergency triggering
triggered_random_buckets = []
with patch("random.random") as mock_random:
# First call: emergency (0.03 < 0.05) - triggers
# Second call: daily_task (0.9 >= 0.15) - doesn't trigger
mock_random.side_effect = [0.03, 0.9]
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_random_buckets.append(bucket_name)
# Combine buckets: user first, then random events
all_active_buckets = [category] + triggered_random_buckets
# Process all transitions
final_next_section_and_step = None
for bucket in all_active_buckets:
transition = step["transitions"][bucket]
# Process metadata_add
if "metadata_add" in transition:
for key, value in transition["metadata_add"].items():
if isinstance(value, str) and value.startswith("n+,"):
suffix = value[3:]
existing = metadata.get(key, "")
metadata[key] = f"{existing},{suffix}" if existing else suffix
else:
metadata[key] = value
# Track navigation
if "next_section_and_step" in transition:
final_next_section_and_step = transition["next_section_and_step"]
# Assertions
self.assertEqual(len(all_active_buckets), 2) # User + 1 random
self.assertIn("torpedo_room", all_active_buckets)
self.assertIn("emergency", all_active_buckets)
self.assertEqual(metadata["visited_sections"], "torpedo_room")
self.assertEqual(metadata["current_section"], "torpedo_room")
self.assertEqual(metadata["emergency_active"], "true")
self.assertEqual(final_next_section_and_step, "emergency:handle") # Last wins
def test_complete_workflow_double_trigger(self):
"""Test complete workflow with two random events"""
metadata = {}
category = "examine"
step = {
"random_buckets": {
"emergency": {"probability": 1.0}, # Guaranteed
"daily_task": {"probability": 1.0}, # Guaranteed
},
"transitions": {
"examine": {
"next_section_and_step": "navigation_hub:forward_escape_trunk",
"counts_as_attempt": False, # Add this so examine doesn't count
},
"emergency": {
"metadata_add": {"emergency_count": "n+1"},
"counts_as_attempt": False,
},
"daily_task": {
"metadata_add": {"task_count": "n+1"},
"counts_as_attempt": False,
},
},
}
# Both random events trigger (100% probability)
triggered_random_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_random_buckets.append(bucket_name)
all_active_buckets = [category] + triggered_random_buckets
# Process all transitions
any_counts_as_attempt = False
for bucket in all_active_buckets:
transition = step["transitions"][bucket]
if "metadata_add" in transition:
for key, value in transition["metadata_add"].items():
if (
isinstance(value, str)
and value.startswith("n+")
and not value.startswith("n+,")
):
increment = int(value[2:])
metadata[key] = metadata.get(key, 0) + increment
if transition.get("counts_as_attempt", True):
any_counts_as_attempt = True
# Assertions - verify double trigger happened
self.assertEqual(len(all_active_buckets), 3) # User + 2 random
self.assertIn("examine", all_active_buckets)
self.assertIn("emergency", all_active_buckets)
self.assertIn("daily_task", all_active_buckets)
self.assertEqual(metadata["emergency_count"], 1)
self.assertEqual(metadata["task_count"], 1)
self.assertFalse(any_counts_as_attempt) # All have counts_as_attempt: false
def test_complete_workflow_triple_trigger(self):
"""Test complete workflow with three random events"""
metadata = {"score": 0}
category = "correct_answer"
step = {
"random_buckets": {
"emergency": {"probability": 1.0}, # Guaranteed
"daily_task": {"probability": 1.0}, # Guaranteed
"bonus_challenge": {"probability": 1.0}, # Guaranteed
},
"transitions": {
"correct_answer": {
"metadata_add": {"score": "n+10"},
"next_section_and_step": "quiz:next_question",
"counts_as_attempt": False,
},
"emergency": {
"metadata_add": {
"emergency_count": "n+1",
"score": "n-5", # Emergency penalty
},
"counts_as_attempt": False,
"next_section_and_step": "emergency:handle",
},
"daily_task": {
"metadata_add": {"task_count": "n+1", "score": "n+2"}, # Task bonus
"counts_as_attempt": False,
},
"bonus_challenge": {
"metadata_add": {
"challenge_count": "n+1",
"score": "n+15", # Big bonus
},
"counts_as_attempt": False,
},
},
}
# All three random events trigger (100% probability)
triggered_random_buckets = []
for bucket_name, config in step["random_buckets"].items():
probability = config.get("probability", 0)
roll = random.random()
if roll < probability:
triggered_random_buckets.append(bucket_name)
all_active_buckets = [category] + triggered_random_buckets
# Process all transitions
any_counts_as_attempt = False
final_next_section_and_step = None
for bucket in all_active_buckets:
transition = step["transitions"][bucket]
if "metadata_add" in transition:
for key, value in transition["metadata_add"].items():
if (
isinstance(value, str)
and value.startswith("n+")
and not value.startswith("n+,")
):
increment = int(value[2:])
metadata[key] = metadata.get(key, 0) + increment
elif (
isinstance(value, str)
and value.startswith("n-")
and not value.startswith("n-,")
):
decrement = int(value[2:])
metadata[key] = metadata.get(key, 0) - decrement
if "next_section_and_step" in transition:
final_next_section_and_step = transition["next_section_and_step"]
if transition.get("counts_as_attempt", True):
any_counts_as_attempt = True
# Assertions - verify triple trigger happened
self.assertEqual(len(all_active_buckets), 4) # User + 3 random
self.assertIn("correct_answer", all_active_buckets)
self.assertIn("emergency", all_active_buckets)
self.assertIn("daily_task", all_active_buckets)
self.assertIn("bonus_challenge", all_active_buckets)
# Verify metadata accumulated from all 4 buckets
self.assertEqual(metadata["emergency_count"], 1)
self.assertEqual(metadata["task_count"], 1)
self.assertEqual(metadata["challenge_count"], 1)
# Verify score calculation: 10 (correct) - 5 (emergency) + 2 (task) + 15 (bonus) = 22
self.assertEqual(metadata["score"], 22)
# Verify last bucket's navigation wins (emergency was last with navigation)
self.assertEqual(final_next_section_and_step, "emergency:handle")
# Verify no attempts counted
self.assertFalse(any_counts_as_attempt)
if __name__ == "__main__":
# Run tests with verbose output
unittest.main(verbosity=2)