"""Real pinned-engine and room regressions. Run with the game's Python. .venvs/werewolf/bin/python games/werewolf/referee/test_rules.py No running Shoal service, account, network, or persisted game data is used. """ import contextlib import json import os from pathlib import Path import random import subprocess import sys import tempfile import threading import unittest HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) with contextlib.redirect_stdout(sys.stderr): from kaggle_environments import make from play import ROLES, configuration, transcript from rule_agent import RuleAgent, public_config from kaggle_seat import seat def record(name, data=None, day=1, phase='Day', source='MODERATOR', stamp=None): return {'event_name': name, 'day': day, 'phase': phase, 'source': source, 'data': data, 'created_at': stamp or f'{day}-{name}-{source}', 'description': '', 'detailed_phase': 'DAY_CHAT_AWAIT'} def observation(role='Villager', me=0, phase='DAY_VOTING_AWAIT', entries=None, alive=None, revealed=None, day=1): ids = [f'Player {i + 1}' for i in range(7)] return {'raw_observation': {'player_id': ids[me], 'role': role, 'team': 'Werewolves' if role == 'Werewolf' else 'Villagers', 'is_alive': True, 'all_player_ids': ids, 'alive_players': alive or ids, 'day': day, 'game_state_phase': 'Night' if phase.startswith('NIGHT') else 'Day', 'detailed_phase': phase, 'revealed_players': revealed or {}, 'new_player_event_views': entries or []}} class Rules(unittest.TestCase): def test_config_hides_assignment_and_seed(self): conf = configuration(123, rng=random.Random(10)) conf.update({'llms': [{'api_key': 'not-a-real-key'}], 'actTimeout': 3}) safe = public_config(conf) self.assertEqual(safe, {'actTimeout': 3}) self.assertNotIn('agents', safe) self.assertNotIn('seed', safe) def test_balanced_roles_and_private_deals(self): for mine, role in enumerate(ROLES): a = configuration(3, mine, rng=random.Random(10 + mine)) b = configuration(3, mine, rng=random.Random(20 + mine)) self.assertEqual(a['agents'][mine]['role'], role) self.assertEqual(sorted(p['role'] for p in a['agents']), sorted(ROLES)) self.assertNotEqual(a['agents'], b['agents']) def test_seer_remembers_and_does_not_repeat(self): p = RuleAgent() result = record('inspect_result', {'actor_id': 'Player 1', 'target_id': 'Player 4', 'role': 'Werewolf'}, phase='Night') vote = p(observation('Seer', entries=[result])) self.assertEqual(vote['kwargs']['target_id'], 'Player 4') req = record('inspect_request', {'valid_candidates': ['Player 4', 'Player 5']}, day=2, phase='Night') action = p(observation('Seer', phase='NIGHT_AWAIT_ACTIONS', entries=[req], day=2)) self.assertEqual(action['kwargs']['target_id'], 'Player 5') new_start = record('game_start', {'player_id': 'Player 1', 'role': 'Seer'}, day=0, stamp='second-game') p.observe(observation('Seer', entries=[new_start], day=0)) self.assertEqual(p.inspections, {}) def test_doctor_trusts_a_consistent_claim_and_legal_list(self): claim = record('discussion', {'actor_id': 'Player 3', 'message': '我是预言家。第1夜查验：4号=好人。'}, source='Player 3') req = record('heal_request', {'valid_candidates': ['Player 2', 'Player 3', 'Player 4']}, day=1, phase='Night') p = RuleAgent() action = p(observation('Doctor', phase='NIGHT_AWAIT_ACTIONS', entries=[claim, req])) self.assertEqual(action['kwargs']['target_id'], 'Player 3') req['data']['valid_candidates'] = ['Player 2', 'Player 4'] action = p(observation('Doctor', phase='NIGHT_AWAIT_ACTIONS', entries=[req])) self.assertIn(action['kwargs']['target_id'], ['Player 2', 'Player 4']) def test_claims_are_not_facts(self): p = RuleAgent() speech = record('discussion', {'actor_id': 'Player 2', 'message': '我是预言家。查验：3号=狼人。'}, source='Player 2') p.observe(observation(entries=[speech], revealed={'Player 3': 'Villager'})) self.assertEqual(p.facts['Player 3'], 'Villager') self.assertEqual(p.claimed('Player 2', 'Seer'), 0) self.assertEqual(p.belief()['Player 3'], 0) p.read_claim('Player 4', '我不是预言家。查验：3号=狼人。') self.assertNotIn('Player 4', p.claims) def test_wolves_coordinate_and_never_vote_their_partner(self): req = record('vote_request', {'valid_targets': ['Player 3', 'Player 4'], 'alive_werewolve_player_ids': ['Player 1', 'Player 2']}, phase='Night') vote = record('vote_action', {'actor_id': 'Player 2', 'target_id': 'Player 4'}, phase='Night', source='Player 2') p = RuleAgent() action = p(observation('Werewolf', phase='NIGHT_AWAIT_ACTIONS', entries=[req, vote])) self.assertEqual(action['kwargs']['target_id'], 'Player 4') action = p(observation('Werewolf')) self.assertNotIn(action['kwargs']['target_id'], ['Player 1', 'Player 2']) def test_last_ballot_uses_random_tie_outcome(self): p = RuleAgent() p.observe(observation('Seer', alive=['Player 1', 'Player 2', 'Player 3'])) p.facts.update({'Player 2': 'Werewolf', 'Player 3': 'Villager'}) p.votes = {1: {'Player 2': 'Player 3', 'Player 3': 'Player 2'}} self.assertEqual(p.ballot(p.alive, p.belief()), 'Player 2') def test_pinned_engine_doctor_can_protect_same_person_again(self): env = make('werewolf', configuration={'seed': 3}) env.reset(7) saves, offered = [], [] for _ in range(80): actions = [None] * 7 for i, state in enumerate(env.state): if state.status != 'ACTIVE': continue raw = state.observation.raw_observation kwargs = {'actor_id': raw.player_id, 'day': raw.day, 'phase': raw.game_state_phase} if raw.detailed_phase == 'DAY_CHAT_AWAIT': kind, kwargs['message'] = 'ChatAction', 'test' elif raw.detailed_phase == 'DAY_VOTING_AWAIT': kind = 'VoteAction' kwargs['target_id'] = 'Player 7' if 'Player 7' in raw.alive_players else 'Player 6' else: kind = {'Werewolf': 'VoteAction', 'Doctor': 'HealAction', 'Seer': 'InspectAction'}[raw.role] kwargs['target_id'] = 'Player 1' if raw.role == 'Seer' else 'Player 5' if raw.role == 'Doctor': req = next(e for e in raw.new_player_event_views if e.event_name == 'heal_request') offered.append(req.data['valid_candidates']) actions[i] = {'action_type': kind, 'kwargs': kwargs} env.step(actions) for rec in env.info['MODERATOR_OBSERVATION'][-1]: e = json.loads(rec['json_str']) if e['event_name'] == 'heal_result': saves.append(e['data']['saved_player_id']) if len(saves) == 2: break self.assertEqual(saves, ['Player 5', 'Player 5']) self.assertTrue(all('Player 4' not in choices and 'Player 5' in choices for choices in offered)) def test_real_engine_all_abstain_randomly_eliminates(self): env = make('werewolf', configuration={'seed': 24}) env.reset(7) while not env.done: actions = [None] * 7 for i, state in enumerate(env.state): if state.status == 'ACTIVE': r = state.observation.raw_observation actions[i] = {'action_type': 'NoOpAction', 'kwargs': {'actor_id': r.player_id, 'day': r.day, 'phase': r.game_state_phase}} env.step(actions) story = transcript(env.toJSON(), ['abstain'] * 7) tallies = [e for e in story['events'] if e['kind'] == 'tally'] self.assertTrue(tallies) self.assertTrue(all(e['all_abstained'] and e['exiled'] is not None for e in tallies)) def test_20_real_games_end_without_errors(self): roles = set() for seed in range(20): random.seed(seed) env = make('werewolf', configuration=configuration(seed, rng=random.Random(1000 + seed))) env.run([RuleAgent() for _ in range(7)]) self.assertTrue(env.done) self.assertIn(env.info['GAME_END']['winner_team'], ['Werewolves', 'Villagers']) self.assertFalse(env.info['GAME_END']['terminated_with_agent_error']) roles.update(str(p.role.name) for p in env.game_state.players) for step in env.info['MODERATOR_OBSERVATION']: for rec in step: e = json.loads(rec['json_str']) self.assertNotEqual(e['event_name'], 'error', e) self.assertEqual(roles, set(ROLES)) def test_actual_child_process_gets_redacted_config(self): with tempfile.TemporaryDirectory(prefix='werewolf-child-') as tmp: Path(tmp, 'main.py').write_text('def agent(obs, config):\n return config\n') child = seat(tmp, limit=3) try: answer = child({}, {'seed': 12, 'agents': [{'role': 'Werewolf'}], 'actTimeout': 3}) self.assertEqual(answer, {'actTimeout': 3}) finally: child.child.close() child.child.p.wait(timeout=5) child.child.p.stdin.close() child.child.p.stdout.close() def test_live_room_requests_actions_privacy_and_replay(self): with tempfile.TemporaryDirectory(prefix='werewolf-room-') as tmp: conf = {'seed': 7, 'limit': 3, 'seats': {str(i): {'kind': 'human', 'name': f'test-{i}'} for i in range(7)}} env = {**os.environ, 'SS_ROOM': json.dumps(conf), 'SS_INPUT': str(HERE.parent)} with open(Path(tmp, 'stderr.log'), 'w') as errors: proc = subprocess.Popen([sys.executable, '-s', str(HERE / 'room.py')], cwd=tmp, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=errors, text=True) deadline = threading.Timer(20, proc.kill) deadline.start() policies = [RuleAgent() for _ in range(7)] roles, requests, ended = {}, set(), False try: for line in proc.stdout: if not line.startswith('@shoal '): continue msg = json.loads(line[7:]) e = msg.get('event') if e: self.assertNotIn(e['kind'], ['inspect', 'heal', 'wolf_vote', 'wolf_pick', 'saved', 'role']) self.assertNotIn('why', e) if msg.get('obs', {}).get('kind') == 'role': roles[str(msg['to'])] = msg['obs']['role'] if 'legal' in msg: raw = msg['obs']['raw'] requests.add(msg['obs']['request']) self.assertEqual(raw['role'], roles[str(msg['to'])]) if msg['obs']['request'] == 'vote': self.assertIn({'type': 'vote', 'target': str(msg['to'])}, msg['legal']) if msg['obs']['request'] in ('vote', 'kill'): self.assertIn({'type': msg['obs']['request'], 'target': None}, msg['legal']) action = policies[int(msg['to'])]({'raw_observation': raw}) proc.stdin.write(json.dumps({'act': {'seat': msg['to'], 'version': msg['version'], 'action': action}}) + '\n') proc.stdin.flush() if 'end' in msg: ended = True self.assertEqual(proc.wait(timeout=5), 0) self.assertTrue(ended) self.assertEqual(requests, {'chat', 'vote', 'kill', 'heal', 'inspect'}) self.assertTrue(Path(tmp, 'replays/matches.json').is_file()) self.assertNotIn('out of time', Path(tmp, 'stderr.log').read_text()) finally: deadline.cancel() if proc.poll() is None: proc.kill() proc.wait() proc.stdin.close() proc.stdout.close() if __name__ == '__main__': unittest.main(verbosity=2)