PFN checking should accept mixed protocols
Motivation
----------

For the add_files_to_datasets() call a list of attachments with pfns can be given to Rucio. In the case that these PFNs have different protocols, the PFN check fails, since the protocol is only initialised once (with the last protocol of the list) and assumes all PFNs have the same protocol.

E.g.: 

`[{'scope': 'panda', 'name': 'panda.0908235529.354928.lib._19050738', 'dids': [{'adler32': '7dd69a5f', 'name': 'panda.0908235529.354928.lib._19050738.18598150846.lib.tgz', 'bytes': 72565480L, 'pfn': 'srm://f-dpm001.grid.sinica.edu.tw:8446/srm/managerv2?SFN=/dpm/grid.sinica.edu.tw/home/atlas/atlasscratchdisk/rucio/panda/23/40/panda.0908235529.354928.lib._19050738.18598150846.lib.tgz', 'meta': {'panda_id': 4478387679, 'guid': '81789F3E-572E-4DFA-BE8B-04667522031E'}, 'scope': 'panda'}, {'adler32': '048f10bf', 'name': 'panda.0908235529.354928.lib._19050738.18598150847.log.tgz', 'bytes': 72614448L, 'pfn': 'davs://f-dpm000.grid.sinica.edu.tw:443/dpm/grid.sinica.edu.tw/home/atlas/atlasscratchdisk/rucio/panda/80/c7/panda.0908235529.354928.lib._19050738.18598150847.log.tgz', 'meta': {'panda_id': 4478387679, 'guid': 'c71d6bf8-d372-437a-b0a8-efeb3dacbfa3'}, 'scope': 'panda'}], 'rse': 'TAIWAN-LCG2_SCRATCHDISK'}]
`

ATLAS JIRA Link: https://its.cern.ch/jira/browse/RUCIO-2890

Modification
------------

PFN check needs to initialise a protocol for each protocol-group of PFNs. 
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 6b4792393..d7a547dde 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -16,7 +16,7 @@
 # - Vincent Garonne <vgaronne@gmail.com>, 2013-2018
 # - Cedric Serfon <cedric.serfon@cern.ch>, 2013-2019
 # - Ralph Vigne <ralph.vigne@cern.ch>, 2013-2014
-# - Martin Barisits <martin.barisits@cern.ch>, 2013-2018
+# - Martin Barisits <martin.barisits@cern.ch>, 2013-2019
 # - Mario Lassnig <mario.lassnig@cern.ch>, 2014-2019
 # - David Cameron <d.g.cameron@gmail.com>, 2014
 # - Thomas Beermann <thomas.beermann@cern.ch>, 2014-2018
@@ -1304,31 +1304,33 @@ def add_replicas(rse_id, files, account, ignore_availability=True,
                                     dataset_meta=dataset_meta,
                                     session=session)
 
-    pfns, scheme = [], None
+    pfns, scheme = {}, None  # {scheme: [pfns], scheme: [pfns]}
     for file in files:
         if 'pfn' not in file:
             if not replica_rse.deterministic:
                 raise exception.UnsupportedOperation('PFN needed for this (non deterministic) RSE %s ' % (replica_rse.rse))
         else:
             scheme = file['pfn'].split(':')[0]
-            pfns.append(file['pfn'])
+            pfns.setdefault(scheme, []).append(file['pfn'])
 
     if pfns:
