fix: fix unit tests for multi-VO handling and address comments from a review.

Author: martynia

src/DIRAC/WorkloadManagementSystem/Agent/PilotLoggingAgent.py

@@ -29,7 +29,7 @@ class PilotLoggingAgent(AgentModule):
 
     def initialize(self):
         """
-        agent's initalisation. Use this agent's CS information to:
+        agent's initialisation. Use this agent's CS information to:
         Determine what Defaults/Shifter shifter proxy to use.,
         get the target SE name from the CS.
         Obtain log file location from Tornado.
@@ -40,7 +40,7 @@ def initialize(self):
         # configured VOs and setup
         res = getVOs()
         if not res["OK"]:
-            return S_ERROR(res["Message"])
+            return res
         self.voList = res.get("Value", [])
 
         if isinstance(self.voList, str):
@@ -93,7 +93,9 @@ def execute(self):
     @executeWithUserProxy
     def executeForVO(self, vo):
         """
-        Execute one agent cycle for VO
+        Execute one agent cycle for a VO. It obtains VO-specific configuration pilot options from the CS:
+        UploadPath - the path where the VO wants to upload pilot logs. It has to start with a VO name (/vo/path).
+        UploadSE - Storage element where the logs will be kept.
 
         :param str vo: vo enabled for remote pilot logging
         :return: S_OK or S_ERROR
@@ -110,14 +112,13 @@ def executeForVO(self, vo):
         uploadPath = pilotOptions.get("UploadPath")
         if uploadPath is None:
             return S_ERROR(f"Upload path on SE {uploadSE} not defined")
-        uploadPath = os.path.join("/", vo, uploadPath)
         self.log.info(f"Pilot upload path: {uploadPath}")
 
         client = TornadoPilotLoggingClient(useCertificates=True)
         resDict = client.getMetadata()
 
         if not resDict["OK"]:
-            return resDict["Message"]
+            return resDict
 
         # vo-specific source log path:
         pilotLogPath = os.path.join(resDict["Value"]["LogPath"], vo)
@@ -141,7 +142,7 @@ def executeForVO(self, vo):
             if not res["OK"]:
                 self.log.error("Could not upload", f"to {uploadSE}: {res['Message']}")
             else:
-                self.log.info("File uploaded: ", f"LFN = {res['Value']}")
+                self.log.verbose("File uploaded: ", f"LFN = {res['Value']}")
                 try:
                     os.remove(name)
                 except Exception as excp:

src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PilotLoggingAgent.py

@@ -1,7 +1,7 @@
 import unittest
 import os
 from contextlib import contextmanager
-from mock import MagicMock, patch, call
+from unittest.mock import MagicMock, patch, call
 
 from DIRAC import gLogger, gConfig, S_OK, S_ERROR
 import DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent
@@ -69,8 +69,8 @@ def test_execute(self, mockExVO, mockSetup, mockVOs, mockOp):
             res = mAgent.execute()
             assert mockOp.called
             mockOp.assert_called_with(vo=vo, setup=mockSetup.return_value)
