Merge pull request #6511 from DIRACGridBot/cherry-pick-2-529778972-integration

[sweep:integration] Bug fixes, and minor style fixes for ES

Author: fstagni

src/DIRAC/Core/Security/VOMS.py

@@ -273,7 +273,6 @@ def setVOMSAttributes(self, proxy, attribute=None, vo=None):
         cmd += [f"{vo}:{attribute}" if attribute and attribute != "NoRole" else vo]
         cmd += ["-valid", f"{hours}:{mins}"]
         cmd += ["-bits", str(bitStrength)]
-        tmpDir = False
         vomsesPath = self.getVOMSESLocation()
         if vomsesPath:
             cmd += ["-vomses", vomsesPath]
@@ -283,8 +282,6 @@ def setVOMSAttributes(self, proxy, attribute=None, vo=None):
         cmd += ["-timeout", "12"]
 
         result = shellCall(self._secCmdTimeout, shlex.join(cmd))
-        if tmpDir:
-            shutil.rmtree(tmpDir)
 
         deleteMultiProxy(proxyDict)
 

src/DIRAC/Core/Utilities/ElasticSearchDB.py

@@ -3,25 +3,19 @@
 It is used to query Elasticsearch instances.
 """
 
-from datetime import datetime
-from datetime import timedelta
-from urllib import parse as urlparse
-
 import copy
 import functools
 import json
+from datetime import datetime, timedelta
+from urllib import parse as urlparse
 
 import certifi
 
 try:
-    from opensearch_dsl import Search, Q, A
+    from opensearch_dsl import A, Q, Search
     from opensearchpy import OpenSearch as Elasticsearch
-    from opensearchpy.exceptions import (
-        ConnectionError as ElasticConnectionError,
-        TransportError,
-        NotFoundError,
-        RequestError,
-    )
+    from opensearchpy.exceptions import ConnectionError as ElasticConnectionError
+    from opensearchpy.exceptions import NotFoundError, RequestError, TransportError
     from opensearchpy.helpers import BulkIndexError, bulk
 except ImportError:
     from elasticsearch_dsl import Search, Q, A
@@ -34,11 +28,10 @@
     )
     from elasticsearch.helpers import BulkIndexError, bulk
 
-from DIRAC import gLogger, S_OK, S_ERROR
+from DIRAC import S_ERROR, S_OK, gLogger
 from DIRAC.Core.Utilities import DErrno, TimeUtilities
 from DIRAC.FrameworkSystem.Client.BundleDeliveryClient import BundleDeliveryClient
 
-
 sLog = gLogger.getSubLogger(__name__)
 
 
@@ -49,9 +42,8 @@ def ifConnected(method):
     def wrapper_decorator(self, *args, **kwargs):
         if self._connected:
             return method(self, *args, **kwargs)
-        else:
-            sLog.error("Not connected")
-            return S_ERROR("Not connected")
+        sLog.error("Not connected")
+        return S_ERROR("Not connected")
 
     return wrapper_decorator
 
@@ -208,7 +200,7 @@ def getIndexPrefix(self):
         return self.__indexPrefix
 
     @ifConnected
-    def query(self, index, query):
+    def query(self, index: str, query):
         """Executes a query and returns its result (uses ES DSL language).
 
         :param self: self reference
@@ -223,18 +215,17 @@ def query(self, index, query):
             return S_ERROR(re)
 
     @ifConnected
-    def update(self, index, query=None, updateByQuery=True, id=None):
+    def update(self, index: str, query=None, updateByQuery: bool = True, docID: str = None):
         """Executes an update of a document, and returns S_OK/S_ERROR
 
-        :param self: self reference
-        :param str index: index name
-        :param dict query: It is the query in ElasticSearch DSL language
-        :param bool updateByQuery: A bool to determine update by query or index values using index function.
-        :param int id: ID for the document to be created.
+        :param index: index name
+        :param query: It is the query in ElasticSearch DSL language
+        :param updateByQuery: A bool to determine update by query or index values using index function.
+        :param docID: ID for the document to be created.
 
         """
 
