factorio/factorio-learning-environment
1
0
Fork 0
mirror of https://github.com/JackHopkins/factorio-learning-environment.git synced 2025-09-06 13:23:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Faster ci cd (#311)
Some checks failed
Lint and Format / lint (push) Has been cancelled

Browse Source 
* sessions based

* try out caching + no sleep

* update fixture usage

* better reset usge

* state less on tech, probably breaking change

* better fixtures + decouple resets

* use pytest-xdist w 2 servers

* using diff grouping for dep

* formatting

* formatting

* caching for image

* formatting

* formatting

* use uv

* use uv caching

* remove docker caching (its slower)

* how about 4 workers?

* no redundant resets

* parameterize

* change names

* update all_technologies_researched usage

change log:

- used uv and cache dependencies
- used 2 factorio headless server instances
- added pytest-xdist & used 2 pytest workers
- parametrized the slowest test -- `test_sleep.py` so as to balance it across workers
- clarified resets in `instance.py` so separate instances arent needed for research testing
- better fixture usage, with autouse reset
- added configure_game callback for per test file setup of inventories & research state.
- updated task abc all_technologies_researched usage, its now a param for reset
- using 4 workers instead of 2, can probably double it again lol
- pytest parameterized a slow test
- fixed redundant reset in conftest

final speedup: 9m 4s -> 1m, ≈9.07× faster
This commit is contained in:
Harshit Sharma
2025-08-21 17:31:28 +05:30
committed by GitHub
parent 2ae77b49cb
commit 8143457e55
33 changed files with 440 additions and 488 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
.github/workflows/factorio-test.yml vendored
View File

@@ -14,10 +14,12 @@ jobs:
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
- name: Set up uv and Python
uses: astral-sh/setup-uv@v5
with:
python-version: '3.11'
enable-cache: true
cache-dependency-glob: "**/pyproject.toml"
- name: Verify scenario exists
run: |
@@ -29,31 +31,10 @@ jobs:
echo "✓ Found scenario directory"
ls -la ./fle/cluster/scenarios/default_lab_scenario
- name: Start Factorio server
- name: Start 4 Factorio servers
run: |
cd fle/cluster
bash run-envs.sh start -n 1
- name: Wait for server to start
run: |
echo "Waiting for Factorio server to initialize..."
sleep 15
echo "Waiting for RCON port 27000 to be ready..."
for i in {1..30}; do
if nc -z localhost 27000 2>/dev/null; then
echo "✓ RCON port 27000 is open!"
break
fi
echo "Waiting... ($i/30)"
sleep 2
done
if ! nc -z localhost 27000 2>/dev/null; then
echo "❌ RCON port 27000 never became available"
exit 1
fi
bash run-envs.sh start -n 4
- name: Check server status
run: |
@@ -63,9 +44,7 @@ jobs:
- name: Install Python dependencies
run: |
python -m pip install --upgrade pip
# Install the package with dev dependencies for testing
pip install -e ".[dev,agents,cluster,eval,all,mcp,env]"
uv sync --all-extras --dev
- name: Debug container IPs and ports
run: |
@@ -99,13 +78,11 @@ jobs:
fi
done
- name: Run Python tests with debugging
- name: Run Python tests (4 workers)
run: |
# Run tests with verbose output and show print statements
python -m pytest -v -s --tb=short tests/actions/
uv run pytest -n 4 --dist=load -v -s --tb=short tests/actions/
env:
FACTORIO_HOST: localhost
FACTORIO_RCON_PORT: 27000
PYTHONUNBUFFERED: 1
- name: Show test logs on failure

49
fle/env/instance.py vendored
View File

