Consolidating test database creation

temoa/tutorial_assets/utopia.sql

@@ -458,7 +458,7 @@ CREATE TABLE demand
 );
 INSERT INTO "demand" VALUES('utopia',1990,'RH',25.2,'PJ','');
 INSERT INTO "demand" VALUES('utopia',2000,'RH',37.8,'PJ','');
-INSERT INTO "demand" VALUES('utopia',2010,'RH',5.669999999999999574e+01,'PJ','');
+INSERT INTO "demand" VALUES('utopia',2010,'RH',56.7,'PJ','');
 INSERT INTO "demand" VALUES('utopia',1990,'RL',5.6,'PJ','');
 INSERT INTO "demand" VALUES('utopia',2000,'RL',8.4,'PJ','');
 INSERT INTO "demand" VALUES('utopia',2010,'RL',12.6,'PJ','');

tests/conftest.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import sqlite3
 from pathlib import Path
 from typing import Any
@@ -34,83 +33,77 @@
 logging.getLogger('pyutilib').setLevel(logging.WARNING)
 
 
+# Central paths
+TEST_DATA_PATH = Path(__file__).parent / 'testing_data'
+TEST_OUTPUT_PATH = Path(__file__).parent / 'testing_outputs'
+SCHEMA_PATH = Path(__file__).parent.parent / 'temoa' / 'db_schema' / 'temoa_schema_v4.sql'
+
+
+def _build_test_db(
+    db_file: Path,
+    data_scripts: list[Path],
+    modifications: list[tuple[str, tuple[Any, ...]]] | None = None,
+) -> None:
+    """Helper to build a test database from central schema + data scripts + mods."""
+    if db_file.exists():
+        db_file.unlink()
+
+    with sqlite3.connect(db_file) as con:
+        con.execute('PRAGMA foreign_keys = OFF')
+        # 1. Load central schema
+        con.executescript(SCHEMA_PATH.read_text(encoding='utf-8'))
+        # Force FK OFF again as schema file might turn it on at the end
+        con.execute('PRAGMA foreign_keys = OFF')
+
+        # 2. Load data scripts
+        for script_path in data_scripts:
+            with open(script_path) as f:
+                con.executescript(f.read())
+
+        # 3. Apply modifications
+        if modifications:
+            for sql, params in modifications:
+                con.execute(sql, params)
+
+        # 4. Turn foreign keys back on
+        con.execute('PRAGMA foreign_keys = ON')
+        con.commit()
+
+
 def refresh_databases() -> None:
     """
     make new databases from source for testing...  removes possibility of contamination by earlier
     runs
     """
-    data_output_path = Path(__file__).parent / 'testing_outputs'
-    data_source_path = Path(__file__).parent / 'testing_data'
-    # utopia.sql is in tutorial_assets (single source of truth for unit-compliant data)
-    tutorial_assets_path = Path(__file__).parent.parent / 'temoa' / 'tutorial_assets'
-
     # Map source files to their locations
-    # (source_dir, source_file, output_file)
     databases = [
-        # Utopia uses the tutorial_assets source (unit-compliant)
-        (tutorial_assets_path, 'utopia.sql', 'utopia.sqlite'),
-        (tutorial_assets_path, 'utopia.sql', 'myo_utopia.sqlite'),
-        # Other test databases use testing_data
-        (data_source_path, 'test_system.sql', 'test_system.sqlite'),
-        (data_source_path, 'storageville.sql', 'storageville.sqlite'),
-        (data_source_path, 'mediumville.sql', 'mediumville.sqlite'),
-        (data_source_path, 'emissions.sql', 'emissions.sqlite'),
-        (data_source_path, 'materials.sql', 'materials.sqlite'),
-        (data_source_path, 'simple_linked_tech.sql', 'simple_linked_tech.sqlite'),
-        (data_source_path, 'seasonal_storage.sql', 'seasonal_storage.sqlite'),
-        (data_source_path, 'survival_curve.sql', 'survival_curve.sqlite'),
-        (data_source_path, 'annualised_demand.sql', 'annualised_demand.sqlite'),
+        # Utopia uses the unit-compliant data-only script
+        ('utopia_data.sql', 'utopia.sqlite'),
+        ('utopia_data.sql', 'myo_utopia.sqlite'),
+        # Other test databases
+        ('test_system.sql', 'test_system.sqlite'),
+        ('mediumville.sql', 'mediumville.sqlite'),
+        ('seasonal_storage.sql', 'seasonal_storage.sqlite'),
+        ('survival_curve.sql', 'survival_curve.sqlite'),
+        ('annualised_demand.sql', 'annualised_demand.sqlite'),
+        # Feature tests (separate for temporal consistency)
+        ('emissions.sql', 'emissions.sqlite'),
+        ('materials.sql', 'materials.sqlite'),
+        ('simple_linked_tech.sql', 'simple_linked_tech.sqlite'),
+        ('storageville.sql', 'storageville.sqlite'),
     ]
