Merge pull request #615 from oda-hub/assign-session-logger-scratch_dir

Assign session logger scratch dir

cdci_data_analysis/flask_app/dispatcher_query.py

@@ -6,6 +6,7 @@
 @author: andrea tramcere
 """
 import os
+import pathlib
 import time
 from builtins import (open, str, range,
                       object)
@@ -638,10 +639,16 @@ def set_session_logger(self, scratch_dir, verbose=False, config=None):
 
         have_handler = False
         for handler in logger.handlers:
-            if isinstance(handler, logging.FileHandler):
-                logger.info("found FileHandler: %s : %s",
-                            handler, handler.baseFilename)
-                have_handler = True
+            if isinstance(handler, logging.FileHandler) and handler.baseFilename:
+                handler_path = pathlib.Path(handler.baseFilename)
+                if handler_path.parent.stem == scratch_dir:
+                    logger.info("found correspondent FileHandler: %s : %s",
+                                handler, handler.baseFilename)
+                    have_handler = True
+                else:
+                    logger.info("found not correspondent FileHandler: %s : %s, assigning a new one",
+                                handler, handler.baseFilename)
+                    logger.removeHandler(handler)
                 #handler.baseFilename == session_log_filename
 
         if not have_handler:

tests/test_job_management.py

@@ -1601,7 +1601,7 @@ def test_email_very_long_request_url(dispatcher_long_living_fixture,
     #  * make this or some other kind of URL shortener
 
     server = dispatcher_long_living_fixture
-
+    DataServerQuery.set_status('submitted')
     DispatcherJobState.remove_scratch_folders()
 
      # let's generate a valid token with high threshold
@@ -2655,3 +2655,95 @@ def test_incident_report(dispatcher_live_fixture, dispatcher_local_mail_server,
             incident_report_str=incident_content,
             decoded_token=decoded_token
         )
+
+
+@pytest.mark.not_safe_parallel
+def test_session_log(dispatcher_live_fixture):
+    server = dispatcher_live_fixture
+
+    DispatcherJobState.remove_scratch_folders()
+
+    token_payload = {**default_token_payload
+                     }
+
+    encoded_token = jwt.encode(token_payload, secret_key, algorithm='HS256')
+
+    dict_param = dict(
+        query_status="new",
+        query_type="Real",
+        instrument="empty-async",
+        product_type="dummy",
+        p=15,
+        token=encoded_token
+    )
+
+    # this should return status submitted, so matrix message sent
+    c = requests.get(os.path.join(server, "run_analysis"),
+                     dict_param
+                     )
+    assert c.status_code == 200
+    jdata = c.json()
+
+    session_id = jdata['session_id']
+    job_id = jdata['job_monitor']['job_id']
+    scratch_dir_fn = f'scratch_sid_{session_id}_jid_{job_id}'
+    session_log_fn = os.path.join(scratch_dir_fn, 'session.log')
+    dispatcher_job_state = DispatcherJobState.from_run_analysis_response(jdata)
+
+    assert os.path.exists(session_log_fn)
+
+    with open(session_log_fn) as session_log_fn_f:
+        session_log_content = session_log_fn_f.read()
+
+    assert '==============================> run query <==============================' in session_log_content
+    assert "'p': '15'," in session_log_content
+
+    time_request = jdata['time_request']
+
+    requests.get(os.path.join(server, "call_back"),
+                 params=dict(
+                     job_id=dispatcher_job_state.job_id,
+                     session_id=dispatcher_job_state.session_id,
+                     instrument_name="empty-async",
+                     action='progress',
+                     node_id='node_0',
+                     message='progressing',
+                     token=encoded_token,
+                     time_original_request=time_request
+                 ))
+
+    with open(session_log_fn) as session_log_fn_f:
+        session_log_content = session_log_fn_f.read()
+
+    assert '.run_call_back with args ' in session_log_content
+    assert "'p': '15'," in session_log_content
+
+    # second run_analysis within the same running session, but resulting a different scratch_dir and therefore session_log
+    dict_param = dict(
+        query_status="new",
+        query_type="Real",
+        instrument="empty-async",
+        product_type="dummy",
+        p=35,
+        token=encoded_token
+    )
+
+    c = requests.get(os.path.join(server, "run_analysis"),
+                     dict_param
+                     )
+    assert c.status_code == 200
+    jdata = c.json()
+
+    session_id = jdata['session_id']
+    job_id = jdata['job_monitor']['job_id']
+    scratch_dir_fn = f'scratch_sid_{session_id}_jid_{job_id}'
+    session_log_fn = os.path.join(scratch_dir_fn, 'session.log')
+
+    assert os.path.exists(session_log_fn)
+
+    with open(session_log_fn) as session_log_fn_f:
+        session_log_content = session_log_fn_f.read()
+
+    assert '==============================> run query <==============================' in session_log_content
+    assert "'p': '35'," in session_log_content
+    assert "'p': '15'," not in session_log_content