Echo-Tavern/tests/test_rooms.py

from __future__ import annotations


def _login(client, username: str) -> None:
    client.post('/api/auth/register', json={'username': username, 'password': 'secret12'})


def test_create_room_and_join_room(client):
    _login(client, 'alice_room')
    create_resp = client.post('/api/rooms/create', json={})
    assert create_resp.status_code == 200
    create_data = create_resp.get_json()
    assert create_data['ok'] is True

    room_id = create_data['room_id']
    host_player_id = create_data['player_id']
    assert len(room_id) == 6

    client.post('/api/auth/logout', json={})
    _login(client, 'bob_room')
    join_resp = client.post(
        '/api/rooms/join',
        json={'room_id': room_id},
    )
    assert join_resp.status_code == 200
    join_data = join_resp.get_json()
    assert join_data['ok'] is True
    assert join_data['player_id'] != host_player_id


def test_join_room_reconnect_with_cached_player_id(client):
    _login(client, 'host_reconnect')
    create_data = client.post('/api/rooms/create', json={}).get_json()
    room_id = create_data['room_id']

    client.post('/api/auth/logout', json={})
    _login(client, 'member_reconnect')
    join_data = client.post(
        '/api/rooms/join',
        json={'room_id': room_id},
    ).get_json()

    reconnect = client.post(
        '/api/rooms/join',
        json={'room_id': room_id, 'player_id': join_data['player_id']},
    )
    assert reconnect.status_code == 200
    reconnect_data = reconnect.get_json()
    assert reconnect_data['player_id'] == join_data['player_id']
    assert reconnect_data['player_name'] == 'member_reconnect'


def test_join_room_missing_room(client):
    _login(client, 'missing_room_user')
    resp = client.post(
        '/api/rooms/join',
        json={'room_id': 'XXXXXX'},
    )
    assert resp.status_code == 400
    data = resp.get_json()
    assert data['ok'] is False
    assert '房间不存在' in data['error']


def test_host_can_delete_room(client):
    _login(client, 'host_delete')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']
    host_player_id = created['player_id']

    deleted = client.delete(f'/api/rooms/{room_id}', json={'player_id': host_player_id})
    assert deleted.status_code == 200
    assert deleted.get_json()['ok'] is True

    snapshot = client.get(f'/api/rooms/{room_id}/snapshot')
    assert snapshot.status_code == 404


def test_non_host_cannot_delete_room(client):
    _login(client, 'host_non_delete')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']

    client.post('/api/auth/logout', json={})
    _login(client, 'guest_non_delete')
    guest = client.post('/api/rooms/join', json={'room_id': room_id}).get_json()

    denied = client.delete(f'/api/rooms/{room_id}', json={'player_id': guest['player_id']})
    assert denied.status_code == 400
    assert '只有房主可以删除房间' in denied.get_json()['error']


def test_lobby_join_requires_login_and_running_allows_spectator(client, app_instance):
    _login(client, 'host_invite')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']
    host_id = created['player_id']

    client.post('/api/auth/logout', json={})
    denied = client.post('/api/rooms/join', json={'room_id': room_id})
    assert denied.status_code == 401

    _login(client, 'mate_invite')
    mate = client.post('/api/rooms/join', json={'room_id': room_id}).get_json()
    client.post('/api/auth/logout', json={})
    _login(client, 'host_invite')
    app, _ = app_instance
    app.extensions['room_manager'].set_ready(room_id, mate['player_id'], True)
    app.extensions['room_manager'].start_game(room_id, host_id)
    client.post('/api/auth/logout', json={})

    spectate = client.post('/api/rooms/join', json={'room_id': room_id})
    payload = spectate.get_json()
    assert spectate.status_code == 200
    assert payload['ok'] is True
    assert payload['is_spectator'] is True


def test_running_room_supports_spectator_page_without_login(client, app_instance):
    _login(client, 'spectator_host')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']
    host_id = created['player_id']
    app, _ = app_instance
    app.extensions['room_manager'].start_game(room_id, host_id)
    client.post('/api/auth/logout', json={})

    page = client.get(f'/room/{room_id}?spectator=1')
    assert page.status_code == 200
    html = page.get_data(as_text=True)
    assert 'data-spectator="1"' in html


