testing

dummy

class DummyModel(model_config: dict, validate_config=True)

Bases: TrackedModel

close = <Mock id='133716815367536'>
initialize = <Mock id='133716815366192'>
install = <Mock id='133716815366864'>
classmethod reset_mocks()
setup = <Mock id='133716815365520'>
shutdown = <Mock id='133716815367200'>
update = <Mock id='133716815366528'>

fixtures

Shared pytest fixtures for Movici model testing.

This module is registered as a pytest plugin via the pytest11 entry point, making these fixtures automatically available to any test suite that depends on movici-simulation-core.

add_init_data(init_data_handler)
additional_attributes()
clean_strategies(global_schema)
config(model_config, init_data, time_scale)
create_model_tester(tmp_path_factory, init_data, global_schema)
global_schema(additional_attributes)
global_timeline_info()
init_data()
init_data_handler(tmp_path_factory)
model_name()
pytest_configure(config)
set_global_timeline_info(global_timeline_info, request)
time_scale()

helpers

assert_dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)

Deep compares two nested structures (such as dict) and asserts that they are equivalent. lists and numpy.ndarray``s are compared using ``numpy.isequal or numpy.isclose with equal_nan=True

Parameters:

  • a – the left dictionary object

  • b – the right dictionary object

  • rtol – relative tolerance used as in numpy.isclose

  • atol – absolute tolerance used as in numpy.isclose

assert_equivalent_data_mask(a, b)
compare_dataset_dicts(a, b, rtol=1e-05, atol=1e-08)
create_entity_group_with_data(entity_type: T | Type[T], data: dict, state: TrackedState | None = None, populate_schema=True) → T
data_mask_compare(data_mask)
dataset_data_to_numpy(data: dict | ndarray | list)
dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)
get_attribute(name='attr', **kwargs)
list_dir(path: Path)

model_schema

model_config_validator(model_schema: dict)

model_tester

class ModelTester(model, settings: Settings = None, init_data_handler=None, tmp_dir=None, schema: AttributeSchema | Sequence[AttributeSpec] | Plugin | None = None, raise_on_premature_shutdown=False)

Bases: object

add_init_data(name: str, data: dict | str | Path)
cleanup()
close()
initialize()
property model

A convenience property to reach the underlying model under test

new_time(timestamp: int)
classmethod run_scenario(model: Type[Model], model_name: str, scenario: dict, rtol=1e-05, atol=1e-08, use_new_time=True, global_schema: Any = None)
update(timestamp: int, data: dict | None, **msg_kwargs)
update_series(timestamp: int, data_series: Sequence[dict | None], **msg_kwargs)
class NumpyPreProcessor(model: TrackedModel, settings: Settings, schema=None)

Bases: PreProcessor

process_input(input_data: dict | None) → dict | None
process_result(result: Tuple[dict | None, int | None]) → Tuple[dict | None, int | None]
class Plugin(*args, **kwargs)

Bases: Protocol

install(obj: Extensible)
class PreProcessor(model: Model, settings: Settings, schema=None)

Bases: object

close(message: QuitMessage)
initialize(data_handler: InitDataHandler) → DataMask
new_time(message: NewTimeMessage)
process_input(input_data: dict | None) → dict | None
process_result(result: Tuple[dict | None, int | None]) → Tuple[dict | None, int | None]
update(msg: UpdateMessage, data) → Tuple[dict | None, int | None]
update_series(msg: UpdateSeriesMessage, data_series) → Tuple[dict | None, int | None]
compare_results(expected: Sequence[Tuple[int, dict | None, int | None]], results: Sequence[Tuple[int, dict | None, int | None]], rtol=1e-05, atol=1e-08) → List[Tuple[int, Dict[str, str]]]
format_errors(errors: List[Tuple[int, Dict[str, str]]])
read_schema(schema: AttributeSchema | Sequence[AttributeSpec] | Plugin | None) → AttributeSchema

road_network

Bases: object

classmethod create(links: List[Tuple[int, int]], id_offset=0, node_idx_offset=0)
from_idx: Sequence[int]
id: Sequence[int]
to_idx: Sequence[int]
class Nodes(id: 't.Sequence[int]', x: 't.Sequence[float]', y: 't.Sequence[float]')

Bases: object

classmethod create(nodes: List[Tuple[float, float]], id_offset=0)
duplicate(id_offset)
id: Sequence[int]
x: Sequence[float]
y: Sequence[float]
class RoadNetworkGenerator(nodes: List[Tuple[float, float]], links: List[Tuple[int, int]], geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)

Bases: object

generate()
static generate_node_entities(nodes: Nodes, ref_prefix='')
generate_road_segments(links: Links, transport_nodes: Nodes)
generate_transport_nodes(nodes: Nodes)
generate_virtual_nodes(nodes: Nodes)
generate_road_network(nodes, links, geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)