test.py

import json
import os
import psycopg
import socketio
import pytest
import requests

from dotenv import load_dotenv

load_dotenv()

API_URL = os.getenv("API_URL")
SOCKET_URL = os.getenv("SOCKET_URL")
POSTGRES_URL = os.getenv("POSTGRES_URL")


## Testing the API


def clean_up_games():
    with psycopg.connect(POSTGRES_URL) as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM games")
            conn.commit()


def create_game(players_needed: int = 2):
    response = requests.get(f"{API_URL}/create-game?players_needed={players_needed}")
    assert response.status_code == 200

    data = response.json()
    assert "id" in data

    return data["id"]


def test_create_game():
    id = create_game()

    # check if there is a row in the db
    with psycopg.connect(POSTGRES_URL) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id FROM games WHERE id = %s", (id,))
            conn.commit()

            assert cur.fetchone() is not None


def test_get_game():
    PLAYERS_NEEDED = 2

    id = create_game(PLAYERS_NEEDED)

    response = requests.get(f"{API_URL}/get-game/{id}")
    assert response.status_code == 200

    data = response.json()

    assert "status" in data
    assert "playersNeeded" in data
    assert "maxTimeS" in data
    assert "code" in data

    assert data["status"] == 0
    assert data["playersNeeded"] == PLAYERS_NEEDED
    assert type(data["code"]) == str and len(data["code"]) > 0


def test_list_games():
    create_game()

    response = requests.get(f"{API_URL}/games-list?amount=5")
    assert response.status_code == 200

    data = response.json()
    assert type(data) == list
    assert len(data) > 0

    for game in data:
        assert "id" in game
        assert "playersNeeded" in game
        assert "maxTimeS" in game


def test_connect_to_socket():
    sio = socketio.Client()
    sio.connect(SOCKET_URL)
    sio.disconnect()


### Testing the worker


def two_players_join_game(id):
    # player 1: join game
    sio = socketio.Client()
    sio.connect(SOCKET_URL)

    sio.emit("join_game", {"gameId": id})

    # player 2: join game
    sio2 = socketio.Client()
    sio2.connect(SOCKET_URL)

    sio2.emit("join_game", {"gameId": id})

    return sio, sio2


def test_join_game():
    id = create_game()
    sio, sio2 = two_players_join_game(id)

    # player 1 should get 'players_joined' event (with player 2)
    # player 2 should get 'players_joined' event (with player 1)

    events_1 = 0
    events_2 = 0
    player_1_in_data = 0
    player_2_in_data = 0

    @sio.on("players_joined")
    def on_players_joined(msg):
        nonlocal events_1
        events_1 += 1

        data = json.loads(msg)

        nonlocal player_2_in_data
        player_2_in_data |= sio2.get_sid() in data["playerIds"]

    @sio2.on("players_joined")
    def on_players_joined(msg):
        nonlocal events_2
        events_2 += 1

        data = json.loads(msg)

        nonlocal player_1_in_data
        player_1_in_data |= sio.get_sid() in data["playerIds"]

    sio.sleep(1)

    assert events_1 >= 1
    assert events_2 >= 1
    assert player_1_in_data
    assert player_2_in_data

    sio.disconnect()
    sio2.disconnect()


def test_update_progress():
    id = create_game()
    sio, sio2 = two_players_join_game(id)

    sio.sleep(1)

    # player 1: update progress
    sio.emit("update_progress", {"progress": 5})

    # player 2 should get 'updated_progress' event
    progress = 0

    @sio2.on("updated_progress")
    def on_updated_progress(msg):
        nonlocal progress

        data = json.loads(msg)

        if data["playerId"] == sio.get_sid():
            progress = data["progress"]

    sio.sleep(1)

    assert progress == 5

    sio.disconnect()
    sio2.disconnect()


def test_finish_typing():
    id = create_game()
    sio, sio2 = two_players_join_game(id)

    sio.sleep(1)

    # player 1: finish typing
    sio.emit("finished_typing")

    # player 2 should get 'player_finished_typing' event
    player_finished = False
    place = -1

    @sio2.on("player_finished")
    def on_player_finished(msg):
        nonlocal player_finished
        nonlocal place

        player_finished = True

        data = json.loads(msg)
        place = data["place"]

    sio.sleep(1)

    assert player_finished
    assert place == 1

    sio.disconnect()
    sio2.disconnect()


def test_end_game():
    id = create_game()
    sio, sio2 = two_players_join_game(id)

    sio.sleep(1)

    # player 1 & 2: finish typing
    sio.emit("finished_typing")
    sio.sleep(1)
    sio2.emit("finished_typing")

    # player 1 & 2 should get 'end_game' event
    game_ended = 0

    def on_end_game():
        nonlocal game_ended
        game_ended += 1

    sio.on("end_game", on_end_game)
    sio2.on("end_game", on_end_game)

    sio.sleep(1)

    assert game_ended == 2

    # player 1 & 2: disconnect
    sio.disconnect()
    sio2.disconnect()

    sio.sleep(1)

    # check if the game has been deleted
    with psycopg.connect(POSTGRES_URL) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id FROM games WHERE id = %s", (id,))
            conn.commit()

            assert cur.fetchone() is None

## Cleanup

@pytest.fixture(autouse=True)
def run_around_tests():
    yield
    clean_up_games()