-        p = rsemgr.create_protocol(rse_settings=rsemgr.get_rse_info(rse=replica_rse['rse'], session=session), operation='write', scheme=scheme)
-        if not replica_rse.deterministic:
-            pfns = p.parse_pfns(pfns=pfns)
-            for file in files:
-                tmp = pfns[file['pfn']]
-                file['path'] = ''.join([tmp['path'], tmp['name']])
-        else:
-            # Check that the pfns match to the expected pfns
-            lfns = [{'scope': i['scope'], 'name': i['name']} for i in files]
-            expected_pfns = p.lfns2pfns(lfns)
-            expected_pfns = clean_surls(expected_pfns.values())
-            pfns = clean_surls(pfns)
-            if pfns != expected_pfns:
-                print('ALERT: One of the PFNs provided does not match the Rucio expected PFN : got %s, expected %s (%s)' % (str(pfns), str(expected_pfns), str(lfns)))
-                raise exception.InvalidPath('One of the PFNs provided does not match the Rucio expected PFN : got %s, expected %s (%s)' % (str(pfns), str(expected_pfns), str(lfns)))
+        for scheme in pfns.keys():
+            p = rsemgr.create_protocol(rse_settings=rsemgr.get_rse_info(rse=replica_rse['rse'], session=session), operation='write', scheme=scheme)
+            if not replica_rse.deterministic:
+                pfns[scheme] = p.parse_pfns(pfns=pfns[scheme])
+                for file in files:
+                    if file['pfn'].startswith(scheme):
+                        tmp = pfns[scheme][file['pfn']]
+                        file['path'] = ''.join([tmp['path'], tmp['name']])
+            else:
+                # Check that the pfns match to the expected pfns
+                lfns = [{'scope': i['scope'], 'name': i['name']} for i in files if i['pfn'].startswith(scheme)]
+                expected_pfns = p.lfns2pfns(lfns)
+                expected_pfns = clean_surls(expected_pfns.values())
+                pfns[scheme] = clean_surls(pfns[scheme])
+                if pfns[scheme] != expected_pfns:
+                    print('ALERT: One of the PFNs provided does not match the Rucio expected PFN : got %s, expected %s (%s)' % (str(pfns), str(expected_pfns), str(lfns)))
+                    raise exception.InvalidPath('One of the PFNs provided does not match the Rucio expected PFN : got %s, expected %s (%s)' % (str(pfns), str(expected_pfns), str(lfns)))
 
     nbfiles, bytes = __bulk_add_replicas(rse_id=rse_id, files=files, account=account, session=session)
     increase(rse_id=rse_id, files=nbfiles, bytes=bytes, session=session)
diff --git a/lib/rucio/tests/test_replica.py b/lib/rucio/tests/test_replica.py
index 963db7a4e..ba5cd6df8 100644
--- a/lib/rucio/tests/test_replica.py
+++ b/lib/rucio/tests/test_replica.py
@@ -27,6 +27,7 @@ from __future__ import print_function
 from datetime import datetime, timedelta
 from json import dumps, loads
 
+import hashlib
 import xmltodict
 
 from nose.tools import assert_equal, assert_in, assert_raises
@@ -435,6 +436,48 @@ class TestReplicaCore:
                                                 client_location={'site': 'SOMEWHERE'})]
         assert_in('root://', replicas[0]['pfns'].keys()[0])
 
+    def test_replica_mixed_protocols(self):
+        """ REPLICA (CORE): Test adding replicas with mixed protocol """
+
+        rc = ReplicaClient()
+
+        rse = 'APERTURE_%s' % rse_name_generator()
+        rse_id = add_rse(rse)
+
+        add_protocol(rse_id, {'scheme': 'root',
+                              'hostname': 'root.aperture.com',
+                              'port': 1409,
+                              'prefix': '//test/chamber/',
+                              'impl': 'rucio.rse.protocols.xrootd.Default',
+                              'domains': {
+                                  'lan': {'read': 1, 'write': 1, 'delete': 1},
+                                  'wan': {'read': 1, 'write': 1, 'delete': 1}}})
+        add_protocol(rse_id, {'scheme': 'http',
+                              'hostname': 'root.aperture.com',
+                              'port': 1409,
+                              'prefix': '//test/chamber/',
+                              'impl': 'rucio.rse.protocols.xrootd.Default',
+                              'domains': {
+                                  'lan': {'read': 1, 'write': 1, 'delete': 1},
+                                  'wan': {'read': 1, 'write': 1, 'delete': 1}}})
+
+        tmp_scope = InternalScope('mock')
+        root = InternalAccount('root')
+
+        files = []
+
+        name = 'element_%s' % generate_uuid()
+        hstr = hashlib.md5(('%s:%s' % (tmp_scope, name)).encode('utf-8')).hexdigest()
+        pfn = 'root://root.aperture.com:1409//test/chamber/mock/%s/%s/%s' % (hstr[0:2], hstr[2:4], name)
+        files.append({'scope': tmp_scope, 'name': name,'bytes': 1234, 'adler32': 'deadbeef', 'pfn': pfn})
+
+        name = 'element_%s' % generate_uuid()
+        hstr = hashlib.md5(('%s:%s' % (tmp_scope, name)).encode('utf-8')).hexdigest()
+        pfn = 'http://root.aperture.com:1409//test/chamber/mock/%s/%s/%s' % (hstr[0:2], hstr[2:4], name)
+        files.append({'scope': tmp_scope, 'name': name,'bytes': 1234, 'adler32': 'deadbeef', 'pfn': pfn})
+
+        add_replicas(rse_id=rse_id, files=files, account=root)
+
     def test_set_tombstone(self):
         """ REPLICA (CORE): set tombstone on replica """
         # Set tombstone on one replica
['lib/rucio/core/replica.py', 'lib/rucio/tests/test_replica.py']