-            mockOp.return_value.getValue.assert_called_with("/Services/JobMonitoring/usePilotsLoggingFlag", False)
-            mockOp.return_value.getOptionsDict.assert_called_with("/Shifter/DataManager")
+            mockOp.return_value.getValue.assert_called_with("/Pilot/RemoteLogging", True)
+            mockOp.return_value.getOptionsDict.assert_called_with("Shifter/DataManager")
             mAgent.executeForVO.assert_called_with(
                 vo,
                 proxyUserName=upDict["Value"]["User"],
@@ -101,44 +101,52 @@ def test_execute(self, mockExVO, mockSetup, mockVOs, mockOp):
                 self.assertIn(vo, res)
                 self.assertIsNotNone(res[vo])
 
+    @patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.Operations")
     @patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.DataManager")
     @patch.object(DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.os, "listdir")
     @patch.object(DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.os, "remove")
     @patch.object(DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.os.path, "isfile")
+    @patch.object(DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.os.path, "exists")
     @patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.TornadoPilotLoggingClient")
     @patch.object(DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent, "executeWithUserProxy")
-    def test_executeForVO(self, mockProxy, mockTC, mockisfile, mockremove, mocklistdir, mockDM):
+    def test_executeForVO(self, mockProxy, mockTC, mockexists, mockisfile, mockremove, mocklistdir, mockDM, mockOp):
         with patch_parent(PilotLoggingAgent) as MockAgent:
+
+            opsHelperValues = {"UploadSE": "testUploadSE", "UploadPath": "/gridpp/uploadPath"}
             mAgent = MockAgent()
-            opsHelperValues = ["uploadSE", "/uploadPath", "tornadoServer"]
-            mAgent.opsHelper.getValue.side_effect = opsHelperValues
+            mockOp.return_value.getOptionsDict.return_value = opsHelperValues
+            mAgent.opsHelper = mockOp.return_value
             mockisfile.return_value = True
             mocklistdir.return_value = ["file1.log", "file2.log", "file3.log"]
             resDict = {"OK": True, "Value": {"LogPath": "/pilot/log/path/"}}
             mockTC.return_value.getMetadata.return_value = resDict
+            vo = "gridpp"
 
             # success route
-            res = mAgent.executeForVO(vo="gridpp")
+            res = mAgent.executeForVO(vo=vo)
 
-            mockTC.assert_called_with("tornadoServer", useCertificates=True)
+            mockTC.assert_called_with(useCertificates=True)
             assert mockTC.return_value.getMetadata.called
-            mocklistdir.assert_called_with(resDict["Value"]["LogPath"])
+            mockexists.return_value = True
+            mocklistdir.assert_called_with(os.path.join(resDict["Value"]["LogPath"], vo))
 
             calls = []
             for elem in mocklistdir.return_value:
-                calls.append(call(os.path.join(resDict["Value"]["LogPath"], elem)))
-            mockisfile.has_calls(calls)
+                calls.append(call(os.path.join(resDict["Value"]["LogPath"], vo, elem)))
+            mockisfile.assert_has_calls(calls)
 
             assert mockDM.called
             mockDM.return_value.putAndRegister.return_value = {"OK": True}
 
             calls = []
+            mockDM.return_value.putAndRegister.assert_called()
             for elem in mocklistdir.return_value:
-                lfn = opsHelperValues[1] + elem
-                name = resDict["Value"]["LogPath"] + elem
-                calls.append(call(lfn=lfn, fileName=name, diracSE=opsHelperValues[0], overwrite=True))
+                lfn = os.path.join(opsHelperValues["UploadPath"], elem)
+                name = os.path.join(resDict["Value"]["LogPath"], vo, elem)
+                mockDM.return_value.putAndRegister.assert_any_call(
+                    lfn=lfn, fileName=name, diracSE=opsHelperValues["UploadSE"], overwrite=True
+                )
 
-            mockDM.return_value.putAndRegister.has_calls(calls)
             call_count = len(mocklistdir.return_value)
             self.assertEqual(call_count, mockDM.return_value.putAndRegister.call_count)
             self.assertEqual(call_count, mockremove.call_count)
@@ -151,26 +159,27 @@ def test_executeForVO(self, mockProxy, mockTC, mockisfile, mockremove, mocklistd
             mAgent.opsHelper.getValue.side_effect = opsHelperValues
             mockTC.reset_mock(return_value=True)
             mockTC.return_value.getMetadata.return_value = {"OK": False, "Message": "Failed, sorry.."}
-            res = mAgent.executeForVO(vo="gridpp")
+            res = mAgent.executeForVO(vo=vo)
             self.assertFalse(res["OK"])
 
             # config values not correct:
             opsHelperValues = [None, "/uploadPath", "tornadoServer"]
             mAgent.opsHelper.getValue.side_effect = opsHelperValues
-            res = mAgent.executeForVO(vo="gridpp")
+            res = mAgent.executeForVO(vo=vo)
             self.assertFalse(res["OK"])
 
             opsHelperValues = ["uploadSE", None, "tornadoServer"]
             mAgent.opsHelper.getValue.side_effect = opsHelperValues
-            res = mAgent.executeForVO(vo="gridpp")
+            res = mAgent.executeForVO(vo=vo)
             self.assertFalse(res["OK"])
 
             opsHelperValues = ["uploadSE", "uploadPath", None]
             mAgent.opsHelper.getValue.side_effect = opsHelperValues
-            res = mAgent.executeForVO(vo="gridpp")
+            res = mAgent.executeForVO(vo=vo)
             self.assertFalse(res["OK"])
 
 
 if __name__ == "__main__":
-    suite = unittest.defaultTestLoader.loadTestsFromTestCase(PilotAgentTestCase)
-    testResult = unittest.TextTestResult(verbosity=2).run(suite)
+    suite = unittest.defaultTestLoader.loadTestsFromTestCase(PilotLoggingAgentTestCase)
+    runner = unittest.TextTestRunner(verbosity=2)
+    runner.run(suite)

src/DIRAC/WorkloadManagementSystem/Client/TornadoPilotLoggingClient.py

@@ -10,20 +10,8 @@ def __init__(self, url=None, **kwargs):
         :param str url: Server URL, if None, defaults to "WorkloadManagement/TornadoPilotLogging"
         :param dict kwargs: additional keyword arguments, currently unused.
         """
-        super(TornadoPilotLoggingClient, self).__init__(**kwargs)
+        super().__init__(**kwargs)
         if not url:
             self.serverURL = "WorkloadManagement/TornadoPilotLogging"
         else:
             self.serverURL = url
-
-    def getMetadata(self):
-        """
-        Get metadata from the server.
-
-        :return: Dirac S_OK/S_ERROR dictionary with server properties.
-        :rtype: dict
-        """
-
-        retVal = self._getRPC().getMetadata()
-
-        return retVal

src/DIRAC/WorkloadManagementSystem/Client/test/Test_TornadoPilotLoggingClient.py

@@ -6,12 +6,11 @@
 
 
 class TestTornadoPilotLoggingClient(unittest.TestCase):
-    @patch.object(tplc.TornadoPilotLoggingClient, "executeRPC", return_value={"key": "value"})
-    def test_client(self, clientMock):
+    def test_client(self):
         client = tplc.TornadoPilotLoggingClient("test.server", useCertificates=True)
-        res = client.getMetadata()
-        clientMock.assert_called_with("getMetadata")
-        self.assertEqual(res, clientMock.return_value)
+        self.assertEqual(client.serverURL, "test.server")
+        client = tplc.TornadoPilotLoggingClient(useCertificates=True)
+        self.assertEqual(client.serverURL, "WorkloadManagement/TornadoPilotLogging")
 
 
 if __name__ == "__main__":

src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg

@@ -26,7 +26,6 @@ Services
     {
       Default = authenticated
       sendMessage = "Operator"
-      sendMessage += "Pilot"
       sendMessage += "GenericPilot"
       getMetadata = "Operator"
       getMetadata += "TrustedHost"
@@ -194,7 +193,6 @@ Agents
   PilotLoggingAgent
   {
     PollingTime = 600
-    LogLevel = DEBUG
   }
   ##END
   JobAgent

src/DIRAC/WorkloadManagementSystem/Service/BasicPilotLoggingPlugin.py

@@ -21,8 +21,7 @@ def sendMessage(self, message, UUID, vo):
         """
         Dummy sendMessage method.
 
-        :param message: text to log
-        :type message: str
+        :param str message: text to log
         :return: None
         :rtype: None
         """

src/DIRAC/WorkloadManagementSystem/Service/FileCacheLoggingPlugin.py

@@ -1,7 +1,9 @@
 """
 File cache logging plugin.
 """
-import os, json, re
+import os
+import json
+import re
 from DIRAC import S_OK, S_ERROR, gLogger
 
 sLog = gLogger.getSubLogger(__name__)
@@ -62,10 +64,10 @@ def sendMessage(self, message, pilotUUID, vo):
                     else:
                         # it could be a string, if emitted by pilot logger StringIO handler
                         pilotLog.write(messageContent)
-                except IOError as ioerr:
-                    sLog.error("Error writing to log file:", str(ioerr))
-                    return S_ERROR(str(ioerr))
-        except IOError as err:
+                except OSError as oserr:
+                    sLog.error("Error writing to log file:", str(oserr))
+                    return S_ERROR(str(oserr))
+        except OSError as err:
             sLog.exception("Error opening a pilot log file:", str(err), lException=err)
             return S_ERROR(str(err))
         return S_OK(f"Message logged successfully for pilot: {pilotUUID} and {vo}")

src/DIRAC/WorkloadManagementSystem/Service/TornadoPilotLoggingHandler.py

@@ -28,9 +28,9 @@ def initializeHandler(cls, infoDict):
         defaultOption, defaultClass = "LoggingPlugin", "BasicPilotLoggingPlugin"
         configValue = getServiceOption(infoDict, defaultOption, defaultClass)
 
-        result = ObjectLoader().loadObject("WorkloadManagementSystem.Service.%s" % (configValue,), configValue)
+        result = ObjectLoader().loadObject(f"WorkloadManagementSystem.Service.{configValue}", configValue)
         if not result["OK"]:
-            cls.log.error("Failed to load LoggingPlugin", "%s: %s" % (configValue, result["Message"]))
+            cls.log.error("Failed to load LoggingPlugin", "{}: {}".format(configValue, result["Message"]))
             return result
 
         componentClass = result["Value"]
@@ -44,15 +44,6 @@ def initializeHandler(cls, infoDict):
             os.makedirs(logPath)
         cls.log.info("Pilot logging directory:", logPath)
 
-    def initializeRequest(self):
-        """
-        Called for each request.
-
-        :return: None
-        """
-
-        self.log.info("Request initialised.. ")
-
     def export_sendMessage(self, message, pilotUUID):
         # def export_sendMessage(self, message, pilotUUID):
         """

src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_TornadoPilotLoggingHandler.py

@@ -1,5 +1,5 @@
 import unittest
-from mock import patch
+from unittest.mock import patch
 import os
 import json
 import tempfile
@@ -84,26 +84,27 @@ def test_FileCachePlugin(self, mockExists, mockGetcwd, mockMakedirs):
             + "2022-02-23 13:48:36.123456 UTC DEBUG [PilotParams] JSON file analysed: pilot.json"
         )
         messageJSON = json.dumps(messsageText)
+        vo = "anyVO"
         pilotUUID = "78f39a90-2073-11ec-98d7-b496913c0cf4"
 
         # use a temporary dir, not the one above. Plugin will create the file to write into.
         with tempfile.TemporaryDirectory(suffix="pilottests") as d:
             plugin.meta["LogPath"] = d
-            res = plugin.sendMessage(messageJSON, pilotUUID)
+            res = plugin.sendMessage(messageJSON, pilotUUID, vo)
             self.assertTrue(res["OK"])
-            with open(os.path.join(d, pilotUUID), "r") as pilotLog:
+            with open(os.path.join(d, vo, pilotUUID)) as pilotLog:
                 content = pilotLog.read()
                 self.assertEqual(content, messsageText)
 
         # failures ?
         with tempfile.TemporaryDirectory(suffix="pilottests") as d:
             plugin.meta["LogPath"] = d
             os.chmod(d, 0o0000)
-            res = plugin.sendMessage(messageJSON, pilotUUID)
+            res = plugin.sendMessage(messageJSON, pilotUUID, vo)
             self.assertFalse(res["OK"])
 
         pilotUUID = "whatever"
-        res = plugin.sendMessage(messageJSON, pilotUUID)
+        res = plugin.sendMessage(messageJSON, pilotUUID, vo)
         self.assertFalse(res["OK"])
 
     @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os.path, "exists")