@@ -108,7 +108,6 @@ class FactorioInstance:
self.persistent_vars = {}
self.tcp_port = tcp_port
self.rcon_client, self.address = self.connect_to_server(address, tcp_port)
self.all_technologies_researched = all_technologies_researched
self.fast = fast
self._speed = 1
self._ticks_elapsed = 0
@@ -134,7 +133,7 @@ class FactorioInstance:
if inventory is None:
inventory = {}
self.initial_inventory = inventory
self.initialise(fast)
self.initialise(fast, all_technologies_researched)
self.initial_score = 0
try:
self.first_namespace.score()
@@ -148,7 +147,7 @@ class FactorioInstance:
**self.lua_script_manager.tool_scripts,
}
self.setup_tools(self.lua_script_manager)
self.initialise(fast)
self.initialise(fast, all_technologies_researched)
self.initial_score, goal = self.first_namespace.score()
# Register the cleanup method to be called on exit (only once per process)
@@ -173,7 +172,12 @@ class FactorioInstance:
def is_multiagent(self):
return self.num_agents > 1
def reset(self, game_state: Optional[GameState] = None):
def reset(
self,
game_state: Optional[GameState] = None,
reset_position: bool = False,
all_technologies_researched: bool = True,
):
# Reset the namespace (clear variables, functions etc)
assert not game_state or len(game_state.inventories) == self.num_agents, (
"Game state must have the same number of inventories as num_agents"
@@ -185,9 +189,9 @@ class FactorioInstance:
if not game_state:
# Reset the game instance
inventories = [self.initial_inventory] * self.num_agents
self._reset(inventories)
self._reset(inventories, reset_position, all_technologies_researched)
# Reset the technologies
if not self.all_technologies_researched:
if not all_technologies_researched:
self.first_namespace._load_research_state(
ResearchState(
technologies={},
@@ -199,7 +203,11 @@ class FactorioInstance:
)
else:
# Reset the game instance with the correct player's inventory and messages if multiagent
self._reset(game_state.inventories)
self._reset(
game_state.inventories,
reset_position,
all_technologies_researched,
)
# Load entities into the game
self.first_namespace._load_entity_state(
@@ -599,10 +607,15 @@ class FactorioInstance:
# print(lua_response)
return _lua2python(command, lua_response, start=start)
def _reset(self, inventories: List[Dict[str, Any]]):
def _reset(
self,
inventories: List[Dict[str, Any]],
reset_position: bool,
all_technologies_researched: bool,
):
self.begin_transaction()
self.add_command(
"/sc global.alerts = {}; game.reset_game_state(); global.actions.reset_production_stats(); global.actions.regenerate_resources(1)",
"/sc global.alerts = {}; game.reset_game_state(); global.actions.reset_production_stats();",
raw=True,
)
# self.add_command('/sc script.on_nth_tick(nil)', raw=True) # Remove all dangling event handlers
@@ -619,6 +632,12 @@ class FactorioInstance:
self.add_command("/sc global.actions.clear_walking_queue()", raw=True)
for i in range(self.num_agents):
player_index = i + 1
if reset_position:
# Ensure players are returned to a known spawn location between tests
self.add_command(
f"/sc if global.agent_characters and global.agent_characters[{player_index}] then global.agent_characters[{player_index}].teleport{{x=0, y={(i) * 2}}} end",
raw=True,
)
self.add_command(
f"/sc global.actions.clear_entities({player_index})", raw=True
)
@@ -629,11 +648,13 @@ class FactorioInstance:
raw=True,
)
if self.all_technologies_researched:
if all_technologies_researched:
self.add_command(
"/sc global.agent_characters[1].force.research_all_technologies()",
raw=True,
)
else:
self.add_command("/sc global.agent_characters[1].force.reset()", raw=True)
self.add_command("/sc global.elapsed_ticks = 0", raw=True)
self.execute_transaction()
@@ -676,7 +697,7 @@ class FactorioInstance:
def execute_transaction(self) -> Dict[str, Any]:
return self._execute_transaction()
def initialise(self, fast=True):
def initialise(self, fast=True, all_technologies_researched=True):
self.begin_transaction()
self.add_command("/sc global.alerts = {}", raw=True)
self.add_command("/sc global.elapsed_ticks = 0", raw=True)
@@ -704,7 +725,11 @@ class FactorioInstance:
self.lua_script_manager.load_init_into_game(script_name)
inventories = [self.initial_inventory] * self.num_agents
self._reset(inventories)
self._reset(
inventories,
reset_position=False,
all_technologies_researched=all_technologies_researched,
)
self.first_namespace._clear_collision_boxes()
def _create_agent_game_characters(self):

3
fle/eval/tasks/task_abc.py
View File

@@ -61,7 +61,6 @@ class TaskABC:
def setup(self, instance):
"""setup function"""
instance.initial_inventory = self.starting_inventory
instance.all_technologies_researched = self.all_technology_reserached
instance.reset()
instance.reset(all_technologies_researched=self.all_technology_reserached)
self.setup_instance(instance)
self.starting_game_state = GameState.from_instance(instance)

1
pyproject.toml
View File

@@ -63,6 +63,7 @@ dev = [
"isort>=5.12.0",
"rich>=14.0.0",
"questionary>=2.1.0",
"pytest-xdist>=3.8.0",
]
agents = [
"anthropic>=0.49.0",

2
tests/actions/_test_inspect_entities.py
View File

@@ -12,9 +12,7 @@ def game(instance):
"iron-plate": 5,
"iron-ore": 10,
}
instance.reset()
yield instance.namespace
instance.reset()
def test_inspect_entities(game):

7
tests/actions/test_can_place.py
View File

@@ -5,10 +5,8 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(reset_position=False)
def test_can_place(game):
@@ -43,4 +41,3 @@ def test_can_place_over_player_large(game):
game.place_entity(
Prototype.SteamEngine, position=Position(x=0, y=0), direction=Direction.UP
)
pass

8
tests/actions/test_connect_entities_dry_run.py
View File

@@ -5,12 +5,8 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"transport-belt": 12,
}
instance.reset()
yield instance.namespace
def game(configure_game):
return configure_game(inventory={"transport-belt": 12})
def test_dry_run(game):

13
tests/actions/test_craft.py
View File

@@ -4,10 +4,9 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.reset()
instance.set_inventory(
{
def game(configure_game):
return configure_game(
inventory={
"iron-plate": 40,
"iron-gear-wheel": 1,
"electronic-circuit": 3,
@@ -15,8 +14,6 @@ def game(instance):
"copper-plate": 10,
}
)
yield instance.namespace
instance.reset()
def test_fail_to_craft_item(game):
@@ -130,9 +127,7 @@ def test_craft_uncraftable_entity(game):
def test_craft_no_technology(game):
game.instance.all_technologies_researched = False
game.instance.reset()
game.instance.reset(all_technologies_researched=False)
try:
game.craft_item(Prototype.AssemblingMachine1, quantity=1)
except:

19
tests/actions/test_extract_item.py
View File

@@ -5,16 +5,15 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"iron-chest": 1,
"iron-plate": 10,
"assembling-machine-1": 1,
"copper-cable": 3,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"iron-chest": 1,
"iron-plate": 10,
"assembling-machine-1": 1,
"copper-cable": 3,
}
)
def test_extract(game):

43
tests/actions/test_get_entities.py
View File

@@ -5,28 +5,27 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
**instance.initial_inventory,
"coal": 10,
"iron-chest": 1,
"iron-plate": 50,
"iron-ore": 10,
"stone-furnace": 1,
"offshore-pump": 1,
"assembly-machine-1": 1,
"burner-mining-drill": 1,
"lab": 1,
"automation-science-pack": 1,
"gun-turret": 1,
"firearm-magazine": 5,
"transport-belt": 200,
"boiler": 1,
"pipe": 20,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"coal": 10,
"iron-chest": 1,
"iron-plate": 50,
"iron-ore": 10,
"stone-furnace": 1,
"offshore-pump": 1,
"assembly-machine-1": 1,
"burner-mining-drill": 1,
"lab": 1,
"automation-science-pack": 1,
"gun-turret": 1,
"firearm-magazine": 5,
"transport-belt": 200,
"boiler": 1,
"pipe": 20,
},
merge=True,
)
def test_get_stone_furnace(game):

