fix(launcher-api): remove orphan JobResults visibility permissions

antarest/launcher/service.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import shutil
-from datetime import datetime, timedelta
+from datetime import datetime
 from http import HTTPStatus
 from pathlib import Path
 from typing import Dict, List, Optional, cast
@@ -59,7 +59,6 @@ def __init__(self, engine: str):
         )
 
 
-ORPHAN_JOBS_VISIBILITY_THRESHOLD = 10  # days
 LAUNCHER_PARAM_NAME_SUFFIX = "output_suffix"
 EXECUTION_INFO_FILE = "execution_info.ini"
 
@@ -305,7 +304,6 @@ def _filter_from_user_permission(self, job_results: List[JobResult], user: Optio
         if not user:
             return []
 
-        orphan_visibility_threshold = datetime.utcnow() - timedelta(days=ORPHAN_JOBS_VISIBILITY_THRESHOLD)
         allowed_job_results = []
 
         study_ids = [job_result.study_id for job_result in job_results]
@@ -330,9 +328,7 @@ def _filter_from_user_permission(self, job_results: List[JobResult], user: Optio
                     raising=False,
                 ):
                     allowed_job_results.append(job_result)
-            elif (
-                user and (user.is_site_admin() or user.is_admin_token())
-            ) or job_result.creation_date >= orphan_visibility_threshold:
+            elif user and (user.is_site_admin() or user.is_admin_token()):
                 allowed_job_results.append(job_result)
         return allowed_job_results
 

tests/integration/launcher_blueprint/test_launcher_local.py

@@ -122,3 +122,44 @@ def test_get_launcher_time_limit(
             "description": "Unknown solver configuration: 'unknown'",
             "exception": "UnknownSolverConfig",
         }
+
+    def test_jobs_permissions(
+        self,
+        client: TestClient,
+        user_access_token: str,
+        admin_access_token: str,
+    ) -> None:
+        # create an admin study with no permissions
+        res = client.post(
+            "/v1/studies",
+            headers={"Authorization": f"Bearer {admin_access_token}"},
+            params={"name": "study_admin"},
+        )
+        res.raise_for_status()
+        # get the study_id
+        study_id = res.json()
+
+        # launch a job with the admin user
+        res = client.post(
+            f"/v1/launcher/run/{study_id}",
+            headers={"Authorization": f"Bearer {admin_access_token}"},
+            json={"launcher": "local"},
+        )
+        res.raise_for_status()
+        job_id = res.json()["job_id"]
+
+        # check that the user cannot see the job
+        res = client.get(
+            "/v1/launcher/jobs",
+            headers={"Authorization": f"Bearer {user_access_token}"},
+        )
+        res.raise_for_status()
+        assert job_id not in [job.get("id") for job in res.json()]
+
+        # check that the admin can see the job
+        res = client.get(
+            "/v1/launcher/jobs",
+            headers={"Authorization": f"Bearer {admin_access_token}"},
+        )
+        res.raise_for_status()
+        assert job_id in [job.get("id") for job in res.json()]

tests/launcher/test_service.py

@@ -40,13 +40,7 @@
     LauncherParametersDTO,
     LogType,
 )
-from antarest.launcher.service import (
-    EXECUTION_INFO_FILE,
-    LAUNCHER_PARAM_NAME_SUFFIX,
-    ORPHAN_JOBS_VISIBILITY_THRESHOLD,
-    JobNotFound,
-    LauncherService,
-)
+from antarest.launcher.service import EXECUTION_INFO_FILE, LAUNCHER_PARAM_NAME_SUFFIX, JobNotFound, LauncherService
 from antarest.login.auth import Auth
 from antarest.login.model import Identity
 from antarest.study.model import OwnerInfo, PublicMode, Study, StudyMetadataDTO
@@ -257,7 +251,7 @@ def test_service_get_jobs_from_database(self, db_session) -> None:
                 job_status=JobStatus.SUCCESS,
                 msg="Hello, World!",
                 exit_code=0,
-                creation_date=now - timedelta(days=ORPHAN_JOBS_VISIBILITY_THRESHOLD + 1),
+                creation_date=now - timedelta(days=11),
                 owner=identity_instance,
             )
         ]
@@ -308,7 +302,7 @@ def test_service_get_jobs_from_database(self, db_session) -> None:
                     )
                 ),
             )
-            == returned_faked_execution_results
+            == []
         )
 
         with pytest.raises(UserHasNotPermissionError):