-        sLog.debug(f"Updating {index} with {query}, updateByQuery={updateByQuery}, id={id}")
+        sLog.debug(f"Updating {index} with {query}, updateByQuery={updateByQuery}, docID={docID}")
 
         if not index or not query:
             return S_ERROR("Missing index or query")
@@ -243,64 +234,64 @@ def update(self, index, query=None, updateByQuery=True, id=None):
             if updateByQuery:
                 esDSLQueryResult = self.client.update_by_query(index=index, body=query)
             else:
-                esDSLQueryResult = self.client.index(index=index, body=query, id=id)
+                esDSLQueryResult = self.client.index(index=index, body=query, id=docID)
             return S_OK(esDSLQueryResult)
         except RequestError as re:
             return S_ERROR(re)
 
     @ifConnected
-    def getDoc(self, index: str, id: str) -> dict:
+    def getDoc(self, index: str, docID: str) -> dict:
         """Retrieves a document in an index.
 
         :param index: name of the index
-        :param id: document ID
+        :param docID: document ID
         """
-        sLog.debug(f"Retrieving document {id} in index {index}")
+        sLog.debug(f"Retrieving document {docID} in index {index}")
         try:
-            return S_OK(self.client.get(index, id)["_source"])
+            return S_OK(self.client.get(index, docID)["_source"])
         except NotFoundError:
             sLog.warn("Could not find the document in index", index)
             return S_OK({})
         except RequestError as re:
             return S_ERROR(re)
 
     @ifConnected
-    def updateDoc(self, index: str, id: str, body: dict) -> dict:
+    def updateDoc(self, index: str, docID: str, body) -> dict:
         """Update an existing document with a script or partial document
 
         :param index: name of the index
-        :param id: document ID
+        :param docID: document ID
         :param body: The request definition requires either `script` or
             partial `doc`
         """
-        sLog.debug(f"Updating document {id} in index {index}")
+        sLog.debug(f"Updating document {docID} in index {index}")
         try:
-            return S_OK(self.client.update(index, id, body))
+            return S_OK(self.client.update(index, docID, body))
         except RequestError as re:
             return S_ERROR(re)
 
     @ifConnected
-    def deleteDoc(self, index: str, id: str):
+    def deleteDoc(self, index: str, docID: str):
         """Deletes a document in an index.
 
         :param index: name of the index
-        :param id: document ID
+        :param docID: document ID
         """
-        sLog.debug(f"Deleting document {id} in index {index}")
+        sLog.debug(f"Deleting document {docID} in index {index}")
         try:
-            return S_OK(self.client.delete(index, id))
+            return S_OK(self.client.delete(index, docID))
         except RequestError as re:
             return S_ERROR(re)
 
     @ifConnected
-    def existsDoc(self, index: str, id: str) -> bool:
+    def existsDoc(self, index: str, docID: str) -> bool:
         """Returns information about whether a document exists in an index.
 
         :param index: name of the index
-        :param id: document ID
+        :param docID: document ID
         """
-        sLog.debug(f"Checking if document {id} in index {index} exists")
-        return self.client.exists(index, id)
+        sLog.debug(f"Checking if document {docID} in index {index} exists")
+        return self.client.exists(index, docID)
 
     @ifConnected
     def _Search(self, indexname):