39
tests/actions/test_get_entity.py
View File

@@ -6,26 +6,25 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
**instance.initial_inventory,
"coal": 5,
"iron-chest": 1,
"iron-plate": 50,
"iron-ore": 10,
"stone-furnace": 1,
"assembling-machine-1": 1,
"burner-mining-drill": 1,
"lab": 1,
"automation-science-pack": 1,
"gun-turret": 1,
"firearm-magazine": 5,
"boiler": 1,
"offshore-pump": 1,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"coal": 5,
"iron-chest": 1,
"iron-plate": 50,
"iron-ore": 10,
"stone-furnace": 1,
"assembling-machine-1": 1,
"burner-mining-drill": 1,
"lab": 1,
"automation-science-pack": 1,
"gun-turret": 1,
"firearm-magazine": 5,
"boiler": 1,
"offshore-pump": 1,
},
merge=True,
)
def test_get_offshore_pump(game):

7
tests/actions/test_get_prototype_recipe.py
View File

@@ -4,11 +4,8 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
instance.initial_inventory = {"assembling-machine-1": 1}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(inventory={"assembling-machine-1": 1})
def test_get_recipe(game):

18
tests/actions/test_get_research_progress.py
View File

@@ -1,25 +1,15 @@
import pytest
from fle.env import FactorioInstance
from fle.env.game_types import Technology
from fle.commons.cluster_ips import get_local_container_ips
@pytest.fixture()
def game(instance):
# game.initial_inventory = {'assembling-machine-1': 1}
ips, udp_ports, tcp_ports = get_local_container_ips()
instance = FactorioInstance(
address="localhost",
bounding_box=200,
tcp_port=tcp_ports[-1], # 27019,
def game(configure_game):
return configure_game(
inventory={"assembling-machine-1": 1},
merge=True,
all_technologies_researched=False,
fast=True,
inventory={},
)
instance.reset()
yield instance.namespace
instance.reset()
def test_get_research_progress_automation(game):

6
tests/actions/test_get_resource_patch.py
View File

@@ -6,10 +6,8 @@ from fle.env.game_types import Resource
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(inventory={"iron-chest": 1})
def test_get_resource_patch(game: FactorioInstance):

9
tests/actions/test_harvest_resource.py
View File

@@ -1,16 +1,7 @@
import pytest
from fle.env.entities import Position
from fle.env.game_types import Resource, Prototype
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
# instance.reset()
def test_harvest_resource_with_full_inventory(game):
"""
Find the nearest coal resource patch and harvest 5 coal from it.

33
tests/actions/test_insert_item.py
View File

@@ -5,23 +5,22 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"iron-chest": 1,
"iron-ore": 500,
"copper-ore": 10,
"iron-plate": 1000,
"iron-gear-wheel": 1000,
"coal": 100,
"stone-furnace": 1,
"transport-belt": 10,
"burner-inserter": 1,
"assembling-machine-1": 1,
"solar-panel": 2,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"iron-chest": 1,
"iron-ore": 500,
"copper-ore": 10,
"iron-plate": 1000,
"iron-gear-wheel": 1000,
"coal": 100,
"stone-furnace": 1,
"transport-belt": 10,
"burner-inserter": 1,
"assembling-machine-1": 1,
"solar-panel": 2,
}
)
def test_insert_and_fuel_furnace(game):

20
tests/actions/test_inspect_inventory.py
View File

@@ -5,16 +5,16 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
**instance.initial_inventory,
"coal": 50,
"iron-chest": 1,
"iron-plate": 5,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"coal": 50,
"iron-chest": 1,
"iron-plate": 5,
},
merge=True,
all_technologies_researched=False,
)
def test_inspect_inventory(game):

39
tests/actions/test_move_to.py
View File

@@ -1,15 +1,18 @@
import pytest
from fle.env.entities import Position
from fle.env import FactorioInstance
from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
# instance.reset()
def game(configure_game):
return configure_game(
inventory={
"coal": 50,
"iron-chest": 1,
"iron-plate": 5,
}
)
def test_move_to(game):
@@ -42,29 +45,3 @@ def test_move_to_check_position(game):
# Move to target position
game.move_to(target_pos)
if __name__ == "__main__":
factorio = FactorioInstance(
address="localhost",
bounding_box=200,
tcp_port=27000,
cache_scripts=True,
fast=True,
inventory={
"coal": 50,
"copper-plate": 50,
"iron-plate": 50,
"iron-chest": 2,
"burner-mining-drill": 3,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"stone-furnace": 9,
"transport-belt": 50,
"boiler": 1,
"burner-inserter": 32,
"pipe": 15,
"steam-engine": 1,
"small-electric-pole": 10,
},
)

9
tests/actions/test_nearest.py
View File

@@ -1,16 +1,7 @@
import pytest
from fle.env.entities import Position
from fle.env.game_types import Resource
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def test_nearest_resource(game):
"""
Test distance to the nearest coal resource.

45
tests/actions/test_nearest_buildable.py
View File

@@ -1,53 +1,13 @@
import pytest
from fle.commons.cluster_ips import get_local_container_ips
from fle.env.game_types import Prototype, Resource
from fle.env.entities import Position, BuildingBox, Direction
from fle.env import FactorioInstance
# @pytest.fixture()
# def game(instance):
# instance.reset()
# instance.set_inventory({
# 'wooden-chest': 100,
# 'electric-mining-drill': 10,
# 'steam-engine': 1,
# 'burner-mining-drill': 5
# })
# yield instance.namespace
@pytest.fixture()
def game():
ips, udp_ports, tcp_ports = get_local_container_ips()
instance = FactorioInstance(
address="localhost",
bounding_box=200,
tcp_port=tcp_ports[-1],
cache_scripts=False,
fast=True,
def game(configure_game):
return configure_game(
inventory={
"coal": 50,
"copper-plate": 50,
"iron-plate": 50,
"iron-chest": 2,
"burner-mining-drill": 3,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"stone-furnace": 9,
"transport-belt": 50,
"boiler": 1,
"burner-inserter": 32,
"pipe": 15,
"steam-engine": 1,
"small-electric-pole": 10,
"pumpjack": 1,
},
)
instance.reset()
instance.set_inventory(
{
"wooden-chest": 100,
"electric-mining-drill": 10,
"steam-engine": 1,
@@ -55,7 +15,6 @@ def game():
"pumpjack": 1,
}
)
yield instance.namespace
def test_nearest_buildable_simple(game):

40
tests/actions/test_pickup_entity.py
View File

@@ -5,27 +5,25 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"stone-furnace": 1,
"boiler": 1,
"steam-engine": 1,
"offshore-pump": 4,
"pipe": 100,
"iron-plate": 50,
"copper-plate": 20,
"coal": 50,
"burner-inserter": 50,
"burner-mining-drill": 50,
"transport-belt": 50,
"stone-wall": 100,
"splitter": 4,
"wooden-chest": 1,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"stone-furnace": 1,
"boiler": 1,
"steam-engine": 1,
"offshore-pump": 4,
"pipe": 100,
"iron-plate": 50,
"copper-plate": 20,
"coal": 50,
"burner-inserter": 50,
"burner-mining-drill": 50,
"transport-belt": 50,
"stone-wall": 100,
"splitter": 4,
"wooden-chest": 1,
}
)
def test_pickup_item_full_inventory(game):

