Merge pull request #16 from ssl-hep/multipath-support

making file_path a list

Author: ivukotic

README.md

@@ -18,7 +18,7 @@ Create an async callback method that `yield`s file info dictionaries. For exampl
     async def my_callback(did_name: str, info: Dict[str, Any]):
         for i in range(0, 10):
             yield {
-                'file_path': f"root://atlas-experiment.cern.ch/dataset1/file{i}.root",
+                'paths': [f"root://atlas-experiment.cern.ch/dataset1/file{i}.root"],
                 'adler32': b183712731,
                 'file_size': 0,
                 'file_events': 0,
@@ -33,7 +33,7 @@ The arguments to the method are straight forward:
 
 Yield the results as you find them - ServiceX will actually start processing the files before your DID lookup is finished if you do this. The fields you need to pass back to the library are as follows:
 
-* `file_path`: A URI that a transformer in ServiceX can access to get at the file. Often these are either `root://` or `http://` schema URI's.
+* `paths`: An ordered list of URIs that a transformer in ServiceX can access to get at the file. Often these are either `root://` or `http://` schema URI's. When accessing the file, URIs will be tried in ordered listed.
 * `adler32`: A CRC number for the file. This CRC is calculated in a special way by rucio and is not used. Leave as 0 if you do not know it.
 * `file_size`: Number of bytes of the file. Used to calculate statistics. Leave as zero if you do not know it (or it is expensive to look up).
 * `file_events`: Number of events in the file. Used to calculate statistics. Leave as zero if you do not know it (or it is expensive to look up).
@@ -118,7 +118,7 @@ In the end, all DID finders for ServiceX will run under Kubernetes. ServiceX com
 
         for i in range(0, 10):
             yield {
-                'file_path': f"root://atlas-experiment.cern.ch/dataset1/file{i}.root",
+                'paths': [f"root://atlas-experiment.cern.ch/dataset1/file{i}.root"]
                 'adler32': b183712731,
                 'file_size': 0,
                 'file_events': 0,

src/servicex_did_finder_lib/servicex_adaptor.py

@@ -72,7 +72,7 @@ def put_file_add(self, file_info):
             try:
                 mesg = {
                     "timestamp": datetime.now().isoformat(),
-                    "file_path": self._prefix_file(file_info['file_path']),
+                    "paths": [self._prefix_file(fp) for fp in file_info['paths']],
                     'adler32': file_info['adler32'],
                     'file_size': file_info['file_size'],
                     'file_events': file_info['file_events']

tests/servicex_did_finder_lib/test_communication.py

@@ -102,7 +102,7 @@ async def my_callback(did_name: str, info: Dict[str, Any]):
         nonlocal seen_name
         seen_name = did_name
         yield {
-            'file_path': "fork/it/over",
+            'paths': ["fork/it/over"],
             'adler32': 'no clue',
             'file_size': 22323,
             'file_events': 0,
@@ -172,7 +172,7 @@ async def my_callback(did_name: str, info: Dict[str, Any]) \
         nonlocal called
         called = True
         yield {
-            'file_path': "fork/it/over",
+            'paths': ["fork/it/over"],
             'adler32': 'no clue',
             'file_size': 22323,
             'file_events': 0,
@@ -226,13 +226,13 @@ async def test_run_file_fetch_loop(SXAdaptor, mocker):
     async def my_user_callback(did, info):
         return_values = [
             {
-                'file_path': '/tmp/foo',
+                'paths': ['/tmp/foo'],
                 'adler32': '13e4f',
                 'file_size': 1024,
                 'file_events': 128
             },
             {
-                'file_path': '/tmp/bar',
+                'paths': ['/tmp/bar'],
                 'adler32': 'f33d',
                 'file_size': 2046,
                 'file_events': 64
@@ -245,8 +245,8 @@ async def my_user_callback(did, info):
     SXAdaptor.post_transform_start.assert_called_once()
 
     assert SXAdaptor.put_file_add.call_count == 2
-    assert SXAdaptor.put_file_add.call_args_list[0][0][0]['file_path'] == '/tmp/foo'
-    assert SXAdaptor.put_file_add.call_args_list[1][0][0]['file_path'] == '/tmp/bar'
+    assert SXAdaptor.put_file_add.call_args_list[0][0][0]['paths'][0] == '/tmp/foo'
+    assert SXAdaptor.put_file_add.call_args_list[1][0][0]['paths'][0] == '/tmp/bar'
 
     SXAdaptor.put_fileset_complete.assert_called_once
     assert SXAdaptor.put_fileset_complete.call_args[0][0]['files'] == 2
@@ -271,6 +271,6 @@ async def my_user_callback(did, info):
     SXAdaptor.put_fileset_complete.assert_not_called
 
     assert SXAdaptor.post_status_update.call_args[0][0] == \
-           'DID Finder found zero files for dataset 123-456'
+        'DID Finder found zero files for dataset 123-456'
 
     assert SXAdaptor.post_status_update.call_args[1]['severity'] == 'fatal'

tests/servicex_did_finder_lib/test_servicex_did.py

@@ -14,15 +14,15 @@ def test_put_file_add():
     responses.add(responses.PUT, 'http://servicex.org/files', status=206)
     sx = ServiceXAdapter("http://servicex.org")
     sx.put_file_add({
-        'file_path': 'root://foo.bar.ROOT',
+        'paths': ['root://foo.bar.ROOT'],
         'adler32': '32',
         'file_size': 1024,
         'file_events': 3141
     })
 
     assert len(responses.calls) == 1
     submitted = json.loads(responses.calls[0].request.body)
-    assert submitted['file_path'] == 'root://foo.bar.ROOT'
+    assert submitted['paths'][0] == 'root://foo.bar.ROOT'
     assert submitted['adler32'] == '32'
     assert submitted['file_events'] == 3141
     assert submitted['file_size'] == 1024
@@ -33,15 +33,15 @@ def test_put_file_add_with_prefix():
     responses.add(responses.PUT, 'http://servicex.org/files', status=206)
     sx = ServiceXAdapter("http://servicex.org", "xcache123:")
     sx.put_file_add({
-        'file_path': 'root://foo.bar.ROOT',
+        'paths': ['root://foo.bar.ROOT'],
         'adler32': '32',
         'file_size': 1024,
         'file_events': 3141
     })
 
     assert len(responses.calls) == 1
     submitted = json.loads(responses.calls[0].request.body)
-    assert submitted['file_path'] == 'xcache123:root://foo.bar.ROOT'
+    assert submitted['paths'][0] == 'xcache123:root://foo.bar.ROOT'
     assert submitted['adler32'] == '32'
     assert submitted['file_events'] == 3141
     assert submitted['file_size'] == 1024