Back to Repositories

Testing Synchronous Vectorized Environment Implementation in OpenAI Gym

This test suite validates the SyncVectorEnv implementation in OpenAI Gym, focusing on synchronous vectorized environment functionality. The tests ensure proper handling of multiple environments running in parallel while maintaining synchronized execution and consistent state management.

Test Coverage Overview

The test suite provides comprehensive coverage of SyncVectorEnv functionality:

Environment creation and initialization
Reset operations and observation space validation
Step execution with different action space configurations
Environment method calls and attribute management
Space compatibility checking
Custom space handling
Deterministic behavior and seeding

Implementation Analysis

The testing approach utilizes pytest’s parametrization features for thorough validation of vectorized environments. Tests employ various environment specifications including CartPole, FrozenLake, and BipedalWalker, verifying both discrete and continuous action/observation spaces.

The implementation focuses on validating synchronous execution patterns, space compatibility, and consistent random state management across vectorized instances.

Technical Details

Testing infrastructure includes:

pytest framework for test organization and execution
NumPy for array operations and assertions
Custom utility functions for environment creation and RNG validation
Environment specifications from Gym’s registration system
Various space implementations (Box, Discrete, MultiDiscrete, Tuple)

Best Practices Demonstrated

The test suite exemplifies several testing best practices:

Systematic validation of core functionality
Comprehensive edge case handling
Proper resource cleanup with environment closing
Parameterized testing for multiple environment types
Clear separation of test cases by functionality
Robust assertion patterns for different data types

openai/gym

tests/vector/test_sync_vector_env.py

            
import numpy as np
import pytest

from gym.envs.registration import EnvSpec
from gym.spaces import Box, Discrete, MultiDiscrete, Tuple
from gym.vector.sync_vector_env import SyncVectorEnv
from tests.envs.utils import all_testing_env_specs
from tests.vector.utils import (
    CustomSpace,
    assert_rng_equal,
    make_custom_space_env,
    make_env,
)


def test_create_sync_vector_env():
    env_fns = [make_env("FrozenLake-v1", i) for i in range(8)]
    env = SyncVectorEnv(env_fns)
    env.close()

    assert env.num_envs == 8


def test_reset_sync_vector_env():
    env_fns = [make_env("CartPole-v1", i) for i in range(8)]
    env = SyncVectorEnv(env_fns)
    observations, infos = env.reset()
    env.close()

    assert isinstance(env.observation_space, Box)
    assert isinstance(observations, np.ndarray)
    assert observations.dtype == env.observation_space.dtype
    assert observations.shape == (8,) + env.single_observation_space.shape
    assert observations.shape == env.observation_space.shape

    del observations


@pytest.mark.parametrize("use_single_action_space", [True, False])
def test_step_sync_vector_env(use_single_action_space):
    env_fns = [make_env("FrozenLake-v1", i) for i in range(8)]

    env = SyncVectorEnv(env_fns)
    observations = env.reset()

    assert isinstance(env.single_action_space, Discrete)
    assert isinstance(env.action_space, MultiDiscrete)

    if use_single_action_space:
        actions = [env.single_action_space.sample() for _ in range(8)]
    else:
        actions = env.action_space.sample()
    observations, rewards, terminateds, truncateds, _ = env.step(actions)

    env.close()

    assert isinstance(env.observation_space, MultiDiscrete)
    assert isinstance(observations, np.ndarray)
    assert observations.dtype == env.observation_space.dtype
    assert observations.shape == (8,) + env.single_observation_space.shape
    assert observations.shape == env.observation_space.shape

    assert isinstance(rewards, np.ndarray)
    assert isinstance(rewards[0], (float, np.floating))
    assert rewards.ndim == 1
    assert rewards.size == 8

    assert isinstance(terminateds, np.ndarray)
    assert terminateds.dtype == np.bool_
    assert terminateds.ndim == 1
    assert terminateds.size == 8

    assert isinstance(truncateds, np.ndarray)
    assert truncateds.dtype == np.bool_
    assert truncateds.ndim == 1
    assert truncateds.size == 8


def test_call_sync_vector_env():
    env_fns = [
        make_env("CartPole-v1", i, render_mode="rgb_array_list") for i in range(4)
    ]

    env = SyncVectorEnv(env_fns)
    _ = env.reset()
    images = env.call("render")
    gravity = env.call("gravity")

    env.close()

    assert isinstance(images, tuple)
    assert len(images) == 4
    for i in range(4):
        assert len(images[i]) == 1
        assert isinstance(images[i][0], np.ndarray)

    assert isinstance(gravity, tuple)
    assert len(gravity) == 4
    for i in range(4):
        assert isinstance(gravity[i], float)
        assert gravity[i] == 9.8


def test_set_attr_sync_vector_env():
    env_fns = [make_env("CartPole-v1", i) for i in range(4)]

    env = SyncVectorEnv(env_fns)
    env.set_attr("gravity", [9.81, 3.72, 8.87, 1.62])
    gravity = env.get_attr("gravity")
    assert gravity == (9.81, 3.72, 8.87, 1.62)

    env.close()


def test_check_spaces_sync_vector_env():
    # CartPole-v1 - observation_space: Box(4,), action_space: Discrete(2)
    env_fns = [make_env("CartPole-v1", i) for i in range(8)]
    # FrozenLake-v1 - Discrete(16), action_space: Discrete(4)
    env_fns[1] = make_env("FrozenLake-v1", 1)
    with pytest.raises(RuntimeError):
        env = SyncVectorEnv(env_fns)
        env.close()


def test_custom_space_sync_vector_env():
    env_fns = [make_custom_space_env(i) for i in range(4)]

    env = SyncVectorEnv(env_fns)
    reset_observations, infos = env.reset()

    assert isinstance(env.single_action_space, CustomSpace)
    assert isinstance(env.action_space, Tuple)

    actions = ("action-2", "action-3", "action-5", "action-7")
    step_observations, rewards, terminateds, truncateds, _ = env.step(actions)

    env.close()

    assert isinstance(env.single_observation_space, CustomSpace)
    assert isinstance(env.observation_space, Tuple)

    assert isinstance(reset_observations, tuple)
    assert reset_observations == ("reset", "reset", "reset", "reset")

    assert isinstance(step_observations, tuple)
    assert step_observations == (
        "step(action-2)",
        "step(action-3)",
        "step(action-5)",
        "step(action-7)",
    )


def test_sync_vector_env_seed():
    env = make_env("BipedalWalker-v3", seed=123)()
    sync_vector_env = SyncVectorEnv([make_env("BipedalWalker-v3", seed=123)])

    assert_rng_equal(env.action_space.np_random, sync_vector_env.action_space.np_random)
    for _ in range(100):
        env_action = env.action_space.sample()
        vector_action = sync_vector_env.action_space.sample()
        assert np.all(env_action == vector_action)


@pytest.mark.parametrize(
    "spec", all_testing_env_specs, ids=[spec.id for spec in all_testing_env_specs]
)
def test_sync_vector_determinism(spec: EnvSpec, seed: int = 123, n: int = 3):
    """Check that for all environments, the sync vector envs produce the same action samples using the same seeds"""
    env_1 = SyncVectorEnv([make_env(spec.id, seed=seed) for _ in range(n)])
    env_2 = SyncVectorEnv([make_env(spec.id, seed=seed) for _ in range(n)])
    assert_rng_equal(env_1.action_space.np_random, env_2.action_space.np_random)

    for _ in range(100):
        env_1_samples = env_1.action_space.sample()
        env_2_samples = env_2.action_space.sample()
        assert np.all(env_1_samples == env_2_samples)