<commit_before>import unittest
import csv
import os
import json
import collections

from flow.core.vehicles import Vehicles
from flow.core.traffic_lights import TrafficLights
from flow.controllers import IDMController, ContinuousRouter, RLController
from flow.core.params import SumoParams, EnvParams, NetParams, InitialConfig, \
    InFlows
from flow.core.util import emission_to_csv
from flow.utils.flow_warnings import deprecation_warning
from flow.utils.registry import make_create_env
from flow.utils.rllib import FlowParamsEncoder, get_flow_params

os.environ["TEST_FLAG"] = "True"


class TestEmissionToCSV(unittest.TestCase):
    """Tests the emission_to_csv function on a small file.

    Ensures that the headers are correct, the length is correct, and some of
    the components are correct.
    """

    def test_emission_to_csv(self):
        # current path
        current_path = os.path.realpath(__file__).rsplit("/", 1)[0]

        # run the emission_to_csv function on a small emission file
        emission_to_csv(current_path + "/test_files/test-emission.xml")

        # import the generated csv file and its headers
        dict1 = []
        filename = current_path + "/test_files/test-emission.csv"
        with open(filename, "r") as infile:
            reader = csv.reader(infile)
            headers = next(reader)
            for row in reader:
                dict1.append(dict())
                for i, key in enumerate(headers):
                    dict1[-1][key] = row[i]

        # check the names of the headers
        expected_headers = \
            ['time', 'CO', 'y', 'CO2', 'electricity', 'type', 'id', 'eclass',
             'waiting', 'NOx', 'fuel', 'HC', 'x', 'route', 'relative_position',
             'noise', 'angle', 'PMx', 'speed', 'edge_id', 'lane_number']

        self.assertCountEqual(headers, expected_headers)

        # check the number of rows of the generated csv file
        # Note that, rl vehicles are missing their final (reset) values, which
        # I don't think is a problem
        self.assertEqual(len(dict1), 104)


class TestWarnings(unittest.TestCase):
    """Tests warning functions located in flow.utils.warnings"""

    def test_deprecation_warning(self):
        # dummy class
        class Foo(object):
            pass

        # dummy attribute name
        dep_from = "bar_deprecated"
        dep_to = "bar_new"

        # check the deprecation warning is printing what is expected
        self.assertWarnsRegex(
            UserWarning, "The attribute bar_deprecated in Foo is deprecated, "
            "use bar_new instead.", deprecation_warning, Foo(), dep_from,
            dep_to)


class TestRegistry(unittest.TestCase):
    """Tests the methods located in flow/utils/registry.py"""

    def test_make_create_env(self):
        """Tests that the make_create_env methods generates an environment with
        the expected flow parameters."""
        # use a flow_params dict derived from flow/benchmarks/figureeight0.py
        vehicles = Vehicles()
        vehicles.add(
            veh_id="human",
            acceleration_controller=(IDMController, {
                "noise": 0.2
            }),
            routing_controller=(ContinuousRouter, {}),
            speed_mode="no_collide",
            num_vehicles=13)
        vehicles.add(
            veh_id="rl",
            acceleration_controller=(RLController, {}),
            routing_controller=(ContinuousRouter, {}),
            speed_mode="no_collide",
            num_vehicles=1)

        flow_params = dict(
            exp_tag="figure_eight_0",
            env_name="AccelEnv",
            scenario="Figure8Scenario",
            sumo=SumoParams(
                sim_step=0.1,
                render=False,
            ),
            env=EnvParams(
                horizon=1500,
                additional_params={
                    "target_velocity": 20,
                    "max_accel": 3,
                    "max_decel": 3,
                },
            ),
            net=NetParams(
                no_internal_links=False,
                additional_params={
                    "radius_ring": 30,
                    "lanes": 1,
                    "speed_limit": 30,
                    "resolution": 40,
                },
            ),
            veh=vehicles,
            initial=InitialConfig(),
            tls=TrafficLights(),
        )

        # some random version number for testing
        v = 23434

        # call make_create_env
        create_env, env_name = make_create_env(params=flow_params, version=v)

        # check that the name is correct
        self.assertEqual(env_name, '{}-v{}'.format(flow_params["env_name"], v))

        # create the gym environment
        env = create_env()

        # Note that we expect the port number in sumo_params to change, and
        # that this feature is in fact needed to avoid race conditions
        flow_params["sumo"].port = env.env.sumo_params.port

        # check that each of the parameter match
        self.assertEqual(env.env.env_params.__dict__,
                         flow_params["env"].__dict__)
        self.assertEqual(env.env.sumo_params.__dict__,
                         flow_params["sumo"].__dict__)
        self.assertEqual(env.env.traffic_lights.__dict__,
                         flow_params["tls"].__dict__)
        self.assertEqual(env.env.scenario.net_params.__dict__,
                         flow_params["net"].__dict__)
        self.assertEqual(env.env.scenario.net_params.__dict__,
                         flow_params["net"].__dict__)
        self.assertEqual(env.env.scenario.initial_config.__dict__,
                         flow_params["initial"].__dict__)
        self.assertEqual(env.env.__class__.__name__, flow_params["env_name"])
        self.assertEqual(env.env.scenario.__class__.__name__,
                         flow_params["scenario"])


