168 lines
4.8 KiB
Python
168 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import socket
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Mapping, Sequence
|
|
|
|
import pytest
|
|
|
|
from bedit_simulation import (
|
|
OpenModelicaRunner,
|
|
ProcessResult,
|
|
Simulation,
|
|
SimulationOptions,
|
|
SimulationStateError,
|
|
)
|
|
|
|
|
|
def _fake_omc(
|
|
command: Sequence[str],
|
|
working_directory: Path,
|
|
timeout: float | None,
|
|
environment: Mapping[str, str] | None,
|
|
) -> ProcessResult:
|
|
del timeout, environment
|
|
script = Path(command[-1]).read_text(encoding="utf-8")
|
|
if "buildModel(Example" in script:
|
|
(working_directory / "Example").touch()
|
|
if "system(" in script:
|
|
port_match = re.search(r"-port=(\d+)", script)
|
|
assert port_match is not None
|
|
with socket.create_connection(
|
|
("127.0.0.1", int(port_match.group(1)))
|
|
) as connection:
|
|
connection.sendall(
|
|
b'<status phase="integration" currentStepSize="0.1" '
|
|
b'time="1" progress="50"/>\n'
|
|
)
|
|
(working_directory / "Example_res.csv").write_text(
|
|
'"time","x"\n0,1\n1,2\n',
|
|
encoding="utf-8",
|
|
)
|
|
return ProcessResult(
|
|
tuple(command),
|
|
0,
|
|
'"Check of Example completed successfully."\n',
|
|
"",
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_runs_omc_through_configured_command(tmp_path: Path) -> None:
|
|
runner = OpenModelicaRunner(
|
|
["./run-omc-in-docker"],
|
|
executor=_fake_omc,
|
|
)
|
|
|
|
async def run() -> object:
|
|
simulation = Simulation(runner)
|
|
await simulation.load_modelica(
|
|
"model Example\nend Example;\n",
|
|
"Example",
|
|
)
|
|
await simulation.build(tmp_path)
|
|
return await simulation.run(
|
|
SimulationOptions(stop_time=2, number_of_intervals=20),
|
|
)
|
|
|
|
result = asyncio.run(run())
|
|
|
|
assert result.data == {"time": [0.0, 1.0], "x": [1.0, 2.0]}
|
|
assert result.result_file == tmp_path / "Example_res.csv"
|
|
script = (tmp_path / "run.mos").read_text(encoding="utf-8")
|
|
assert f'cd("{tmp_path}")' in script
|
|
assert "stepSize=0.1" in script
|
|
assert "-logFormat=xmltcp" in script
|
|
assert "simulate(Example" not in script
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_rejects_invalid_time_range() -> None:
|
|
with pytest.raises(ValueError, match="stop_time"):
|
|
SimulationOptions(start_time=1, stop_time=1)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_run_requires_a_built_model() -> None:
|
|
simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc))
|
|
|
|
async def run() -> None:
|
|
await simulation.load_modelica(
|
|
"model Example\nend Example;\n",
|
|
"Example",
|
|
)
|
|
with pytest.raises(SimulationStateError, match=r"call build\(\) first"):
|
|
await simulation.run()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_checks_existing_modelica_source(tmp_path: Path) -> None:
|
|
simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc))
|
|
|
|
async def check() -> object:
|
|
await simulation.load_modelica(
|
|
"model Example\nend Example;\n",
|
|
"Example",
|
|
)
|
|
return await simulation.check(working_directory=tmp_path)
|
|
|
|
result = asyncio.run(check())
|
|
|
|
assert result.successful
|
|
script = (tmp_path / "model-command.mos").read_text(encoding="utf-8")
|
|
assert "checkModel(Example);" in script
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_async_execution_uses_a_worker_thread(tmp_path: Path) -> None:
|
|
caller_thread = threading.get_ident()
|
|
execution_threads: list[int] = []
|
|
|
|
def executor(
|
|
command: Sequence[str],
|
|
working_directory: Path,
|
|
timeout: float | None,
|
|
environment: Mapping[str, str] | None,
|
|
) -> ProcessResult:
|
|
del working_directory, timeout, environment
|
|
execution_threads.append(threading.get_ident())
|
|
return ProcessResult(tuple(command), 0, '"OpenModelica test"', "")
|
|
|
|
simulation = Simulation(OpenModelicaRunner("omc", executor=executor))
|
|
|
|
result = asyncio.run(
|
|
simulation.execute(
|
|
"getVersion()",
|
|
working_directory=tmp_path,
|
|
)
|
|
)
|
|
|
|
assert result.return_code == 0
|
|
assert len(execution_threads) == 1
|
|
assert execution_threads[0] != caller_thread
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_requires_and_retains_a_loaded_model(tmp_path: Path) -> None:
|
|
simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc))
|
|
|
|
async def use_session() -> None:
|
|
with pytest.raises(SimulationStateError, match="no model is loaded"):
|
|
await simulation.check()
|
|
|
|
await simulation.load_modelica(
|
|
"model Example\nend Example;\n",
|
|
"Example",
|
|
)
|
|
check = await simulation.check(working_directory=tmp_path)
|
|
|
|
assert simulation.model_name == "Example"
|
|
assert simulation.last_check is check
|
|
|
|
asyncio.run(use_session())
|