Repository: rucio/rucio
Issue: rucio upload --no-register --pfn fails

Motivation
----------
rucio upload --no-register --pfns fails while it works with 1.16.4.

```
[wguan@lxplus067 wguan]$ /usr/bin/env rucio -v upload --rse CERN-PROD_ES --scope transient --no-register --pfn s3://cs3.cern.ch:443//atlas-eventservice/05cc36b8-383f-444f-a384-592fad3b918e /bin/hostname
2018-07-25 14:54:14,552 WARNING Upload with given pfn implies that no_register is True
2018-07-25 14:54:14,658 INFO    Preparing upload for file hostname
2018-07-25 14:54:16,608 ERROR   An unknown exception occurred.
Details: no error information passed (http status code: 500 ('internal_server_error', 'server_error', '/o\\\\', '\\xe2\\x9c\\x97'))
Completed in 2.1462 sec.
```

The problem is related to rsemanager.exists call, which trys to resolve the did on the server with lfns2pfns, which does not work as the replica does not exists. In 1.16.4 with bin/rucio it was first checked if the did exists, and if not, rsemanager.exists was not even called.

Changed files:
lib/rucio/client/uploadclient.py

diff --git a/lib/rucio/client/uploadclient.py b/lib/rucio/client/uploadclient.py
index a9561af36..6ace6feef 100644
--- a/lib/rucio/client/uploadclient.py
+++ b/lib/rucio/client/uploadclient.py
@@ -139,15 +139,31 @@ class UploadClient:
             file_did = {'scope': file['did_scope'], 'name': file['did_name']}
             dataset_did_str = file.get('dataset_did_str')
 
+            rse = file['rse']
+            rse_settings = self.rses[rse]
+            is_deterministic = rse_settings.get('deterministic', True)
+            if not is_deterministic and not pfn:
+                logger.error('PFN has to be defined for NON-DETERMINISTIC RSE.')
+                continue
+            if pfn and not is_deterministic:
+                logger.warning('Upload with given pfn implies that no_register is True, except non-deterministic RSEs')
+                no_register = True
+
             if not no_register:
                 self._register_file(file, registered_dataset_dids)
 
-            rse = file['rse']
-            rse_settings = self.rses[rse]
             # if file already exists on RSE we're done
-            if rsemgr.exists(rse_settings, file_did):
-                logger.info('File already exists on RSE. Skipping upload')
-                continue
+            if not is_deterministic and not no_register:
+                if rsemgr.exists(rse_settings, pfn):
+                    logger.info('File already exists on RSE with given pfn. Skipping upload. Existing replica has to be removed first.')
+                    continue
+                elif rsemgr.exists(rse_settings, file_did):
+                    logger.info('File already exists on RSE with different pfn. Skipping upload.')
+                    continue
+            else:
+                if rsemgr.exists(rse_settings, pfn if pfn else file_did):
+                    logger.info('File already exists on RSE. Skipping upload')
+                    continue
 
             protocols = rsemgr.get_protocols_ordered(rse_settings=rse_settings, operation='write', scheme=force_scheme)
             protocols.reverse()
@@ -173,7 +189,7 @@ class UploadClient:
                                           force_scheme=cur_scheme,
                                           force_pfn=pfn,
                                           transfer_timeout=file.get('transfer_timeout'))
-                    success = True
+                    success = state['success']
                     file['upload_result'] = state
                 except (ServiceUnavailable, ResourceTemporaryUnavailable) as error:
                     logger.warning('Upload attempt failed')
@@ -366,9 +382,6 @@ class UploadClient:
                 logger.warning('Skipping file %s because no rse was given' % path)
                 continue
             if pfn:
-                if item.get('no_register'):
-                    logger.warning('Upload with given pfn implies that no_register is True')
-                    item['no_register'] = True
                 item['force_scheme'] = pfn.split(':')[0]
 
             if os.path.isdir(path):
@@ -409,4 +422,7 @@ class UploadClient:
         replica['md5'] = file['md5']
         replica['meta'] = file['meta']
         replica['state'] = file['state']
+        pfn = file.get('pfn')
+        if pfn:
+            replica['pfn'] = pfn
         return replica