-    for source_dir, src, db in databases:
-        if Path.exists(data_output_path / db):
-            os.remove(data_output_path / db)
-        # make a new one and fill it
-        con = sqlite3.connect(data_output_path / db)
-        with open(source_dir / src) as script:
-            con.executescript(script.read())
-        con.close()
-
-
-def create_unit_test_db_from_sql(
-    source_sql_path: Path, output_db_path: Path, modifications: list[tuple[str, tuple[Any, ...]]]
-) -> None:
-    """Create a unit test database from SQL source with specific modifications.
-
-    Args:
-        source_sql_path: Path to the source SQL file
-        output_db_path: Path where the database should be created
-        modifications: List of (sql, params) tuples to apply after creation
-    """
-    if output_db_path.exists():
-        output_db_path.unlink()
-
-    # Generate database from SQL source and apply modifications
-    with sqlite3.connect(output_db_path) as conn:
-        # Execute the SQL source to create the database
-        conn.executescript(source_sql_path.read_text(encoding='utf-8'))
 
-        # Apply modifications
-        for sql, params in modifications:
-            conn.execute(sql, params)
-        conn.commit()
+    for src, db in databases:
+        _build_test_db(TEST_OUTPUT_PATH / db, [TEST_DATA_PATH / src])
 
 
 def create_unit_test_dbs() -> None:
     """Create unit test databases from SQL source for unit checking tests.
 
-    Generates databases from the single SQL source of truth (tutorial_assets/utopia.sql),
+    Generates databases from the single SQL source of truth (utopia_data.sql),
     applying modifications for each test case.
     """
-    test_output_dir = Path(__file__).parent / 'testing_outputs'
-    test_output_dir.mkdir(exist_ok=True)
-
-    # Source SQL file path (single source of truth)
-    source_sql = Path(__file__).parent.parent / 'temoa' / 'tutorial_assets' / 'utopia.sql'
-
-    if not source_sql.exists():
-        raise FileNotFoundError(
-            f'Source SQL not found at: {source_sql}. Please ensure the Utopia tutorial SQL exists.'
-        )
+    TEST_OUTPUT_PATH.mkdir(exist_ok=True)
 
     # Define unit test variations with their modifications
     unit_test_variations = [
@@ -169,8 +162,11 @@ def create_unit_test_dbs() -> None:
     ]
 
     for db_name, modifications in unit_test_variations:
-        output_path = test_output_dir / db_name
-        create_unit_test_db_from_sql(source_sql, output_path, modifications)
+        _build_test_db(
+            TEST_OUTPUT_PATH / db_name,
+            [TEST_DATA_PATH / 'utopia_data.sql'],
+            modifications,
+        )
         logger.info('Created unit test DB: %s', db_name)
 
 

tests/testing_configs/config_annualised_demand.toml

@@ -1,73 +1,21 @@
-# this config is used for testing in test_full_runs.py
 scenario = "test run"
 scenario_mode = "perfect_foresight"
-
 input_database = "tests/testing_outputs/annualised_demand.sqlite"
 output_database = "tests/testing_outputs/annualised_demand.sqlite"
 neos = false
-
-# solver
 solver_name = "appsi_highs"
-
-# generate an excel file in the output_files folder
 save_excel = false
-
-# save the duals in the output .sqlite database
 save_duals = false
-
-# save a copy of the pyomo-generated lp file to the outputs folder (may be large file!)
 save_lp_file = false
+time_sequencing = "representative_periods"
+reserve_margin = "static"
 
