Coverage for tests/test_SimpleBuilding.py: 93%
57 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-03-26 09:43 +0000
1import pytest
2import pandas as pd
3import os
4import sys
5from pathlib import Path
6import importlib.util
7import json
8from util import module_cleanup, round_floats_in_structure
10# Add the project root to the Python path to allow for absolute imports
11# This helps in locating the agentlib_flexquant package if needed
12root_path = Path(__file__).parent.parent
13sys.path.insert(0, str(root_path))
16def create_dataframe_summary(df: pd.DataFrame, precision: int = 6) -> dict:
17 """
18 Creates a robust, compact summary of a DataFrame for snapshotting.
20 This summary is designed to be insensitive to minor floating-point differences
21 while being highly sensitive to meaningful data changes.
23 Args:
24 df: The pandas DataFrame to summarize.
25 precision: The number of decimal places to round float values to.
27 Returns:
28 A dictionary containing the summary.
29 """
30 if df is None or df.empty:
31 return {"error": "DataFrame is empty or None"}
33 # Get descriptive statistics and round them to handle float precision issues
34 summary_stats = df.describe().round(precision)
36 # Convert the stats DataFrame to a dictionary. This may have tuple keys.
37 stats_dict_raw = summary_stats.to_dict()
39 # Create a new dictionary, converting any tuple keys into strings.
40 # e.g., ('lower', 'P_el') becomes 'lower.P_el'
41 stats_dict_clean = {
42 ".".join(map(str, k)) if isinstance(k, tuple) else str(k): v
43 for k, v in stats_dict_raw.items()
44 }
46 # Create the final summary object
47 summary = {
48 "shape": df.shape,
49 "columns": df.columns.tolist(),
50 "index_start": str(tuple(float(x) for x in df.index.min())),
51 "index_end": str(tuple(float(x) for x in df.index.max())),
52 "statistics": stats_dict_clean,
53 "head_5_rows": df.head(5).round(precision).to_dict(orient='split'),
54 "tail_5_rows": df.tail(5).round(precision).to_dict(orient='split'),
55 }
56 return summary
59def assert_frame_matches_summary_snapshot(snapshot, df: pd.DataFrame,
60 snapshot_name: str):
61 """
62 Asserts that a DataFrame's summary matches a stored snapshot.
64 This function creates a summary of the dataframe and uses pytest-snapshot
65 to compare it against a stored version.
66 """
67 # Create a summary of the dataframe
68 summary = create_dataframe_summary(df)
70 # Round all numbers in the summary to handle cross-platform differences
71 rounded_summary = round_floats_in_structure(summary, precision=5)
73 # Convert the summary dictionary to a formatted JSON string
74 summary_json = json.dumps(rounded_summary, indent=2, sort_keys=True)
76 # Use snapshot.assert_match on the small, stable JSON string
77 snapshot.assert_match(summary_json, snapshot_name)
80def run_example_from_path(example_path: Path):
81 """
82 Dynamically imports and runs the 'run_example' function from a script
83 in the specified directory.
85 This function robustly handles changing the working directory AND the
86 Python import path, ensuring the script can find both its local files
87 and its local modules.
88 """
89 run_script_path = example_path / 'main_single_run.py'
90 if not run_script_path.is_file():
91 raise FileNotFoundError(
92 f"Could not find the run script at {run_script_path}. "
93 "Please ensure it is named 'run.py' or adjust the test code."
94 )
96 # --- SETUP: Store original paths before changing them ---
97 original_cwd = Path.cwd()
98 original_sys_path = sys.path[:] # Create a copy of the sys.path list
100 module_name = f"agentlib_flexquant.tests.examples.{example_path.name}"
102 try:
103 # --- STEP 1: Change CWD for file access (e.g., config.json) ---
104 os.chdir(example_path)
106 # --- STEP 2: Add example dir to sys.path for module imports ---
107 sys.path.insert(0, str(example_path))
109 # Dynamically import the run_example function from the script
110 spec = importlib.util.spec_from_file_location(module_name, run_script_path)
111 run_module = importlib.util.module_from_spec(spec)
112 sys.modules[module_name] = run_module
113 spec.loader.exec_module(run_module)
115 if not hasattr(run_module, 'run_example'):
116 raise AttributeError(
117 "The 'run.py' script must contain a 'run_example' function.")
119 run_module.sim_config = "mpc_and_sim/fmu_config_linux.json"
121 # Execute the function and get the results
122 results = run_module.run_example(until=3600)
123 return results
125 finally:
126 # --- TEARDOWN: Always restore original paths to avoid side-effects ---
127 os.chdir(original_cwd)
128 sys.path[:] = original_sys_path # Restore the original sys.path
131def test_simplebuilding(snapshot, module_cleanup):
132 """
133 Unit test for the SimpleBuilding example using snapshot testing.
135 This test runs the example via its own run script and compares the
136 full resulting dataframes against stored snapshots.
137 """
138 # Define the path to the example directory
139 example_path = root_path / 'examples' / 'SimpleBuilding'
141 # Run the example and get the results object
142 res = run_example_from_path(example_path)
144 # Extract the full resulting dataframes as requested
145 df_neg_flex_res = res["NegFlexMPC"]["NegFlexMPC"]
146 df_pos_flex_res = res["PosFlexMPC"]["PosFlexMPC"]
147 df_baseline_res = res["Baseline"]["Baseline"]
148 df_indicator_res = res["FlexibilityIndicator"]["FlexibilityIndicator"]
150 # Assert that a summary of each result DataFrame matches its snapshot
151 assert_frame_matches_summary_snapshot(
152 snapshot,
153 df_neg_flex_res,
154 'SimpleBuilding_neg_flex_summary.json'
155 )
156 assert_frame_matches_summary_snapshot(
157 snapshot,
158 df_pos_flex_res,
159 'SimpleBuilding_pos_flex_summary.json'
160 )
161 assert_frame_matches_summary_snapshot(
162 snapshot,
163 df_baseline_res,
164 'SimpleBuilding_baseline_summary.json'
165 )
166 assert_frame_matches_summary_snapshot(
167 snapshot,
168 df_indicator_res,
169 'SimpleBuilding_indicator_summary.json'
170 )
173if __name__ == "__main__":
174 pytest.main([__file__, "-v"])