def test_my_rooms_api_keeps_ended_status(client, app_instance):
    _login(client, 'ended_room_owner')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']
    host_id = created['player_id']

    app, _ = app_instance
    manager = app.extensions['room_manager']
    persistence = app.extensions['persistence']
    manager.start_game(room_id, host_id)

    room = persistence.get_room(room_id)
    state = dict(room['game_state'])
    state['game_phase'] = 'ended'
    state['ending_state'] = 'custom_end'
    state.setdefault('global_flags', {})
    state['global_flags']['ending_summary'] = '测试结局'
    persistence.update_room_state(room_id, 'ended', state)

    rooms_resp = client.get('/api/me/rooms')
    payload = rooms_resp.get_json()
    assert rooms_resp.status_code == 200
    assert payload['ok'] is True
    row = next((item for item in payload['rooms'] if item['room_id'] == room_id), None)
    assert row is not None
    assert row['status'] == 'ended'


def test_host_can_set_room_public_and_public_list_can_see_it(client):
    _login(client, 'host_public_1')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']
    host_id = created['player_id']

    toggled = client.patch(
        f'/api/rooms/{room_id}/visibility',
        json={'player_id': host_id, 'is_public': True},
    )
    assert toggled.status_code == 200
    data = toggled.get_json()
    assert data['ok'] is True
    assert data['is_public'] is True
    assert data['snapshot']['is_public'] is True

    public_resp = client.get('/api/rooms/public?limit=20')
    assert public_resp.status_code == 200
    public_data = public_resp.get_json()
    assert public_data['ok'] is True
    row = next((item for item in public_data['rooms'] if item['room_id'] == room_id), None)
    assert row is not None
    assert row['status'] == 'lobby'


def test_non_host_cannot_set_room_public(client):
    _login(client, 'host_public_2')
    created = client.post('/api/rooms/create', json={}).get_json()
    room_id = created['room_id']

    client.post('/api/auth/logout', json={})
    _login(client, 'guest_public_2')
    guest = client.post('/api/rooms/join', json={'room_id': room_id}).get_json()

    denied = client.patch(
        f'/api/rooms/{room_id}/visibility',
        json={'player_id': guest['player_id'], 'is_public': True},
    )
    assert denied.status_code == 400
    payload = denied.get_json()
    assert payload['ok'] is False
    assert '只有房主可以设置房间公开状态' in payload['error']


def test_leaderboard_api_returns_ranked_players(client, app_instance):
    app, _ = app_instance
    manager = app.extensions['room_manager']
    persistence = app.extensions['persistence']

    _login(client, 'leader_alpha')
    alpha_1 = client.post('/api/rooms/create', json={}).get_json()
    manager.start_game(alpha_1['room_id'], alpha_1['player_id'])
    room = persistence.get_room(alpha_1['room_id'])
    state = dict(room['game_state'])
    state['game_phase'] = 'ended'
    state['ending_state'] = 'escaped'
    state['round_no'] = 6
    persistence.update_room_state(alpha_1['room_id'], 'ended', state)

    alpha_2 = client.post('/api/rooms/create', json={}).get_json()
    manager.start_game(alpha_2['room_id'], alpha_2['player_id'])
    room = persistence.get_room(alpha_2['room_id'])
    state = dict(room['game_state'])
    state['game_phase'] = 'ended'
    state['ending_state'] = 'victory'
    state['round_no'] = 8
    persistence.update_room_state(alpha_2['room_id'], 'ended', state)

    client.post('/api/auth/logout', json={})
    _login(client, 'leader_bravo')
    bravo_1 = client.post('/api/rooms/create', json={}).get_json()
    manager.start_game(bravo_1['room_id'], bravo_1['player_id'])
    room = persistence.get_room(bravo_1['room_id'])
    state = dict(room['game_state'])
    state['game_phase'] = 'ended'
    state['ending_state'] = 'trapped'
    state['round_no'] = 9
    persistence.update_room_state(bravo_1['room_id'], 'ended', state)

    leaderboard_resp = client.get('/api/leaderboard?limit=10')
    assert leaderboard_resp.status_code == 200
    payload = leaderboard_resp.get_json()
    assert payload['ok'] is True

    rows = payload['leaderboard']
    assert len(rows) >= 2
    assert rows[0]['username'] == 'leader_alpha'
    assert rows[0]['wins'] == 2
    assert rows[0]['played_games'] == 2
    assert rows[0]['rank'] == 1

    bravo = next((row for row in rows if row['username'] == 'leader_bravo'), None)
    assert bravo is not None
    assert bravo['wins'] == 0
    assert bravo['played_games'] == 1