-# ------------------------------------
-#          MODEL PARAMETERS
-#  these are specific to each model
-# ------------------------------------
-
-# What seasons represent in the model
-# Options:
-#   'consecutive_days'
-#       Seasons are a set of days in order, with each season representing only one day. Examples
-#       might be a model of a representative week with 7 days or a whole-year model with 365 days.
-#       Seasonal storage need not be tagged and the time_season_sequential table can be left empty.
-#   'representative_periods'
-#       Each season represents a number of days, though not necessarily in any particular order.
-#       If using inter-season constraints like seasonal storage or ramp rates, the true sequence
-#       must be defined using the time_season_sequential table. Seasonal storage must also be tagged in
-#       the technology table.
-#   'seasonal_timeslices'
-#       Each season represents a sequential slice of the year, with one or many days represented per
-#       season. We assume that the true sequence is the same as the TimeSeason sequence, so the
-#       time_season_sequential table can be left empty. Seasonal storage must still be tagged.
-#   'manual'
-#       The sequence of time slices is defined manually in the TimeNext table (which is commented out
-#       in the schema). This is an advanced feature and not recommended for most users. Seasonal
-#       storage must be tagged and the time_season_sequential table filled.
-time_sequencing = 'representative_periods'
-
-# How contributions to the planning reserve margin are calculated
-# Options:
-#   'static'
-#       Traditional planning reserve formulation. Contributions are independent of hourly availability:
-#                   capacity value = net capacity * capacity credit
-#   'dynamic'
-#       Contributions are available output including a capacity derate factor (e.g., forced outage rate).
-#       For most generators, contributions are available (derated) output in each time slice:
-#                   capacity value = net capacity * reserve capacity derate * capacity factor
-#       For storage, contributions are (derated) actual output in each time slice:
-#                   capacity value = flow out * reserve capacity derate
-reserve_margin = 'static'
-
-# ---------------------------------------------------
-#                   MODE OPTIONS
-# options below are mode-specific and will be ignored
-# if the run is not executed in that mode.
-# ---------------------------------------------------
 [MGA]
-cost_epsilon = 0.03    # 3% relaxation on optimal cost
-iteration_limit = 15   # max iterations to perform
-time_limit_hrs = 1     # max time
-axis = "tech_category_activity"   # use the tech activity Manager to control exploration based on categories in Tech
-weighting = "hull_expansion"  # use a convex hull expansion algorithm to weight exploration
+cost_epsilon = 0.03
+iteration_limit = 15
+time_limit_hrs = 1
+axis = "tech_category_activity"
+weighting = "hull_expansion"
 
 [myopic]
-myopic_view = 2  # number of periods seen at one iteration
+myopic_view = 2

tests/testing_configs/config_emissions.toml

@@ -1,71 +1,19 @@
-# this config is used for testing in test_full_runs.py
 scenario = "test run"
 scenario_mode = "perfect_foresight"
-
 input_database = "tests/testing_outputs/emissions.sqlite"
 output_database = "tests/testing_outputs/emissions.sqlite"
 neos = false
-
-# solver
 solver_name = "appsi_highs"
-
-# generate an excel file in the output_files folder
 save_excel = false
-
-# save the duals in the output .sqlite database
 save_duals = false
-
-# save a copy of the pyomo-generated lp file to the outputs folder (may be large file!)
 save_lp_file = false
+time_sequencing = "seasonal_timeslices"
+reserve_margin = "static"
 
-# ------------------------------------
-#          MODEL PARAMETERS
-#  these are specific to each model
-# ------------------------------------
-
-# What seasons represent in the model
-# Options:
-#   'consecutive_days'
-#       Seasons are a set of days in order, with each season representing only one day. Examples
-#       might be a model of a representative week with 7 days or a whole-year model with 365 days.
-#       Seasonal storage need not be tagged and the time_season_sequential table can be left empty.
-#   'representative_periods'
-#       Each season represents a number of days, though not necessarily in any particular order.
-#       If using inter-season constraints like seasonal storage or ramp rates, the true sequence
-#       must be defined using the time_season_sequential table. Seasonal storage must also be tagged in
-#       the technology table.
-#   'seasonal_timeslices'
-#       Each season represents a sequential slice of the year, with one or many days represented per
-#       season. We assume that the true sequence is the same as the TimeSeason sequence, so the
-#       time_season_sequential table can be left empty. Seasonal storage must still be tagged.
-#   'manual'
-#       The sequence of time slices is defined manually in the TimeNext table (which is commented out
-#       in the schema). This is an advanced feature and not recommended for most users. Seasonal
-#       storage must be tagged and the time_season_sequential table filled.
-time_sequencing = 'seasonal_timeslices'
-
-# How contributions to the planning reserve margin are calculated
-# Options:
-#   'static'
-#       Traditional planning reserve formulation. Contributions are independent of hourly availability:
-#                   capacity value = net capacity * capacity credit
-#   'dynamic'
-#       Contributions are available output including a capacity derate factor (e.g., forced outage rate).
-#       For most generators, contributions are available (derated) output in each time slice:
-#                   capacity value = net capacity * reserve capacity derate * capacity factor
-#       For storage, contributions are (derated) actual output in each time slice:
-#                   capacity value = flow out * reserve capacity derate
-reserve_margin = 'static'
-
-# ---------------------------------------------------
-#                   MODE OPTIONS
-# options below are mode-specific and will be ignored
-# if the run is not executed in that mode.
-# ---------------------------------------------------
 [MGA]
 slack = 0.1
 iterations = 4
-weight = "integer"   # currently supported: [integer, normalized]
+weight = "integer"
 
 [myopic]
-myopic_view = 2  # number of periods seen at one iteration
+myopic_view = 2