@@ -124,25 +125,28 @@ def test_finaliseLogs(self, mockGetcwd, mockExists):
         mockExists.return_value = True  # will not create a file
         mockGetcwd.return_value = "/tornado/document/root"  # so we have a path defined (will overwrite it below)
         plugin = FileCacheLoggingPlugin()
+        vo = "anyVO"
 
         with tempfile.TemporaryDirectory(suffix="pilottests") as d:
             plugin.meta["LogPath"] = d
             payload = '{"retCode": 0}'
             logfile = "78f39a90-2073-11ec-98d7-b496913c0cf4"  # == pilotUUID
+            os.mkdir(os.path.join(d, vo))  # vo specific directory, normally created by sending the first message
             # will fail here...
-            res = plugin.finaliseLogs(payload, logfile)
+            res = plugin.finaliseLogs(payload, logfile, vo)
             self.assertFalse(res["OK"])
             # create a file ..
-            with open(os.path.join(d, logfile), "w") as f:
+            with open(os.path.join(d, vo, logfile), "w") as f:
                 f.write("Create a dummy logfile")
-            res = plugin.finaliseLogs(payload, logfile)
+            res = plugin.finaliseLogs(payload, logfile, vo)
             self.assertTrue(res["OK"])
 
             logfile = "invalid!"
-            res = plugin.finaliseLogs(payload, logfile)
+            res = plugin.finaliseLogs(payload, logfile, vo)
             self.assertFalse(res["OK"])
 
 
 if __name__ == "__main__":
     suite = unittest.defaultTestLoader.loadTestsFromTestCase(TornadoPilotLoggingHandlerTestCase)
-    testResult = unittest.TextTestResult(verbosity=2).run(suite)
+    runner = unittest.TextTestRunner(verbosity=2)
+    runner.run(suite)