class TestRllib(unittest.TestCase):
    """Tests the methods located in flow/utils/rllib.py"""

    def test_encoder_and_get_flow_params(self):
        """Tests both FlowParamsEncoder and get_flow_params.

        FlowParamsEncoder is used to serialize the data from a flow_params dict
        for replay by the visualizer later. Then, the get_flow_params method is
        used to try and read the parameters from the config file, and is
        checked to match expected results.
        """
        # use a flow_params dict derived from flow/benchmarks/merge0.py
        vehicles = Vehicles()
        vehicles.add(
            veh_id="human",
            acceleration_controller=(IDMController, {}),
            speed_mode="no_collide",
            num_vehicles=5)
        vehicles.add(
            veh_id="rl",
            acceleration_controller=(RLController, {}),
            speed_mode="no_collide",
            num_vehicles=0)

        inflow = InFlows()
        inflow.add(
            veh_type="human",<commit_msg>extended unit test coverage of utils (#246)<commit_after>import unittest
import csv
import os
import json
import collections

from flow.core.vehicles import Vehicles
from flow.core.traffic_lights import TrafficLights
from flow.controllers import IDMController, ContinuousRouter, RLController
from flow.core.params import SumoParams, EnvParams, NetParams, InitialConfig, \
    InFlows
from flow.core.util import emission_to_csv
from flow.utils.flow_warnings import deprecation_warning
from flow.utils.registry import make_create_env
from flow.utils.rllib import FlowParamsEncoder, get_flow_params

os.environ["TEST_FLAG"] = "True"


class TestEmissionToCSV(unittest.TestCase):
    """Tests the emission_to_csv function on a small file.

    Ensures that the headers are correct, the length is correct, and some of
    the components are correct.
    """

    def test_emission_to_csv(self):
        # current path
        current_path = os.path.realpath(__file__).rsplit("/", 1)[0]

        # run the emission_to_csv function on a small emission file
        emission_to_csv(current_path + "/test_files/test-emission.xml")

        # import the generated csv file and its headers
        dict1 = []
        filename = current_path + "/test_files/test-emission.csv"
        with open(filename, "r") as infile:
            reader = csv.reader(infile)
            headers = next(reader)
            for row in reader:
                dict1.append(dict())
                for i, key in enumerate(headers):
                    dict1[-1][key] = row[i]

        # check the names of the headers
        expected_headers = \
            ['time', 'CO', 'y', 'CO2', 'electricity', 'type', 'id', 'eclass',
             'waiting', 'NOx', 'fuel', 'HC', 'x', 'route', 'relative_position',
             'noise', 'angle', 'PMx', 'speed', 'edge_id', 'lane_number']

        self.assertCountEqual(headers, expected_headers)

        # check the number of rows of the generated csv file
        # Note that, rl vehicles are missing their final (reset) values, which
        # I don't think is a problem
        self.assertEqual(len(dict1), 104)


class TestWarnings(unittest.TestCase):
    """Tests warning functions located in flow.utils.warnings"""

    def test_deprecation_warning(self):
        # dummy class
        class Foo(object):
            pass

        # dummy attribute name
        dep_from = "bar_deprecated"
        dep_to = "bar_new"

        # check the deprecation warning is printing what is expected
        self.assertWarnsRegex(
            UserWarning, "The attribute bar_deprecated in Foo is deprecated, "
            "use bar_new instead.", deprecation_warning, Foo(), dep_from,
            dep_to)


class TestRegistry(unittest.TestCase):
    """Tests the methods located in flow/utils/registry.py"""

    def test_make_create_env(self):
        """Tests that the make_create_env methods generates an environment with
        the expected flow parameters."""
        # use a flow_params dict derived from flow/benchmarks/figureeight0.py
        vehicles = Vehicles()
        vehicles.add(
            veh_id="human",
            acceleration_controller=(IDMController, {
                "noise": 0.2
            }),
            routing_controller=(ContinuousRouter, {}),
            speed_mode="no_collide",
            num_vehicles=13)
        vehicles.add(
            veh_id="rl",
            acceleration_controller=(RLController, {}),
            routing_controller=(ContinuousRouter, {}),
            speed_mode="no_collide",
            num_vehicles=1)

        flow_params = dict(
            exp_tag="figure_eight_0",
            env_name="AccelEnv",
            scenario="Figure8Scenario",
            sumo=SumoParams(
                sim_step=0.1,
                render=False,
            ),
            env=EnvParams(
                horizon=1500,
                additional_params={
                    "target_velocity": 20,
                    "max_accel": 3,
                    "max_decel": 3,
                },
            ),
            net=NetParams(
                no_internal_links=False,
                additional_params={
                    "radius_ring": 30,
                    "lanes": 1,
                    "speed_limit": 30,
                    "resolution": 40,
                },
            ),
            veh=vehicles,
            initial=InitialConfig(),
            tls=TrafficLights(),
        )

        # some random version number for testing
        v = 23434

        # call make_create_env
        create_env, env_name = make_create_env(params=flow_params, version=v)

        # check that the name is correct
        self.assertEqual(env_name, '{}-v{}'.format(flow_params["env_name"], v))

        # create the gym environment
        env = create_env()

        # Note that we expect the port number in sumo_params to change, and
        # that this feature is in fact needed to avoid race conditions
        flow_params["sumo"].port = env.env.sumo_params.port

        # check that each of the parameter match
        self.assertEqual(env.env.env_params.__dict__,
                         flow_params["env"].__dict__)
        self.assertEqual(env.env.sumo_params.__dict__,
                         flow_params["sumo"].__dict__)
        self.assertEqual(env.env.traffic_lights.__dict__,
                         flow_params["tls"].__dict__)
        self.assertEqual(env.env.scenario.net_params.__dict__,
                         flow_params["net"].__dict__)
        self.assertEqual(env.env.scenario.net_params.__dict__,
                         flow_params["net"].__dict__)
        self.assertEqual(env.env.scenario.initial_config.__dict__,
                         flow_params["initial"].__dict__)
        self.assertEqual(env.env.__class__.__name__, flow_params["env_name"])
        self.assertEqual(env.env.scenario.__class__.__name__,
                         flow_params["scenario"])


class TestRllib(unittest.TestCase):
    """Tests the methods located in flow/utils/rllib.py"""

    def test_encoder_and_get_flow_params(self):
        """Tests both FlowParamsEncoder and get_flow_params.

        FlowParamsEncoder is used to serialize the data from a flow_params dict
        for replay by the visualizer later. Then, the get_flow_params method is
        used to try and read the parameters from the config file, and is
        checked to match expected results.
        """
        # use a flow_params dict derived from flow/benchmarks/merge0.py
        vehicles = Vehicles()
        vehicles.add(
            veh_id="human",
            acceleration_controller=(IDMController, {}),
            # for testing coverage purposes, we add a routing controller
            routing_controller=(ContinuousRouter, {}),
            speed_mode="no_collide",
            num_vehicles=5)
        vehicles.add(
            veh_id="rl",
            acceleration_controller=(RLController, {}),
            speed_mode="no_collide",
            num_vehicles=0)

        inflow = InFlows()
        inflow.add(
            veh_type="human",
{'fraction_non_alphanumeric': 0.09225723656067304, 'fraction_numerical': 0.006487950948238985, 'mean_word_length': 2.110889332446219, 'pattern_counts': {'":': 16, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '751301', 'n_tokens_mistral': 3892, 'n_tokens_neox': 3700, 'n_words': 949}