Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ addopts = [
"--durations-min=0.5",
"--disable-warnings",
"-s",
"-m not slow",
]
markers = [
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
Expand Down
24 changes: 15 additions & 9 deletions tests/database/test_database.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,10 @@ def test_create_new_model(self, tmp_path: Path):
# Check that the database file was created
assert db_path.exists()

def test_create_existing_model_raises_error(self, sirius_db: Path):
def test_create_existing_model_raises_error(self, session_sirius_db: Path):
"""Test that creating an existing model raises FileExistsError."""
with pytest.raises(FileExistsError):
Database.create(sirius_db)
Database.create(session_sirius_db)

def test_import_epanet(self, tmp_path: Path):
"""Test importing from an EPANET .inp file."""
Expand Down Expand Up @@ -110,9 +110,9 @@ def test_import_swmm(self, tmp_path: Path):
class TestDatabase:
"""Tests for the Database class."""

@pytest.fixture
@pytest.fixture(scope="class")
def db(self, class_sirius_db: Path):
"""Fixture providing a test Database instance."""
"""Fixture providing a test Database instance shared across the class."""
db = Database(class_sirius_db)
yield db
db.close()
Expand Down Expand Up @@ -187,11 +187,6 @@ def test_active_simulation(self, db):
simulation = db.active_simulation
assert simulation == "Sirius_1_DEMO"

def test_close(self, db):
"""Test close method."""
db.close()
assert not db.is_open

def test_top_level_imports(self, tmp_path: Path):
"""Test that top-level imports work correctly."""
from mikeplus import Database
Expand All @@ -211,3 +206,14 @@ def test_top_level_imports(self, tmp_path: Path):
db = open(new_db_path)
assert db.is_open
db.close()


class TestDatabaseClose:
"""Tests for the Database close functionality."""

def test_close(self, session_sirius_db: Path):
"""Test close method."""
db = Database(session_sirius_db)
assert db.is_open
db.close()
assert not db.is_open
12 changes: 9 additions & 3 deletions tests/database/test_queries.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,17 @@ class BaseQueryTest(BaseQuery[bool]):
def _execute_impl(self) -> bool:
return True

@pytest.fixture(scope="class")
def _db(self, session_sirius_db):
"""Open the database once for all tests in this class."""
db = Database(session_sirius_db)
yield db
db.close()

@pytest.fixture
def base_query(self, session_sirius_db):
def base_query(self, _db):
"""Fixture providing a BaseQuery instance."""
db = Database(session_sirius_db)
table = db.tables.msm_Link
table = _db.tables.msm_Link
return self.BaseQueryTest(table)

def test_base_query_error_if_not_open(self, session_sirius_db):
Expand Down
134 changes: 70 additions & 64 deletions tests/test_scenarios.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,56 +7,77 @@
import mikeplus as mp
from mikeplus.scenarios import Scenario, Alternative, AlternativeGroup

def test_api_access_scenarios_and_alternatives(sirius_db):
"""
Test the basic usage flow of the scenario API using an actual database.
"""
# Open a database using a test-specific isolated copy
db = mp.open(sirius_db)

# Access scenarios
active_scenario = db.scenarios.active
base_scenario = db.scenarios.base

# Verify scenario properties
assert isinstance(active_scenario, Scenario)
assert isinstance(base_scenario, Scenario)
assert active_scenario.name is not None
assert base_scenario.name is not None

# Test accessing specific alternative groups we know exist in the sirius_db
expected_groups = [
"CS Network data",
"River network data",
"Loads and boundaries data",
"Catchments and hydrology data",
"Transport data",
"Control rules data",
"Long Term Statistics data",
"Profiles and curves",
"2D overland",
"2D boundaries data"
]

for group_name in expected_groups:
# Verify we can access the group by name
alt_group = db.alternative_groups[group_name]
assert alt_group is not None
assert alt_group.name == group_name
assert len(alt_group.tables) > 0

class TestScenarioReadOnly:
"""Read-only scenario tests sharing a single database connection."""

@pytest.fixture(scope="class")
def db(self, session_sirius_db):
"""Open the database once for all read-only tests in this class."""
db = mp.open(session_sirius_db)
yield db
db.close()

def test_api_access_scenarios_and_alternatives(self, db):
"""
Test the basic usage flow of the scenario API using an actual database.
"""
# Access scenarios
active_scenario = db.scenarios.active
base_scenario = db.scenarios.base

# Check base and active alternatives are available
assert alt_group.base is not None
assert alt_group.active is not None

# Test getting alternative group by table name
# Use the "Catchments and hydrology data" group which should have msm_Catchment table
catchment_group = db.alternative_groups["Catchments and hydrology data"]
if catchment_group.tables: # Make sure there are tables in this group
table_name = catchment_group.tables[0] # Get the first table
group_by_table = db.alternative_groups.by_table(table_name)
assert group_by_table is not None
assert table_name in group_by_table.tables
# Verify scenario properties
assert isinstance(active_scenario, Scenario)
assert isinstance(base_scenario, Scenario)
assert active_scenario.name is not None
assert base_scenario.name is not None

# Test accessing specific alternative groups we know exist in the sirius_db
expected_groups = [
"CS Network data",
"River network data",
"Loads and boundaries data",
"Catchments and hydrology data",
"Transport data",
"Control rules data",
"Long Term Statistics data",
"Profiles and curves",
"2D overland",
"2D boundaries data"
]

for group_name in expected_groups:
# Verify we can access the group by name
alt_group = db.alternative_groups[group_name]
assert alt_group is not None
assert alt_group.name == group_name
assert len(alt_group.tables) > 0

# Check base and active alternatives are available
assert alt_group.base is not None
assert alt_group.active is not None

# Test getting alternative group by table name
# Use the "Catchments and hydrology data" group which should have msm_Catchment table
catchment_group = db.alternative_groups["Catchments and hydrology data"]
if catchment_group.tables: # Make sure there are tables in this group
table_name = catchment_group.tables[0] # Get the first table
group_by_table = db.alternative_groups.by_table(table_name)
assert group_by_table is not None
assert table_name in group_by_table.tables

def test_alternative_group_finding(self, db):
"""
Tests that AlternativeGroup.find_by_name returns the correct alternative.
"""
network_group = db.alternative_groups["CS Network data"]
found = network_group.find_by_name("Base Alternative")
assert len(found) == 1
assert found[0].name == "Base Alternative"

assert len(list(network_group)) == 2

assert network_group.by_name("Base Alternative").name == "Base Alternative"


def test_api_create_and_activate_scenario(sirius_db):
Expand Down Expand Up @@ -100,18 +121,3 @@ def test_api_create_and_activate_scenario(sirius_db):
assert found_scenario is not None
assert found_scenario.id == new_scenario.id


def test_alternative_group_finding(sirius_db):
"""
Tests that AlternativeGroup.find_by_name returns the correct alternative.
"""
db = mp.open(sirius_db)
network_group = db.alternative_groups["CS Network data"]
found = network_group.find_by_name("Base Alternative")
assert len(found) == 1
assert found[0].name == "Base Alternative"

assert len(list(network_group)) == 2

assert network_group.by_name("Base Alternative").name == "Base Alternative"

4 changes: 4 additions & 0 deletions tests/test_simulation_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from mikeplus.simulation_runner import SimulationRunner


@pytest.mark.slow
def test_mike_engine_run_1d(sirius_db):
"""Test running a MIKE 1D simulation."""
db = Database(sirius_db)
Expand All @@ -32,6 +33,7 @@ def test_mike_engine_run_1d(sirius_db):
db.close()

# TODO: Fix license issue on CI
@pytest.mark.slow
@pytest.mark.skip_ci(reason="Passes locally, license issue on CI")
def test_mike_engine_run_flood(flood_db):
"""Test running a Flood simulation."""
Expand All @@ -57,6 +59,7 @@ def test_mike_engine_run_flood(flood_db):

db.close()

@pytest.mark.slow
def test_mike_engine_run_swmm(swmm_db):
"""Test running a SWMM simulation."""
db = Database(swmm_db)
Expand Down Expand Up @@ -84,6 +87,7 @@ def test_mike_engine_run_swmm(swmm_db):
db.close()


@pytest.mark.slow
def test_mike_engine_run_epanet(epanet_demo_db):
"""Test running an EPANET simulation."""
db = Database(epanet_demo_db)
Expand Down
Loading