@@ -329,9 +320,9 @@ def getIndexes(self, indexName=None):
         """
         if not indexName:
             indexName = self.__indexPrefix
-        sLog.debug("Getting indices alias of %s" % indexName)
+        sLog.debug(f"Getting indices alias of {indexName}")
         # we only return indexes which belong to a specific prefix for example 'lhcb-production' or 'dirac-production etc.
-        return list(self.client.indices.get_alias("%s*" % indexName))
+        return list(self.client.indices.get_alias(f"{indexName}*"))
 
     @ifConnected
     def getDocTypes(self, indexName):

src/DIRAC/Core/Utilities/Graphs/GraphData.py

@@ -570,17 +570,6 @@ def getPlotNumData(self):
 
         return zip(self.num_keys, self.values, self.errors)
 
-    def getPlotDataForKeys(self, keys):
-
-        result_pairs = []
-        for key in keys:
-            if key in self.parsed_data:
-                result_pairs.append(key, self.parsed_data[key], self.parsed_errors[key])
-            else:
-                result_pairs.append(key, None, 0.0)
-
-        return result_pairs
-
     def getPlotDataForNumKeys(self, num_keys, zeroes=False):
 
         result_pairs = []

src/DIRAC/Core/Workflow/Parameter.py

@@ -438,7 +438,7 @@ def linkUp(self, opt, prefix="", postfix="", objname="self"):
                 if par is None:
                     print("ERROR ParameterCollection.linkUp() can not find parameter with the name=%s" % (s))
                 else:
-                    par.link(objname, prefix + p.getName() + postfix)
+                    par.link(objname, prefix + par.getName() + postfix)
         elif isinstance(opt, str):
             par = self.find(opt)
             if par is None:
@@ -480,7 +480,7 @@ def unlink(self, opt):
         elif isinstance(opt, str):
             par = self.find(opt)
             if par is None:
-                print("ERROR ParameterCollection.unlink() can not find parameter with the name=%s" % (s))
+                print("ERROR ParameterCollection.unlink() can not find parameter with the name=%s" % (opt))
             else:
                 par.unlink()
         else:

src/DIRAC/DataManagementSystem/Agent/RequestOperations/test/Test_ArchiveFiles.py

@@ -163,10 +163,7 @@ def test_run_IgnoreMissingFiles(archiveFiles, _myMocker, listOfLFNs):
     )
     for index, opFile in enumerate(archiveFiles.operation):
         LOG.debug("%s", opFile)  # lazy evaluation of the argument!
-        if index == 5:
-            assert opFile.Status == "Done"
-        else:
-            assert opFile.Status == "Done"
+        assert opFile.Status == "Done"
 
 
 def test_checkFilePermissions(archiveFiles, _myMocker):

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/FileMetadata/FileMetadata.py

@@ -457,7 +457,7 @@ def __createMetaSelection(self, value):
                 result = self.db._escapeValues(value)
                 if not result["OK"]:
                     return result
-                query = "( $s )" % ", ".join(result["Value"])
+                query = "( %s )" % ", ".join(result["Value"])
                 queryList.append(("IN", query))
         elif isinstance(value, dict):
             for operation, operand in value.items():

src/DIRAC/DataManagementSystem/scripts/dirac_dms_directory_sync.py

@@ -273,7 +273,7 @@ def doUpload(fc, dm, result, source_dir, dest_dir, storage, delete, nthreads):
         if len(lfns) > 0:
             res = removeRemoteFiles(dm, lfns)
             if not res["OK"]:
-                gLogger.fatal("Deleting of files: " + lfns + " -X- [FAILED]" + res["Message"])
+                gLogger.fatal("Deleting of files: " + str(lfns) + " -X- [FAILED]" + res["Message"])
                 DIRAC.exit(1)
             else:
                 gLogger.notice("Deleting " + ", ".join(lfns) + " -> [DONE]")

src/DIRAC/FrameworkSystem/scripts/dirac_proxy_destroy.py

@@ -112,9 +112,6 @@ def deleteLocalProxy(proxyLoc):
     except OSError:
         gLogger.error("IOError: Failed to delete local proxy.")
         return
-    except OSError:
-        gLogger.error("OSError: Failed to delete local proxy.")
-        return
     gLogger.notice("Local proxy deleted.")
 
 

src/DIRAC/Interfaces/scripts/dirac_admin_sync_users_from_file.py

@@ -40,7 +40,7 @@ def main():
     try:
         usersCFG = CFG().loadFromFile(args[0])
     except Exception as e:
-        errorList.append("file open", f"Can't parse file {args[0]}: {str(e)}")
+        errorList.append(("file open", f"Can't parse file {args[0]}: {str(e)}"))
         errorCode = 1
     else:
         if not diracAdmin.csSyncUsersWithCFG(usersCFG):

src/DIRAC/Resources/Computing/BatchSystems/TimeLeft/LSFResourceUsage.py

@@ -186,8 +186,6 @@ def __init__(self):
                                 self.log.info("Host Normalization", f"{self.host}: {self.hostNorm}")
                             except ValueError as e:
                                 self.log.exception("Exception parsing lshosts output", l1, e)
-                            finally:
-                                break
 
             if self.hostNorm and self.normRef:
                 self.hostNorm /= self.normRef