Pass activities and checksums to FTS3

Mover.py

@@ -48,6 +48,7 @@ def __init__(self, manager, config, fts_client, memory_log, filedesc):
         self.TransferTimeout = config.TransferTimeout
         self.TransferTime = None
         self.TransferStarted = None
+        self.Activity = config.Activity
         self.log("created for:", filedesc.Path)
 
     def __str__(self):
@@ -89,14 +90,29 @@ def do_run(self):
             self.failed("Metadata download failed: %s" % (output,))
             return
 
-        json_text = open(meta_tmp, "r").read()
+        with  open(meta_tmp, "r") is mtf:
+            json_text = mtf.read()
 
         self.debug("original metadata JSON: %s" % (json_text,))
 
+        adler32_checksum = ''
         try:    
             metadata = json.loads(json_text)
             if ("file_size" not in metadata and "size" not in metadata) or ("checksum" not in metadata and "checksums" not in metadata):
                 return self.failed("metadata does not include size or checksum")
+
+            if "checksum" in metadata:
+                adler32_checksum = metadata["checksum"]
+            if "checksums" in metadata:
+                adler32_checksum = metadata["checksums"]["adler32"]
+            if isinstance(adler32_checksum, list) and adler32_checksum:
+                 adler32_checksum = adler32_checksum[0]
+
+            if ':' in adler32_checksum:
+                ctype, value = adler32_checksum.split(':', 1)
+                assert ctype == "adler32"
+                adler32_checksum = value
+
         except: 
             self.log_record("metadata parsing error: %s" % (traceback.format_exc(),))
             self.log_record("metadata file contents -------\n%s\n----- end if metadata file contents -----" % (open(meta_tmp, "r").read(),))
@@ -114,7 +130,7 @@ def do_run(self):
         # transfer data
         #
         self.updateStatus("transferring data")
-        request = self.FTSClient.submit(self.FileSrcURL, self.FileDstURL)
+        request = self.FTSClient.submit(self.FileSrcURL, self.FileDstURL, adler32_checksum=adler32_checksum, activity=Activity)
         done = request.wait(self.TransferTimeout)
         if done:
             if request.Failed:
@@ -131,7 +147,7 @@ def do_run(self):
         # transfer metadata
         #
         self.updateStatus("transferring metadata")
-        request = self.FTSClient.submit(self.MetaSrcURL, self.MetaDstURL)
+        request = self.FTSClient.submit(self.MetaSrcURL, self.MetaDstURL, activity=activity)
         self.debug("Metadata transfer request submitted:", request.JobID)
         done = request.wait(self.TransferTimeout)
         if done:
@@ -214,6 +230,7 @@ def __init__(self, filename = None, envname = None):
         config.read(filename)
         self.Config = config
         #self.CopyTemplate = config.get("Mover", "CopyTemplate")
+        self.Activity = config.get("Mover", "Activity", "")
         self.DownloadTemplate = config.get("Mover", "DownloadTemplate")
         #self.UploadTemplate = config.get("Mover", "UploadTemplate")
         self.DeleteTemplate = config.get("Mover", "DeleteTemplate")

fts3client/client.py

@@ -85,7 +85,7 @@ def __init__(self, url_head, proxy_file):
         self.ProxyFile = proxy_file
         self.Context = Context(url_head, self.ProxyFile, None, verify=False)
 
-    def submit(self, src_url, dst_url, metadata = None, overwrite=True, **meta_args):
+    def submit(self, src_url, dst_url,   metadata = None, overwrite=True, adler32_checksum = '', activity = '', **meta_args):
         params = {"overwrite":overwrite}
         request = {
             "params": params,
@@ -96,6 +96,14 @@ def submit(self, src_url, dst_url, metadata = None, overwrite=True, **meta_args)
                 }
             ]    
         }
+
+        if adler32_checksum:
+            request["files"][0]["checksum"] = "ADLER32:" + adler32_checksum
+            request["params"]["verify_checksum"] = "target"
+
+        if activity:
+            request["files"][0]["activity"] = activity
+
         if metadata or meta_args:
             metadata = (metadata or {}).copy()
             metadata.update(meta_args)

mover.cfg

@@ -1,4 +1,5 @@
 [Mover]
+Activity = Production
 RemovePathPrefix = /
 DestinationURLPattern = https://dst_host/dst_root/$tail
 SourceURLPattern = https://src_host/src_root/$tail