40
tests/actions/test_place.py
View File

@@ -5,27 +5,25 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"stone-furnace": 1,
"boiler": 1,
"steam-engine": 1,
"offshore-pump": 4,
"pipe": 100,
"iron-plate": 50,
"copper-plate": 20,
"coal": 50,
"burner-inserter": 50,
"burner-mining-drill": 50,
"transport-belt": 50,
"stone-wall": 100,
"splitter": 4,
"wooden-chest": 1,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"stone-furnace": 1,
"boiler": 1,
"steam-engine": 1,
"offshore-pump": 4,
"pipe": 100,
"iron-plate": 50,
"copper-plate": 20,
"coal": 50,
"burner-inserter": 50,
"burner-mining-drill": 50,
"transport-belt": 50,
"stone-wall": 100,
"splitter": 4,
"wooden-chest": 1,
}
)
def test_place(game):

131
tests/actions/test_place_entity_next_to.py
View File

@@ -6,23 +6,23 @@ from fle.env.game_types import Prototype, Resource
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"boiler": 3,
"transport-belt": 1,
"stone-furnace": 1,
"burner-mining-drill": 1,
"burner-inserter": 5,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"steam-engine": 1,
"pipe": 1,
"offshore-pump": 1,
"wooden-chest": 3,
}
instance.reset()
yield instance.namespace
# instance.reset()
def game(configure_game):
return configure_game(
inventory={
"boiler": 3,
"transport-belt": 1,
"stone-furnace": 1,
"burner-mining-drill": 1,
"burner-inserter": 5,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"steam-engine": 1,
"pipe": 1,
"offshore-pump": 1,
"wooden-chest": 3,
},
persist_inventory=True,
)
@pytest.fixture
@@ -98,66 +98,63 @@ def calculate_expected_position(
)
def test_place_entities_of_different_sizes(game):
entity_pairs = [
@pytest.mark.parametrize(
"ref_proto,placed_proto",
[
(Prototype.Boiler, Prototype.SteamEngine),
(Prototype.ElectricMiningDrill, Prototype.Boiler),
(Prototype.SteamEngine, Prototype.Pipe),
(Prototype.AssemblingMachine1, Prototype.BurnerInserter),
(Prototype.Boiler, Prototype.TransportBelt),
]
],
)
def test_place_entities_of_different_sizes(game, ref_proto, placed_proto):
if ref_proto != Prototype.OffshorePump:
starting_position = game.nearest(Resource.IronOre)
else:
starting_position = game.nearest(Resource.Water)
nearby_position = Position(x=starting_position.x + 1, y=starting_position.y - 1)
game.move_to(nearby_position)
for ref_proto, placed_proto in entity_pairs:
if ref_proto != Prototype.OffshorePump:
starting_position = game.nearest(Resource.IronOre)
else:
starting_position = game.nearest(Resource.Water)
nearby_position = Position(x=starting_position.x + 1, y=starting_position.y - 1)
game.move_to(nearby_position)
for spacing in range(3):
for direction in [
Direction.LEFT,
Direction.DOWN,
Direction.RIGHT,
Direction.UP,
]:
ref_entity = game.place_entity(
ref_proto, direction=Direction.RIGHT, position=starting_position
)
game.move_to(Position(x=starting_position.x + 3, y=starting_position.y - 3))
placed_entity = game.place_entity_next_to(
placed_proto, ref_entity.position, direction, spacing
)
expected_position = calculate_expected_position(
ref_entity.position, direction, spacing, ref_entity, placed_entity
)
assert placed_entity.position.is_close(expected_position, tolerance=1), (
f"Misplacement: {ref_proto.value[0]} -> {placed_proto.value[0]}, "
f"Direction: {direction}, Spacing: {spacing}, "
f"Expected: {expected_position}, Got: {placed_entity.position}"
)
for spacing in range(3):
for direction in [
Direction.LEFT,
Direction.DOWN,
Direction.RIGHT,
Direction.UP,
]:
ref_entity = game.place_entity(
ref_proto, direction=Direction.RIGHT, position=starting_position
if placed_proto == Prototype.SteamEngine:
dir = placed_entity.direction.value in [
direction.value,
DirectionInternal.opposite(direction).value,
]
assert dir, (
f"Expected direction {direction}, got {placed_entity.direction}"
)
game.move_to(
Position(x=starting_position.x + 3, y=starting_position.y - 3)
)
placed_entity = game.place_entity_next_to(
placed_proto, ref_entity.position, direction, spacing
)
expected_position = calculate_expected_position(
ref_entity.position, direction, spacing, ref_entity, placed_entity
)
assert placed_entity.position.is_close(
expected_position, tolerance=1
), (
f"Misplacement: {ref_proto.value[0]} -> {placed_proto.value[0]}, "
f"Direction: {direction}, Spacing: {spacing}, "
f"Expected: {expected_position}, Got: {placed_entity.position}"
# Check direction unless we are dealing with a pipe, which has no direction
elif placed_proto != Prototype.Pipe:
assert placed_entity.direction.value == direction.value, (
f"Expected direction {direction}, got {placed_entity.direction}"
)
if placed_proto == Prototype.SteamEngine:
dir = placed_entity.direction.value in [
direction.value,
DirectionInternal.opposite(direction).value,
]
assert dir, (
f"Expected direction {direction}, got {placed_entity.direction}"
)
# Check direction unless we are dealing with a pipe, which has no direction
elif placed_proto != Prototype.Pipe:
assert placed_entity.direction.value == direction.value, (
f"Expected direction {direction}, got {placed_entity.direction}"
)
game.instance.reset()
game.move_to(nearby_position)
game.instance.reset()
game.move_to(nearby_position)
def test_place_pipe_next_to_offshore_pump(game):

30
tests/actions/test_place_next_to_and_rotate.py
View File

@@ -6,21 +6,21 @@ from fle.env import DirectionInternal
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"boiler": 1,
"transport-belt": 1,
"stone-furnace": 1,
"burner-mining-drill": 1,
"burner-inserter": 2,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"steam-engine": 1,
"pipe": 1,
"offshore-pump": 1,
}
instance.reset()
yield instance.namespace
def game(configure_game):
return configure_game(
inventory={
"boiler": 1,
"transport-belt": 1,
"stone-furnace": 1,
"burner-mining-drill": 1,
"burner-inserter": 2,
"electric-mining-drill": 1,
"assembling-machine-1": 1,
"steam-engine": 1,
"pipe": 1,
"offshore-pump": 1,
}
)
def calculate_expected_position(

10
tests/actions/test_print.py
View File

@@ -1,13 +1,3 @@
import pytest
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def test_print_tuple(game):
"""
Print a tuple

27
tests/actions/test_render.py
View File

@@ -5,20 +5,19 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"iron-chest": 1,
"small-electric-pole": 20,
"iron-plate": 10,
"assembling-machine-1": 1,
"pipe-to-ground": 10,
"pipe": 30,
"transport-belt": 50,
"underground-belt": 30,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"iron-chest": 1,
"small-electric-pole": 20,
"iron-plate": 10,
"assembling-machine-1": 1,
"pipe-to-ground": 10,
"pipe": 30,
"transport-belt": 50,
"underground-belt": 30,
}
)
def test_basic_render(game):

9
tests/actions/test_request_path.py
View File

@@ -1,15 +1,6 @@
import pytest
from fle.env.entities import Position
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def test_path(game):
"""
Get a path from (0, 0) to (10, 0)

27
tests/actions/test_rotate.py
View File

@@ -5,20 +5,19 @@ from fle.env.entities import Position, Direction
@pytest.fixture()
def game(instance):
instance.initial_inventory = {
"iron-chest": 1,
"pipe": 10,
"assembling-machine-2": 2,
"transport-belt": 10,
"burner-inserter": 10,
"iron-plate": 10,
"assembling-machine-1": 1,
"copper-cable": 3,
}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(
inventory={
"iron-chest": 1,
"pipe": 10,
"assembling-machine-2": 2,
"transport-belt": 10,
"burner-inserter": 10,
"iron-plate": 10,
"assembling-machine-1": 1,
"copper-cable": 3,
}
)
def test_rotate_assembling_machine_2(game):

9
tests/actions/test_score.py
View File

@@ -1,12 +1,3 @@
import pytest
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
def test_get_score(game):
score, _ = game.score()
assert isinstance(score, int)

7
tests/actions/test_set_entity_recipe.py
View File

@@ -5,11 +5,8 @@ from fle.env.game_types import Prototype
@pytest.fixture()
def game(instance):
game.initial_inventory = {"assembling-machine-1": 1}
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(inventory={"assembling-machine-1": 1})
def test_set_entity_recipe(game):

19
tests/actions/test_set_research.py
View File

@@ -1,26 +1,11 @@
import pytest
from fle.env import FactorioInstance
from fle.env.game_types import Technology
from fle.commons.cluster_ips import get_local_container_ips
@pytest.fixture()
def game(instance):
# game.initial_inventory = {'assembling-machine-1': 1}
# from gym import FactorioInstance
ips, udp_ports, tcp_ports = get_local_container_ips()
instance = FactorioInstance(
address="localhost",
bounding_box=200,
tcp_port=tcp_ports[-1], # 27019,
all_technologies_researched=False,
fast=True,
inventory={},
)
instance.reset()
yield instance.namespace
instance.reset()
def game(configure_game):
return configure_game(all_technologies_researched=False)
def test_set_research(game):

26
tests/actions/test_sleep.py
View File

@@ -1,23 +1,11 @@
import time
import pytest
@pytest.fixture()
def game(instance):
instance.reset()
yield instance.namespace
instance.reset()
def test_sleep(game):
for i in range(10):
game.instance.set_speed(i)
speed = game.instance.get_speed()
start_time = time.time()
game.sleep(10)
end_time = time.time()
elapsed_seconds = end_time - start_time
assert elapsed_seconds * speed - 10 < 1, (
f"Sleep function did not work as expected for speed {i}"
)
@pytest.mark.parametrize("speed", range(10)) # 10 independent items
def test_sleep(game, speed):
game.instance.set_speed(speed)
start = time.time()
game.sleep(10)
elapsed = time.time() - start
assert elapsed * speed - 10 < 1, f"Sleep behaved unexpectedly at speed {speed}"

142
tests/conftest.py
View File

@@ -22,16 +22,62 @@ project_root = Path(__file__).parent.parent.parent
# sys.path.insert(0, str(project_root / 'src'))
@pytest.fixture() # scope="session")
def instance():
@pytest.fixture(scope="session")
def instance(pytestconfig, worker_id):
# from gym import FactorioInstance
ips, udp_ports, tcp_ports = get_local_container_ips()
# --- Parallel mapping (pytest-xdist) ---
# Docs-backed approach:
# - Use the built-in `worker_id` fixture to identify the worker ("gw0", "gw1", or "master"). [xdist how-to]
# - Use PYTEST_XDIST_WORKER_COUNT for total workers when present. [xdist how-to]
# Ref: https://pytest-xdist.readthedocs.io/en/stable/how-to.html#identifying-the-worker-process-during-a-test
xdist_count_env = os.environ.get("PYTEST_XDIST_WORKER_COUNT")
try:
opt_numproc = pytestconfig.getoption("numprocesses")
except Exception:
opt_numproc = None
if xdist_count_env and xdist_count_env.isdigit():
num_workers = int(xdist_count_env)
elif isinstance(opt_numproc, int) and opt_numproc > 0:
num_workers = opt_numproc
else:
num_workers = 1
# Determine the zero-based index for this worker.
if worker_id == "master":
worker_index = 0
elif worker_id.startswith("gw") and worker_id[2:].isdigit():
worker_index = int(worker_id[2:])
else:
worker_index = 0
ports_sorted = sorted(tcp_ports)
if num_workers > 1:
if len(ports_sorted) < num_workers:
raise pytest.UsageError(
f"pytest -n {num_workers} requested, but only {len(ports_sorted)} Factorio TCP ports were found: "
f"{ports_sorted}. Start {num_workers} servers, e.g. './run-envs.sh start -n {num_workers}'."
)
selected_port = ports_sorted[worker_index]
else:
# Single-process run: allow explicit override via env, else use last discovered port.
port_env = os.getenv("FACTORIO_RCON_PORT")
if port_env:
selected_port = int(port_env)
else:
if not ports_sorted:
raise pytest.UsageError(
"No Factorio TCP ports discovered. Did you start the headless server?"
)
selected_port = ports_sorted[-1]
try:
instance = FactorioInstance(
address="localhost",
bounding_box=200,
tcp_port=tcp_ports[-1], # 27019,
cache_scripts=False,
all_technologies_researched=True,
tcp_port=selected_port, # prefer env (CI) else last discovered
cache_scripts=True,
fast=True,
inventory={
"coal": 50,
@@ -50,6 +96,12 @@ def instance():
"small-electric-pole": 10,
},
)
instance.set_speed(10)
# Keep a canonical copy of the default test inventory to restore between tests
try:
instance.default_initial_inventory = dict(instance.initial_inventory)
except Exception:
instance.default_initial_inventory = instance.initial_inventory
yield instance
except Exception as e:
raise e
@@ -57,3 +109,83 @@ def instance():
# Cleanup RCON connections to prevent connection leaks
if "instance" in locals():
instance.cleanup()
# # Reset state between tests without recreating the instance
@pytest.fixture(autouse=True)
def _reset_between_tests(instance, request):
"""
Ensure clean state between tests without reloading Lua/scripts.
"""
# If this test explicitly uses `configure_game`, let that fixture perform
# the reset to avoid double resets and allow per-test options.
if "configure_game" in getattr(request, "fixturenames", []):
yield
return
# Restore the default inventory in case a previous test changed it
if hasattr(instance, "default_initial_inventory"):
try:
instance.initial_inventory = dict(instance.default_initial_inventory)
except Exception:
instance.initial_inventory = instance.default_initial_inventory
instance.reset(reset_position=True)
yield
# Provide a lightweight fixture that yields the game namespace derived from the
# already-maintained `instance`. Many tests only need `namespace` and not the
# full `instance`.
@pytest.fixture()
def namespace(instance):
yield instance.namespace
# Backwards-compatible alias used by many tests; simply yields `namespace`.
@pytest.fixture()
def game(namespace):
yield namespace
# Flexible configuration fixture for tests that need to tweak flags like
# `all_technologies_researched` and/or inventory in one step and receive a fresh namespace.
@pytest.fixture()
def configure_game(instance):
def _configure_game(
inventory: dict | None = None,
merge: bool = False,
persist_inventory: bool = False,
*,
reset_position: bool = True,
all_technologies_researched: bool = True,
):
# Always start from the canonical default inventory to avoid leakage
# from previous tests when this fixture is used.
if hasattr(instance, "default_initial_inventory"):
try:
instance.initial_inventory = dict(instance.default_initial_inventory)
except Exception:
instance.initial_inventory = instance.default_initial_inventory
instance.reset(
reset_position=reset_position,
all_technologies_researched=all_technologies_researched,
)
# Apply inventory first, so the subsequent reset reflects desired items
if inventory is not None:
print(f"Setting inventory: {inventory}")
if merge:
try:
updated = {**instance.initial_inventory, **inventory}
except Exception:
updated = dict(instance.initial_inventory)
updated.update(inventory)
else:
updated = dict(inventory)
if persist_inventory:
instance.initial_inventory = updated
instance.set_inventory(updated)
return instance.namespace
return _configure_game