From 3c763676faf45c1e51c86c7e8e8ef2bc14ea47e3 Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 15:17:31 +0100 Subject: [PATCH 1/8] fix: update makefile to use new --ports option --- Makefile | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/Makefile b/Makefile index 4a1c64a..093ad55 100644 --- a/Makefile +++ b/Makefile @@ -22,6 +22,10 @@ BALATRO_SCRIPT := ./balatro.sh # Test ports for parallel testing TEST_PORTS := 12346 12347 12348 12349 +# Helper variables for comma-separated port list +comma := , +space := $(subst ,, ) + help: ## Show this help message @echo "$(BLUE)BalatroBot Development Makefile$(RESET)" @echo "" @@ -71,7 +75,7 @@ test: ## Run tests with single Balatro instance (auto-starts if needed) @echo "$(YELLOW)Running tests...$(RESET)" @if ! $(BALATRO_SCRIPT) --status | grep -q "12346"; then \ echo "Starting Balatro on port 12346..."; \ - $(BALATRO_SCRIPT) --headless --fast -p 12346; \ + $(BALATRO_SCRIPT) --headless --fast --ports 12346; \ sleep 1; \ fi $(PYTEST) @@ -81,7 +85,7 @@ test-parallel: ## Run tests in parallel on 4 instances (auto-starts if needed) @running_count=$$($(BALATRO_SCRIPT) --status | grep -E "($(word 1,$(TEST_PORTS))|$(word 2,$(TEST_PORTS))|$(word 3,$(TEST_PORTS))|$(word 4,$(TEST_PORTS)))" | wc -l); \ if [ "$$running_count" -ne 4 ]; then \ echo "Starting Balatro instances on ports: $(TEST_PORTS)"; \ - $(BALATRO_SCRIPT) --headless --fast -p $(word 1,$(TEST_PORTS)) -p $(word 2,$(TEST_PORTS)) -p $(word 3,$(TEST_PORTS)) -p $(word 4,$(TEST_PORTS)); \ + $(BALATRO_SCRIPT) --headless --fast --ports $(subst $(space),$(comma),$(TEST_PORTS)); \ sleep 1; \ fi $(PYTEST) -n 4 --port $(word 1,$(TEST_PORTS)) --port $(word 2,$(TEST_PORTS)) --port $(word 3,$(TEST_PORTS)) --port $(word 4,$(TEST_PORTS)) tests/lua/ @@ -91,7 +95,7 @@ test-migrate: ## Run replay.py on all JSONL files in tests/runs/ using 4 paralle @running_count=$$($(BALATRO_SCRIPT) --status | grep -E "($(word 1,$(TEST_PORTS))|$(word 2,$(TEST_PORTS))|$(word 3,$(TEST_PORTS))|$(word 4,$(TEST_PORTS)))" | wc -l); \ if [ "$$running_count" -ne 4 ]; then \ echo "Starting Balatro instances on ports: $(TEST_PORTS)"; \ - $(BALATRO_SCRIPT) --headless --fast -p $(word 1,$(TEST_PORTS)) -p $(word 2,$(TEST_PORTS)) -p $(word 3,$(TEST_PORTS)) -p $(word 4,$(TEST_PORTS)); \ + $(BALATRO_SCRIPT) --headless --fast --ports $(subst $(space),$(comma),$(TEST_PORTS)); \ sleep 1; \ fi @jsonl_files=$$(find tests/runs -name "*.jsonl" -not -name "*.skip" | sort); \ From f19c43d9a5f9d53c219e9906e428b0a8edd0c719 Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 15:40:21 +0100 Subject: [PATCH 2/8] refactor: simplify the Makefile --- Makefile | 151 +++++++------------------------------------------------ 1 file changed, 19 insertions(+), 132 deletions(-) diff --git a/Makefile b/Makefile index 093ad55..56234e6 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ .DEFAULT_GOAL := help -.PHONY: help install install-dev lint lint-fix format format-md typecheck quality test test-parallel test-migrate test-teardown docs-serve docs-build docs-clean build clean all dev +.PHONY: help install lint format typecheck quality test all # Colors for output YELLOW := \033[33m @@ -8,152 +8,39 @@ BLUE := \033[34m RED := \033[31m RESET := \033[0m -# Project variables -PYTHON := python3 -UV := uv -PYTEST := pytest -RUFF := ruff -STYLUA := stylua -TYPECHECK := basedpyright -MKDOCS := mkdocs -MDFORMAT := mdformat -BALATRO_SCRIPT := ./balatro.sh - -# Test ports for parallel testing -TEST_PORTS := 12346 12347 12348 12349 - -# Helper variables for comma-separated port list -comma := , -space := $(subst ,, ) - help: ## Show this help message @echo "$(BLUE)BalatroBot Development Makefile$(RESET)" @echo "" @echo "$(YELLOW)Available targets:$(RESET)" @awk 'BEGIN {FS = ":.*?## "} /^[a-zA-Z_-]+:.*?## / {printf " $(GREEN)%-18s$(RESET) %s\n", $$1, $$2}' $(MAKEFILE_LIST) -# Installation targets -install: ## Install package dependencies - @echo "$(YELLOW)Installing dependencies...$(RESET)" - $(UV) sync - -install-dev: ## Install package with development dependencies - @echo "$(YELLOW)Installing development dependencies...$(RESET)" - $(UV) sync --all-extras +install: ## Install balatrobot and all dependencies (including dev) + @echo "$(YELLOW)Installing all dependencies...$(RESET)" + uv sync --all-extras -# Code quality targets lint: ## Run ruff linter (check only) @echo "$(YELLOW)Running ruff linter...$(RESET)" - $(RUFF) check --select I . - $(RUFF) check . + ruff check --fix --select I . + ruff check --fix . -lint-fix: ## Run ruff linter with auto-fixes - @echo "$(YELLOW)Running ruff linter with fixes...$(RESET)" - $(RUFF) check --select I --fix . - $(RUFF) check --fix . - -format: ## Run ruff formatter +format: ## Run ruff and mdformat formatters @echo "$(YELLOW)Running ruff formatter...$(RESET)" - $(RUFF) check --select I --fix . - $(RUFF) format . - @echo "$(YELLOW)Running stylua formatter...$(RESET)" - $(STYLUA) src/lua - -format-md: ## Run markdown formatter - @echo "$(YELLOW)Running markdown formatter...$(RESET)" - $(MDFORMAT) . + ruff check --select I --fix . + ruff format . + @echo "$(YELLOW)Running mdformat formatter...$(RESET)" + mdformat ./docs README.md CLAUDE.md typecheck: ## Run type checker @echo "$(YELLOW)Running type checker...$(RESET)" - $(TYPECHECK) + basedpyright src/balatrobot -quality: lint format typecheck ## Run all code quality checks - @echo "$(GREEN) All quality checks completed$(RESET)" +quality: lint typecheck format ## Run all code quality checks + @echo "$(GREEN)✓ All checks completed$(RESET)" -# Testing targets -test: ## Run tests with single Balatro instance (auto-starts if needed) +test: ## Run tests head-less @echo "$(YELLOW)Running tests...$(RESET)" - @if ! $(BALATRO_SCRIPT) --status | grep -q "12346"; then \ - echo "Starting Balatro on port 12346..."; \ - $(BALATRO_SCRIPT) --headless --fast --ports 12346; \ - sleep 1; \ - fi - $(PYTEST) - -test-parallel: ## Run tests in parallel on 4 instances (auto-starts if needed) - @echo "$(YELLOW)Running parallel tests...$(RESET)" - @running_count=$$($(BALATRO_SCRIPT) --status | grep -E "($(word 1,$(TEST_PORTS))|$(word 2,$(TEST_PORTS))|$(word 3,$(TEST_PORTS))|$(word 4,$(TEST_PORTS)))" | wc -l); \ - if [ "$$running_count" -ne 4 ]; then \ - echo "Starting Balatro instances on ports: $(TEST_PORTS)"; \ - $(BALATRO_SCRIPT) --headless --fast --ports $(subst $(space),$(comma),$(TEST_PORTS)); \ - sleep 1; \ - fi - $(PYTEST) -n 4 --port $(word 1,$(TEST_PORTS)) --port $(word 2,$(TEST_PORTS)) --port $(word 3,$(TEST_PORTS)) --port $(word 4,$(TEST_PORTS)) tests/lua/ - -test-migrate: ## Run replay.py on all JSONL files in tests/runs/ using 4 parallel instances - @echo "$(YELLOW)Running replay migration on tests/runs/ files...$(RESET)" - @running_count=$$($(BALATRO_SCRIPT) --status | grep -E "($(word 1,$(TEST_PORTS))|$(word 2,$(TEST_PORTS))|$(word 3,$(TEST_PORTS))|$(word 4,$(TEST_PORTS)))" | wc -l); \ - if [ "$$running_count" -ne 4 ]; then \ - echo "Starting Balatro instances on ports: $(TEST_PORTS)"; \ - $(BALATRO_SCRIPT) --headless --fast --ports $(subst $(space),$(comma),$(TEST_PORTS)); \ - sleep 1; \ - fi - @jsonl_files=$$(find tests/runs -name "*.jsonl" -not -name "*.skip" | sort); \ - if [ -z "$$jsonl_files" ]; then \ - echo "$(RED)No .jsonl files found in tests/runs/$(RESET)"; \ - exit 1; \ - fi; \ - file_count=$$(echo "$$jsonl_files" | wc -l); \ - echo "Found $$file_count .jsonl files to process"; \ - ports=($(TEST_PORTS)); \ - port_idx=0; \ - for file in $$jsonl_files; do \ - port=$${ports[$$port_idx]}; \ - echo "Processing $$file on port $$port..."; \ - $(PYTHON) bots/replay.py --input "$$file" --port $$port & \ - port_idx=$$((port_idx + 1)); \ - if [ $$port_idx -eq 4 ]; then \ - port_idx=0; \ - fi; \ - done; \ - wait; \ - echo "$(GREEN)✓ All replay migrations completed$(RESET)" - -test-teardown: ## Kill all Balatro instances - @echo "$(YELLOW)Killing all Balatro instances...$(RESET)" - $(BALATRO_SCRIPT) --kill - @echo "$(GREEN) All instances stopped$(RESET)" - -# Documentation targets -docs-serve: ## Serve documentation locally - @echo "$(YELLOW)Starting documentation server...$(RESET)" - $(MKDOCS) serve - -docs-build: ## Build documentation - @echo "$(YELLOW)Building documentation...$(RESET)" - $(MKDOCS) build - -docs-clean: ## Clean built documentation - @echo "$(YELLOW)Cleaning documentation build...$(RESET)" - rm -rf site/ - -# Build targets -build: ## Build package for distribution - @echo "$(YELLOW)Building package...$(RESET)" - $(PYTHON) -m build - -clean: ## Clean build artifacts and caches - @echo "$(YELLOW)Cleaning build artifacts...$(RESET)" - rm -rf build/ dist/ *.egg-info/ - rm -rf .pytest_cache/ .coverage htmlcov/ coverage.xml - rm -rf .ruff_cache/ - find . -type d -name __pycache__ -exec rm -rf {} + - find . -type f -name "*.pyc" -delete - @echo "$(GREEN) Cleanup completed$(RESET)" - -# Convenience targets -dev: format lint typecheck ## Quick development check (no tests) - @echo "$(GREEN) Development checks completed$(RESET)" + ./balatro.sh --fast --headless --ports 12346 + pytest -all: format lint typecheck test ## Complete quality check with tests - @echo "$(GREEN) All checks completed successfully$(RESET)" +all: lint format typecheck test ## Run all code quality checks and tests + @echo "$(GREEN)✓ All checks completed$(RESET)" From f00f73d024835a31a807ddf5beeaca5db5b8276f Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 15:42:41 +0100 Subject: [PATCH 3/8] chore: add stirby as new contributor --- pyproject.toml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 8e2d84f..3ff9abd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,8 @@ description = "A framework for Balatro bot development" readme = "README.md" authors = [ { name = "S1M0N38", email = "bertolottosimone@gmail.com" }, - { name = "giewev", email = "giewev@gmail.com" }, + { name = "stirby" }, + { name = "giewev" }, { name = "besteon" }, { name = "phughesion" }, ] From 042228e2211835b0a55117bf8bff04e4d07247a2 Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 17:39:14 +0100 Subject: [PATCH 4/8] chore: rename old test file to .old.py --- .../lua/{test_protocol_errors.py => test_protocol_errors.old.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/lua/{test_protocol_errors.py => test_protocol_errors.old.py} (100%) diff --git a/tests/lua/test_protocol_errors.py b/tests/lua/test_protocol_errors.old.py similarity index 100% rename from tests/lua/test_protocol_errors.py rename to tests/lua/test_protocol_errors.old.py From e9c6b7b89dbbe80dfba2a928387034d2b6147d6a Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 17:39:32 +0100 Subject: [PATCH 5/8] test(lua): update conftest.py with two simple functions --- tests/lua/conftest.py | 266 ++++++++++++++++++++---------------------- 1 file changed, 126 insertions(+), 140 deletions(-) diff --git a/tests/lua/conftest.py b/tests/lua/conftest.py index a2742f6..7c7c6d0 100644 --- a/tests/lua/conftest.py +++ b/tests/lua/conftest.py @@ -1,165 +1,151 @@ """Lua API test-specific configuration and fixtures.""" import json -import platform -import shutil import socket -from pathlib import Path from typing import Any, Generator import pytest -# Connection settings -HOST = "127.0.0.1" -TIMEOUT: float = 60.0 # timeout for socket operations in seconds BUFFER_SIZE: int = 65536 # 64KB buffer for TCP messages @pytest.fixture -def tcp_client(port: int) -> Generator[socket.socket, None, None]: - """Create and clean up a TCP client socket. +def client( + host: str = "127.0.0.1", + port: int = 12346, + timeout: float = 60, + buffer_size: int = BUFFER_SIZE, +) -> Generator[socket.socket, None, None]: + """Create a TCP socket client connected to Balatro game instance. + + Args: + host: The hostname or IP address of the Balatro game server (default: "127.0.0.1"). + port: The port number the Balatro game server is listening on (default: 12346). + timeout: Socket timeout in seconds for connection and operations (default: 60). + buffer_size: Size of the socket receive buffer (default: 65536, i.e. 64KB). Yields: - Configured TCP socket for testing. + A connected TCP socket for communicating with the game. """ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(TIMEOUT) - # Set socket receive buffer size - sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE) - sock.connect((HOST, port)) + sock.settimeout(timeout) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, buffer_size) + sock.connect((host, port)) yield sock -def send_api_message(sock: socket.socket, name: str, arguments: dict) -> None: - """Send a properly formatted JSON API message. - - Args: - sock: Socket to send through. - name: Function name to call. - arguments: Arguments dictionary for the function. - """ - message = {"name": name, "arguments": arguments} - sock.send(json.dumps(message).encode() + b"\n") - - -def receive_api_message(sock: socket.socket) -> dict[str, Any]: - """Receive a properly formatted JSON API message from the socket. - - Args: - sock: Socket to receive from. - - Returns: - Received message as a dictionary. - """ - data = sock.recv(BUFFER_SIZE) - return json.loads(data.decode().strip()) - - -def send_and_receive_api_message( - sock: socket.socket, name: str, arguments: dict +def api( + client: socket.socket, + name: str, + arguments: dict = {}, ) -> dict[str, Any]: - """Send a properly formatted JSON API message and receive the response. + """Send an API call to the Balatro game and get the response. Args: - sock: Socket to send through. - name: Function name to call. - arguments: Arguments dictionary for the function. + sock: The TCP socket connected to the game. + name: The name of the API function to call. + arguments: Dictionary of arguments to pass to the API function (default: {}). Returns: - The game state after the message is sent and received. - """ - send_api_message(sock, name, arguments) - game_state = receive_api_message(sock) - return game_state - - -def assert_error_response( - response, - expected_error_text, - expected_context_keys=None, - expected_error_code=None, -): + The game state response as a dictionary. """ - Helper function to assert the format and content of an error response. - - Args: - response (dict): The response dictionary to validate. Must contain at least - the keys "error", "state", and "error_code". - expected_error_text (str): The expected error message text to check within - the "error" field of the response. - expected_context_keys (list, optional): A list of keys expected to be present - in the "context" field of the response, if the "context" field exists. - expected_error_code (str, optional): The expected error code to check within - the "error_code" field of the response. - - Raises: - AssertionError: If the response does not match the expected format or content. - """ - assert isinstance(response, dict) - assert "error" in response - assert "state" in response - assert "error_code" in response - assert expected_error_text in response["error"] - if expected_error_code: - assert response["error_code"] == expected_error_code - if expected_context_keys: - assert "context" in response - for key in expected_context_keys: - assert key in response["context"] - - -def prepare_checkpoint(sock: socket.socket, checkpoint_path: Path) -> dict[str, Any]: - """Prepare a checkpoint file for loading and load it into the game. - - This function copies a checkpoint file to Love2D's save directory and loads it - directly without requiring a game restart. - - Args: - sock: Socket connection to the game. - checkpoint_path: Path to the checkpoint .jkr file to load. - - Returns: - Game state after loading the checkpoint. - - Raises: - FileNotFoundError: If checkpoint file doesn't exist. - RuntimeError: If loading the checkpoint fails. - """ - if not checkpoint_path.exists(): - raise FileNotFoundError(f"Checkpoint file not found: {checkpoint_path}") - - # First, get the save directory from the game - game_state = send_and_receive_api_message(sock, "get_save_info", {}) - - # Determine the Love2D save directory - # On Linux with Steam, convert Windows paths - - save_dir_str = game_state["save_directory"] - if platform.system() == "Linux" and save_dir_str.startswith("C:"): - # Replace C: with Linux Steam Proton prefix - linux_prefix = ( - Path.home() / ".steam/steam/steamapps/compatdata/2379780/pfx/drive_c" - ) - save_dir_str = str(linux_prefix) + "/" + save_dir_str[3:] - - save_dir = Path(save_dir_str) - - # Copy checkpoint to a test profile in Love2D save directory - test_profile = "test_checkpoint" - test_dir = save_dir / test_profile - test_dir.mkdir(parents=True, exist_ok=True) - - dest_path = test_dir / "save.jkr" - shutil.copy2(checkpoint_path, dest_path) - - # Load the save using the new load_save API function - love2d_path = f"{test_profile}/save.jkr" - game_state = send_and_receive_api_message( - sock, "load_save", {"save_path": love2d_path} - ) - - # Check for errors - if "error" in game_state: - raise RuntimeError(f"Failed to load checkpoint: {game_state['error']}") - - return game_state + payload = {"name": name, "arguments": arguments} + client.send(json.dumps(payload).encode() + b"\n") + response = client.recv(BUFFER_SIZE) + gamestate = json.loads(response.decode().strip()) + return gamestate + + +# import platform +# from pathlib import Path +# import shutil +# def assert_error_response( +# response, +# expected_error_text, +# expected_context_keys=None, +# expected_error_code=None, +# ): +# """ +# Helper function to assert the format and content of an error response. +# +# Args: +# response (dict): The response dictionary to validate. Must contain at least +# the keys "error", "state", and "error_code". +# expected_error_text (str): The expected error message text to check within +# the "error" field of the response. +# expected_context_keys (list, optional): A list of keys expected to be present +# in the "context" field of the response, if the "context" field exists. +# expected_error_code (str, optional): The expected error code to check within +# the "error_code" field of the response. +# +# Raises: +# AssertionError: If the response does not match the expected format or content. +# """ +# assert isinstance(response, dict) +# assert "error" in response +# assert "state" in response +# assert "error_code" in response +# assert expected_error_text in response["error"] +# if expected_error_code: +# assert response["error_code"] == expected_error_code +# if expected_context_keys: +# assert "context" in response +# for key in expected_context_keys: +# assert key in response["context"] +# +# +# def prepare_checkpoint(sock: socket.socket, checkpoint_path: Path) -> dict[str, Any]: +# """Prepare a checkpoint file for loading and load it into the game. +# +# This function copies a checkpoint file to Love2D's save directory and loads it +# directly without requiring a game restart. +# +# Args: +# sock: Socket connection to the game. +# checkpoint_path: Path to the checkpoint .jkr file to load. +# +# Returns: +# Game state after loading the checkpoint. +# +# Raises: +# FileNotFoundError: If checkpoint file doesn't exist. +# RuntimeError: If loading the checkpoint fails. +# """ +# if not checkpoint_path.exists(): +# raise FileNotFoundError(f"Checkpoint file not found: {checkpoint_path}") +# +# # First, get the save directory from the game +# game_state = send_and_receive_api_message(sock, "get_save_info", {}) +# +# # Determine the Love2D save directory +# # On Linux with Steam, convert Windows paths +# +# save_dir_str = game_state["save_directory"] +# if platform.system() == "Linux" and save_dir_str.startswith("C:"): +# # Replace C: with Linux Steam Proton prefix +# linux_prefix = ( +# Path.home() / ".steam/steam/steamapps/compatdata/2379780/pfx/drive_c" +# ) +# save_dir_str = str(linux_prefix) + "/" + save_dir_str[3:] +# +# save_dir = Path(save_dir_str) +# +# # Copy checkpoint to a test profile in Love2D save directory +# test_profile = "test_checkpoint" +# test_dir = save_dir / test_profile +# test_dir.mkdir(parents=True, exist_ok=True) +# +# dest_path = test_dir / "save.jkr" +# shutil.copy2(checkpoint_path, dest_path) +# +# # Load the save using the new load_save API function +# love2d_path = f"{test_profile}/save.jkr" +# game_state = send_and_receive_api_message( +# sock, "load_save", {"save_path": love2d_path} +# ) +# +# # Check for errors +# if "error" in game_state: +# raise RuntimeError(f"Failed to load checkpoint: {game_state['error']}") +# +# return game_state From 9de8ac00ee1598c2ea4440c822facde421602944 Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 17:40:44 +0100 Subject: [PATCH 6/8] test(lua): add test for connection.lua --- tests/lua/test_connection.py | 135 ++++++++--------------------------- 1 file changed, 28 insertions(+), 107 deletions(-) diff --git a/tests/lua/test_connection.py b/tests/lua/test_connection.py index 18c684b..b75c49e 100644 --- a/tests/lua/test_connection.py +++ b/tests/lua/test_connection.py @@ -1,119 +1,40 @@ -"""Tests for BalatroBot TCP API connection and protocol handling.""" +"""Tests for BalatroBot TCP API connection. + +This module tests the core TCP communication layer between the Python bot +and the Lua game mod, ensuring proper connection handling. + +Connection Tests: +- test_basic_connection: Verify TCP connection and basic game state retrieval +- test_rapid_messages: Test multiple rapid API calls without connection drops +- test_connection_wrong_port: Ensure connection refusal on wrong port +""" import json import socket import pytest -from .conftest import HOST, assert_error_response, receive_api_message, send_api_message +from .conftest import BUFFER_SIZE, api -def test_basic_connection(tcp_client: socket.socket) -> None: +def test_basic_connection(client: socket.socket): """Test basic TCP connection and response.""" - send_api_message(tcp_client, "get_game_state", {}) + gamestate = api(client, "get_game_state") + assert isinstance(gamestate, dict) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - -def test_rapid_messages(tcp_client: socket.socket) -> None: +def test_rapid_messages(client: socket.socket): """Test rapid succession of get_game_state messages.""" - responses = [] - - for _ in range(3): - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - responses.append(game_state) - - assert all(isinstance(resp, dict) for resp in responses) - assert len(responses) == 3 - - -def test_connection_timeout() -> None: - """Test behavior when no server is listening.""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(0.2) - - with pytest.raises((socket.timeout, ConnectionRefusedError)): - sock.connect((HOST, 12345)) # Unused port - - -def test_invalid_json_message(tcp_client: socket.socket) -> None: - """Test that invalid JSON messages return error responses.""" - # Send invalid JSON - tcp_client.send(b"invalid json\n") - - # Should receive error response for invalid JSON - error_response = receive_api_message(tcp_client) - assert_error_response(error_response, "Invalid JSON") - - # Verify server is still responsive - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - - -def test_missing_name_field(tcp_client: socket.socket) -> None: - """Test message without name field returns error response.""" - message = {"arguments": {}} - tcp_client.send(json.dumps(message).encode() + b"\n") - - # Should receive error response for missing name field - error_response = receive_api_message(tcp_client) - assert_error_response(error_response, "Message must contain a name") - - # Verify server is still responsive - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - - -def test_missing_arguments_field(tcp_client: socket.socket) -> None: - """Test message without arguments field returns error response.""" - message = {"name": "get_game_state"} - tcp_client.send(json.dumps(message).encode() + b"\n") - - # Should receive error response for missing arguments field - error_response = receive_api_message(tcp_client) - assert_error_response(error_response, "Message must contain arguments") - - # Verify server is still responsive - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - - -def test_unknown_message(tcp_client: socket.socket) -> None: - """Test that unknown messages return error responses.""" - # Send unknown message - send_api_message(tcp_client, "unknown_function", {}) - - # Should receive error response for unknown function - error_response = receive_api_message(tcp_client) - assert_error_response(error_response, "Unknown function name", ["name"]) - - # Verify server is still responsive - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - - -def test_large_message_handling(tcp_client: socket.socket) -> None: - """Test handling of large messages within TCP limits.""" - # Create a large but valid message - large_args = {"data": "x" * 1000} # 1KB of data - send_api_message(tcp_client, "get_game_state", large_args) - - # Should still get a response - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) - - -def test_empty_message(tcp_client: socket.socket) -> None: - """Test sending an empty message.""" - tcp_client.send(b"\n") - - # Verify server is still responsive - send_api_message(tcp_client, "get_game_state", {}) - game_state = receive_api_message(tcp_client) - assert isinstance(game_state, dict) + NUM_MESSAGES = 5 + gamestates = [api(client, "get_game_state") for _ in range(NUM_MESSAGES)] + assert all(isinstance(gamestate, dict) for gamestate in gamestates) + assert len(gamestates) == NUM_MESSAGES + + +def test_connection_wrong_port(): + """Test behavior when wrong port is specified.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client: + client.settimeout(0.2) + client.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE) + with pytest.raises(ConnectionRefusedError): + client.connect(("127.0.0.1", 12345)) From adcbb7d4e274eb3592493ebee83be071da9aa725 Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 17:40:55 +0100 Subject: [PATCH 7/8] test(lua): add tests for protocol handling --- tests/lua/test_protocol.py | 66 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 tests/lua/test_protocol.py diff --git a/tests/lua/test_protocol.py b/tests/lua/test_protocol.py new file mode 100644 index 0000000..b177de1 --- /dev/null +++ b/tests/lua/test_protocol.py @@ -0,0 +1,66 @@ +"""Tests for BalatroBot protocol handling. + +This module tests the core TCP communication layer between the Python bot and +the Lua game mod, ensuring proper message protocol, and error response +validation. + +Protocol Payload Tests: +- test_empty_payload: Verify error response for empty messages (E001) +- test_missing_name: Test error when API call name is missing (E002) +- test_unknown_name: Test error for unknown API call names (E004) +- test_missing_arguments: Test error when arguments field is missing (E003) +- test_malformed_arguments: Test error for malformed JSON arguments (E001) +- test_invalid_arguments: Test error for invalid argument types (E005) +""" + +import json +import socket + +from .conftest import BUFFER_SIZE, api + + +def test_empty_payload(client: socket.socket): + """Test sending an empty payload.""" + client.send(b"\n") + response = client.recv(BUFFER_SIZE) + gamestate = json.loads(response.decode().strip()) + assert gamestate["error_code"] == "E001" # Invalid JSON + + +def test_missing_name(client: socket.socket): + """Test message without name field returns error response.""" + payload = {"arguments": {}} + client.send(json.dumps(payload).encode() + b"\n") + response = client.recv(BUFFER_SIZE) + gamestate = json.loads(response.decode().strip()) + assert gamestate["error_code"] == "E002" # MISSING NAME + + +def test_unknown_name(client: socket.socket): + """Test message with unknown name field returns error response.""" + gamestate = api(client, "unknown") + assert gamestate["error_code"] == "E004" # UNKNOWN NAME + + +def test_missing_arguments(client: socket.socket): + """Test message without name field returns error response.""" + payload = {"name": "get_game_state"} + client.send(json.dumps(payload).encode() + b"\n") + response = client.recv(BUFFER_SIZE) + gamestate = json.loads(response.decode().strip()) + assert gamestate["error_code"] == "E003" # MISSING ARGUMENTS + + +def test_malformed_arguments(client: socket.socket): + """Test message with malformed arguments returns error response.""" + payload = '{"name": "start_run", "arguments": {this is not valid JSON} }' + client.send(payload.encode() + b"\n") + response = client.recv(BUFFER_SIZE) + gamestate = json.loads(response.decode().strip()) + assert gamestate["error_code"] == "E001" # Invalid JSON + + +def test_invalid_arguments(client: socket.socket): + """Test that invalid JSON messages return error responses.""" + gamestate = api(client, "start_run", arguments="this is not a dict") # type: ignore + assert gamestate["error_code"] == "E005" # Invalid Arguments From 56cf26f892cfa0ab5114f5f8cfab9bc996ffc79d Mon Sep 17 00:00:00 2001 From: S1M0N38 Date: Thu, 6 Nov 2025 17:41:32 +0100 Subject: [PATCH 8/8] chore: remove pytest config from pyproject.toml --- pyproject.toml | 3 --- 1 file changed, 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 3ff9abd..8933df4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,9 +38,6 @@ lint.task-tags = ["FIX", "TODO", "HACK", "WARN", "PERF", "NOTE", "TEST"] [tool.pyright] typeCheckingMode = "basic" -[tool.pytest.ini_options] -addopts = "--cov=src/balatrobot --cov-report=term-missing --cov-report=html --cov-report=xml" - [dependency-groups] dev = [ "basedpyright>=1.29.5",