diff --git a/antarest/launcher/model.py b/antarest/launcher/model.py index 85225cd8d6..9106c703eb 100644 --- a/antarest/launcher/model.py +++ b/antarest/launcher/model.py @@ -81,7 +81,7 @@ class LauncherParametersDTO(AntaresBaseModel, extra="forbid"): time_limit: int = 240 * 3600 # Default value set to 240 hours (in seconds) xpansion: XpansionParametersDTO | bool | None = None xpansion_r_version: bool = False - archive_output: bool = True + archive_output: bool = Field(deprecated=True, default=True) auto_unzip: bool = True output_suffix: FileNameStr | None = None other_options: str | None = None diff --git a/antarest/launcher/service.py b/antarest/launcher/service.py index 6a4ab54d13..b4ded6dae5 100644 --- a/antarest/launcher/service.py +++ b/antarest/launcher/service.py @@ -64,9 +64,10 @@ from antarest.login.service import LoginService from antarest.login.utils import current_user_context, get_current_user, require_current_user from antarest.output.service import OutputService +from antarest.study.model import Study from antarest.study.repository import AccessPermissions, StudyFilter from antarest.study.service import StudyService -from antarest.study.storage.utils import assert_permission, extract_output_name, find_single_output_path +from antarest.study.storage.utils import assert_permission, extract_output_name, find_single_output_path, is_managed logger = logging.getLogger(__name__) @@ -532,32 +533,43 @@ def _import_output( job_result = self.job_result_repository.get(job_id) if not job_result: raise JobNotFound() - study_id = job_result.study_id + job_owner_id = job_result.owner_id job_launch_params = LauncherParametersDTO.from_launcher_params(job_result.launcher_params) output_true_path = find_single_output_path(output_path) - if not output_true_path.is_dir() and not is_zip(output_true_path): raise NoValidOutputError(f"No valid output for job {job_id}: {output_true_path}") self._save_solver_stats(job_result, output_true_path) + study_id = job_result.study_id + study = db.session.get(Study, study_id) + if study is None: + return self._import_fallback_output(job_id, output_true_path, job_launch_params.output_suffix) + is_study_managed = is_managed(study) + zip_path: Path | None = None - # Optimized path for studies stored on external devices, that will then be unarchived there. - # TODO: that whole optimization path should be refactored to: - # - be more explicit - # - not affect internal studies - if job_launch_params.archive_output: - stopwatch = StopWatch() - logger.info("Re zipping output for transfer") - zip_path = output_true_path.parent / f"{output_true_path.name}.zip" - archive_dir(output_true_path, target_archive_path=zip_path, archive_format=ArchiveFormat.ZIP) - logger.info(f"Zipped output for job {job_id} in {stopwatch}s") - final_output_path = zip_path - else: + + if is_zip(output_true_path): + # Possible if the option `-z` was used to run the solver. final_output_path = output_true_path + else: + if is_study_managed and job_launch_params.auto_unzip: + # Nothing to do, the output is already unarchived. + final_output_path = output_true_path + else: + # For studies stored on external devices, it's faster to re-zip the output for transfer and unarchive it there. + # Also, for managed studies when the user did not ask for auto-unzip, we'd better re-zip the output here instead of copying the tree and then zip it. + # TODO: that whole optimization should be refactored to be more explicit + stopwatch = StopWatch() + logger.info("Re zipping output for transfer") + zip_path = output_true_path.parent / f"{output_true_path.name}.zip" + archive_dir(output_true_path, target_archive_path=zip_path, archive_format=ArchiveFormat.ZIP) + logger.info(f"Zipped output for job {job_id} in {stopwatch}s") + final_output_path = zip_path + with db(): try: if job_owner_id: @@ -575,15 +587,11 @@ def _import_output( logs=additional_logs, ) except StudyNotFoundError: - return self._import_fallback_output( - job_id, - final_output_path, - job_launch_params.output_suffix, - ) + return self._import_fallback_output(job_id, final_output_path, job_launch_params.output_suffix) finally: # Delete the temporary zip file, which now has been imported if zip_path: - os.unlink(zip_path) + zip_path.unlink(missing_ok=True) def _download_fallback_output(self, job_id: str) -> FileDownloadTaskDTO: output_path = self._get_job_output_fallback_path(job_id) diff --git a/antarest/output/service.py b/antarest/output/service.py index b78f86b4a8..c36e521b35 100644 --- a/antarest/output/service.py +++ b/antarest/output/service.py @@ -410,9 +410,10 @@ def import_output( logger.info(f"output added to study {uuid}") - # Optimized path for studies stored on external devices, that will then be unarchived there. # TODO: as commented elsewhere, that workflow should be refactored to not span multiple files if output_id and isinstance(output, Path) and output.suffix == ArchiveFormat.ZIP and auto_unzip: + # Always the case for studies stored on external devices, as they will be unarchived there for performance reasons. + # It is also possible for managed studies if the option `-z` was used to run the solver. self.unarchive_output(uuid, output_id) return output_id diff --git a/antarest/output/storage/file/abstract_storage.py b/antarest/output/storage/file/abstract_storage.py index 0d151f50b2..0ad1284e5d 100644 --- a/antarest/output/storage/file/abstract_storage.py +++ b/antarest/output/storage/file/abstract_storage.py @@ -182,14 +182,14 @@ def _output_exists(outputs_root: Path, output_id: str) -> bool: def _import_zip_as_archived( study_id: str, output_zip_path: Path, study_outputs_path: Path, output_name_suffix: str | None, logs: SimulationLogs ) -> str: - """Simply copies the zip to destination study/output/.zip, with the right name extracted from output + """Simply moves the zip to destination study/output/.zip, with the right name extracted from output files.""" t = StopWatch() output_full_name = extract_output_name(output_zip_path, output_name_suffix) final_path = _archived_output_path(study_outputs_path, output_full_name) study_outputs_path.mkdir(exist_ok=True) - shutil.copyfile(output_zip_path, final_path) + shutil.move(output_zip_path, final_path) _add_logs(final_path, logs) diff --git a/tests/integration/variant_blueprint/test_variant_manager.py b/tests/integration/variant_blueprint/test_variant_manager.py index 79db7a88cb..ba618e993e 100644 --- a/tests/integration/variant_blueprint/test_variant_manager.py +++ b/tests/integration/variant_blueprint/test_variant_manager.py @@ -49,9 +49,7 @@ def variant_id_fixture(client: TestClient, admin_access_token: str, base_study_i @pytest.fixture(name="generate_snapshots") -def generate_snapshot_fixture( - client: TestClient, admin_access_token: str, base_study_id: str, monkeypatch: pytest.MonkeyPatch -) -> t.List[str]: +def generate_snapshot_fixture(client: TestClient, admin_access_token: str, base_study_id: str) -> list[str]: """Generate some snapshots with different date of update and last access""" # Initialize variant_ids list @@ -372,11 +370,7 @@ def test_outputs(client: TestClient, admin_access_token: str, variant_id: str, t def test_clear_snapshots( - client: TestClient, - admin_access_token: str, - tmp_path: Path, - generate_snapshots: t.List[str], - monkeypatch: pytest.MonkeyPatch, + client: TestClient, admin_access_token: str, tmp_path: Path, generate_snapshots: list[str] ) -> None: """ The `snapshot/` directory must not exist after a call to `clear-snapshot`. diff --git a/tests/launcher/test_service.py b/tests/launcher/test_service.py index 1c9bf9a68e..981adf385e 100644 --- a/tests/launcher/test_service.py +++ b/tests/launcher/test_service.py @@ -13,9 +13,10 @@ import json import os import time +import uuid from datetime import datetime, timedelta, timezone from pathlib import Path -from unittest.mock import Mock, call +from unittest.mock import Mock, call, patch from uuid import uuid4 from zipfile import ZIP_DEFLATED, ZipFile @@ -39,7 +40,7 @@ from antarest.core.model import PermissionInfo, PublicMode from antarest.core.requests import UserHasNotPermissionError from antarest.core.utils.fastapi_sqlalchemy import DBSessionMiddleware -from antarest.core.utils.fastapi_sqlalchemy.middleware import init_db_singleton +from antarest.core.utils.fastapi_sqlalchemy.middleware import db, init_db_singleton from antarest.core.utils.utils import current_time from antarest.dbmodel import Base from antarest.launcher.adapters.abstractlauncher import SimulationLogs @@ -70,6 +71,7 @@ from antarest.login.utils import current_user_context, get_current_user from antarest.output.service import OutputService from antarest.study.model import ( + DEFAULT_WORKSPACE_NAME, STUDY_VERSION_8_8, STUDY_VERSION_9_2, OwnerInfo, @@ -83,7 +85,7 @@ from antarest.study.storage.variantstudy.command_factory import CommandFactory from antarest.study.storage.variantstudy.model.command_context import CommandContext from antarest.study.storage.variantstudy.variant_study_service import VariantStudyService -from tests.helpers import with_admin_user +from tests.helpers import create_raw_study, with_admin_user, with_db_context class TestLauncherService: @@ -776,6 +778,7 @@ def test_get_logs(self, tmp_path: Path) -> None: ) @with_admin_user + @with_db_context def test_manage_output(self, tmp_path: Path) -> None: # TODO: finish adaptation study_service = Mock() @@ -812,7 +815,12 @@ def test_manage_output(self, tmp_path: Path) -> None: output_data.writestr("some output", "0\n1") job_id = "job_id" zipped_job_id = "zipped_job_id" - study_id = "study_id" + study_id = str(uuid.uuid4()) + # Adds the study linked to the job inside DB + study = create_raw_study(study_id, "study-test", tmp_path) + db.session.add(study) + db.session.commit() + # Defines the side effects launcher_service.job_result_repository.get.side_effect = [ None, JobResult(id=job_id, study_id=study_id), @@ -1002,6 +1010,7 @@ def test_save_solver_stats(self, tmp_path: Path) -> None: ) assert actual_obj.to_dto().model_dump() == expected_obj.to_dto().model_dump() + @with_db_context def test_import_output_is_called_with_the_right_user(self, tmp_path: Path) -> None: # Create user jwt_user = JWTUser(id=2, impersonator=2, type="users") @@ -1009,9 +1018,14 @@ def test_import_output_is_called_with_the_right_user(self, tmp_path: Path) -> No login_service = Mock() login_service.get_jwt.return_value = jwt_user # Put this user as the job owner - job_result = JobResult(study_id="study_id", owner_id=jwt_user.id) + study_id = str(uuid.uuid4()) + job_result = JobResult(study_id=study_id, owner_id=jwt_user.id) job_repository = Mock() job_repository.get.return_value = job_result + # Adds the study linked to the job inside DB + study = create_raw_study(study_id, "study-test", tmp_path) + db.session.add(study) + db.session.commit() # fake import_output function that checks the current user def fake_import_output( @@ -1039,6 +1053,7 @@ def fake_import_output( # Ensures the output_service.import_output method was called with the right user launcher_service._import_output("job_id", tmp_path, SimulationLogs.no_logs()) + launcher_service.output_service.import_output.assert_called_once() @with_admin_user def test_run_study_with_solver_presets(self) -> None: @@ -1140,6 +1155,60 @@ def get_mock_solver_presets(solver_presets_id: str): with pytest.raises(IncompatibleSolverPresets): launcher_service.run_study("study_uuid", "local", params_with_other_options, "config-1", "8.0") + @with_db_context + @pytest.mark.parametrize("managed", [True, False]) + def test_import_output_for_different_workspaces(self, tmp_path: Path, managed: bool) -> None: + ########################## + # Set Up + ########################## + + # Create a study in DB + study_id = str(uuid.uuid4()) + study = create_raw_study(study_id, "study-test", path="") + if managed: + study.workspace = DEFAULT_WORKSPACE_NAME + else: + study.workspace = "other-workspace" + db.session.add(study) + db.session.commit() + + # Create a fake job + job_result = JobResult(study_id=study_id, owner_id=1) + job_repository = Mock() + job_repository.get.return_value = job_result + + # Builds the service + output_service = Mock() + output_service.import_output.side_effect = None + launcher_service = LauncherService( + config=Mock(), + study_service=Mock(), + output_service=output_service, + login_service=Mock(), + job_result_repository=job_repository, + solver_presets_repository=Mock(), + event_bus=Mock(), + factory_launcher=Mock(), + file_transfer_manager=Mock(), + task_service=Mock(), + cache=Mock(), + ) + + ########################## + # Test + ########################## + + # We patch the `archive_dir` function to always raise. + # This way we can check if it was called or not. + with patch("antarest.launcher.service.archive_dir", side_effect=ValueError("Output archiving failed for test")): + if managed: + # We should not raise here as we do not need to archive the output. + launcher_service._import_output("job_id", tmp_path, SimulationLogs.no_logs()) + else: + # Here we expect the method to raise as we need to archive the output in order to unarchive it later on the Windows VM. + with pytest.raises(ValueError, match="Output archiving failed for test"): + launcher_service._import_output("job_id", tmp_path, SimulationLogs.no_logs()) + class TestNormalizeScheduledAt: NOW = datetime(2026, 7, 7, 12, 0, 0) diff --git a/tests/output/storage/file/test_file_output_storage.py b/tests/output/storage/file/test_file_output_storage.py index 3514fd0378..04509d6843 100644 --- a/tests/output/storage/file/test_file_output_storage.py +++ b/tests/output/storage/file/test_file_output_storage.py @@ -547,9 +547,7 @@ def test_import_output_directory(output_storage: IOutputStorage, tmp_path: Path) assert output_storage.get_logs("my-study", f"{expected_date}eco-other", LogType.STDERR) == "some error" -def test_import_output_zip_should_import_it_as_archived( - output_storage: IOutputStorage, tmp_path: Path, sta_mini_zip_path: Path -) -> None: +def test_import_output_zip_should_import_it_as_archived(output_storage: IOutputStorage, tmp_path: Path) -> None: # Checks the "optimized path" for zipped outputs, see TODOs # Use the `20201014-1430adq-2` output as it's already zipped @@ -564,6 +562,10 @@ def test_import_output_zip_should_import_it_as_archived( zip_path = outputs_dir / "STA-mini" / "20201014-1430adq-2.zip" in_study = False + # Copies the `zip_path` as `import_output` cleans it. + second_zip_path = tmp_path / "output_copied.zip" + shutil.copy(zip_path, second_zip_path) + # Import zip file output_id = output_storage.import_output("my-study", zip_path) @@ -581,7 +583,7 @@ def test_import_output_zip_should_import_it_as_archived( err_logs = tmp_path / "err.log" err_logs.write_text("some error") output_id = output_storage.import_output( - "my-study", zip_path, output_name_suffix="other", logs=SimulationLogs(out_logs, err_logs) + "my-study", second_zip_path, output_name_suffix="other", logs=SimulationLogs(out_logs, err_logs) ) assert output_id == f"{expected_date}adq-other" assert output_storage.list_outputs("my-study") == [ diff --git a/tests/study/storage/variantstudy/test_variant_study_service.py b/tests/study/storage/variantstudy/test_variant_study_service.py index 5708dc7362..40cbcd105c 100644 --- a/tests/study/storage/variantstudy/test_variant_study_service.py +++ b/tests/study/storage/variantstudy/test_variant_study_service.py @@ -220,7 +220,6 @@ def test_clear_all_snapshots( variant_study_service: VariantStudyService, raw_study_service: RawStudyService, fs_dao: FileStudyTreeDao, - monkeypatch: pytest.MonkeyPatch, ) -> None: """ - Test return value in case the user is not allowed to call the function,