From ab0a4e5132169abc0f9cf2d6f95c522ec7e30ad6 Mon Sep 17 00:00:00 2001 From: Jason Marechal Date: Thu, 26 Oct 2023 11:45:56 +0200 Subject: [PATCH] Revert driver changes (out of scope) --- .../antares_xpansion/full_run_driver.py | 3 + .../problem_generator_driver.py | 10 +++ tests/python/test_problem_generator_driver.py | 68 +++++++++++++++++++ 3 files changed, 81 insertions(+) diff --git a/src/python/antares_xpansion/full_run_driver.py b/src/python/antares_xpansion/full_run_driver.py index d10ef46bb..d88baa2fc 100644 --- a/src/python/antares_xpansion/full_run_driver.py +++ b/src/python/antares_xpansion/full_run_driver.py @@ -29,6 +29,9 @@ def prepare_drivers(self, output_path: Path, """ problem generation step : getnames + lp_namer """ + # Pb Gen pre-step + self.problem_generation_driver.clear_old_log() + self.problem_generation_driver.is_relaxed = problem_generation_is_relaxed self.keep_mps = benders_keep_mps diff --git a/src/python/antares_xpansion/problem_generator_driver.py b/src/python/antares_xpansion/problem_generator_driver.py index 09af5e0fd..1df3e4e3b 100644 --- a/src/python/antares_xpansion/problem_generator_driver.py +++ b/src/python/antares_xpansion/problem_generator_driver.py @@ -3,6 +3,7 @@ """ import os +import shutil import subprocess import sys from dataclasses import dataclass @@ -65,6 +66,7 @@ def launch(self, output_path: Path, is_relaxed: bool): """ problem generation step : getnames + lp_namer """ + self.clear_old_log() self.logger.info("Problem Generation") self.output_path = output_path @@ -79,6 +81,8 @@ def set_output_path(self, output_path): self._output_path = output_path self.xpansion_output_dir = output_path.parent / \ (output_path.stem + "-Xpansion") + if self.xpansion_output_dir.exists(): + shutil.rmtree(self.xpansion_output_dir) os.makedirs(self.xpansion_output_dir) self._lp_path = os.path.normpath( os.path.join(self.xpansion_output_dir, 'lp')) @@ -89,6 +93,10 @@ def set_output_path(self, output_path): def get_output_path(self): return self._output_path + def clear_old_log(self): + if (os.path.isfile(str(self.lp_namer_exe_path) + '.log')): + os.remove(str(self.lp_namer_exe_path) + '.log') + def _lp_step(self): """ copies area and interco files and launches the lp_namer @@ -104,6 +112,8 @@ def _lp_step(self): "ERROR: exited lpnamer with status %d" % returned_l.returncode) def create_lp_dir(self): + if os.path.isdir(self._lp_path): + shutil.rmtree(self._lp_path) os.makedirs(self._lp_path) def lp_namer_options(self): diff --git a/tests/python/test_problem_generator_driver.py b/tests/python/test_problem_generator_driver.py index 21ae640e4..7d7647407 100644 --- a/tests/python/test_problem_generator_driver.py +++ b/tests/python/test_problem_generator_driver.py @@ -1,6 +1,7 @@ import shutil from datetime import date, datetime from pathlib import Path +from unittest.mock import patch import pytest from antares_xpansion.problem_generator_driver import ProblemGeneratorData, ProblemGeneratorDriver @@ -39,6 +40,12 @@ def test_problem_generator_data(self): assert problem_generator_driver.user_weights_file_path == self.empty_pblm_gen_data.user_weights_file_path assert problem_generator_driver.lp_namer_exe_path == self.empty_pblm_gen_data.lp_namer_exe_path + def test_output_path(self, tmp_path): + problem_generator_driver = ProblemGeneratorDriver( + self.empty_pblm_gen_data) + with pytest.raises(ProblemGeneratorDriver.OutputPathError): + problem_generator_driver.launch(tmp_path / "i_don_t_exist", False) + def test_lp_namer_exe_does_not_exit(self, tmp_path): self._create_empty_area_file(tmp_path) @@ -52,6 +59,67 @@ def test_lp_namer_exe_does_not_exit(self, tmp_path): def xpansion_output(self, tmp_path): return tmp_path.parent / (tmp_path.name + '-Xpansion') + def test_clear_old_log(self, tmp_path): + + lp_namer_file = tmp_path / self.lp_exe + lp_namer_file.write_text("") + pblm_gen_data = ProblemGeneratorData(keep_mps=False, + additional_constraints="", + user_weights_file_path=Path(""), + weight_file_name_for_lp="", + lp_namer_exe_path=lp_namer_file, + active_years=[]) + self._create_empty_area_file(tmp_path) + self._create_empty_interco_file(tmp_path) + output_zipped = get_zipped_output(tmp_path) + log_file_name = self.lp_exe + ".log" + xpansion_dir = self.xpansion_output(tmp_path) + log_file = xpansion_dir / log_file_name + xpansion_dir.mkdir() + log_file.write_text("bla bla") + assert log_file.exists() + + pblm_gen = ProblemGeneratorDriver(pblm_gen_data) + with patch(SUBPROCESS_RUN, autospec=True): + with pytest.raises(ProblemGeneratorDriver.LPNamerExecutionError): + pblm_gen.launch(output_zipped, False) + + assert not log_file.exists() + + def test_clean_lp_dir_before_run(self, tmp_path): + + lp_namer_file = tmp_path / self.lp_exe + lp_namer_file.write_text("") + pblm_gen_data = ProblemGeneratorData(keep_mps=False, + additional_constraints="", + user_weights_file_path=Path(""), + weight_file_name_for_lp="", + lp_namer_exe_path=lp_namer_file, + active_years=[]) + self._create_empty_area_file(tmp_path) + self._create_empty_interco_file(tmp_path) + output_zipped = get_zipped_output(tmp_path) + log_file_name = self.lp_exe + ".log" + xpansion_dir = self.xpansion_output(tmp_path) + xpansion_dir.mkdir() + log_file = xpansion_dir / log_file_name + log_file.write_text("bla bla") + + lp_dir = xpansion_dir / "lp" + lp_dir.mkdir() + lp_dir_sub_file_1 = lp_dir / "file1" + lp_dir_sub_file_1.write_text("") + + assert lp_dir.exists() + assert lp_dir_sub_file_1.exists() + problem_generator_driver = ProblemGeneratorDriver(pblm_gen_data) + with patch(SUBPROCESS_RUN, autospec=True) as run_function: + run_function.return_value.returncode = 0 + problem_generator_driver.launch(output_zipped, False) + + assert lp_dir.exists() + assert not lp_dir_sub_file_1.exists() + def _get_expected_mps_txt(self, tmp_path): weeks = [1, 2, 3]