Core list_replicas() does not make use of ignore_availability
Motivation
----------

If an RSE is excluded for reading, the the download client should not attempt to download from that RSE.

The REST layer already does this, but once `ignore_availability` propagates to the core, it is not actually used.
https://github.com/rucio/rucio/blob/960855d6e86f4e38ebe584b1d26e65751d52a210/lib/rucio/web/rest/webpy/v1/replica.py#L336
https://github.com/rucio/rucio/blob/960855d6e86f4e38ebe584b1d26e65751d52a210/lib/rucio/core/replica.py#L1164

Modification
------------
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 3a7af583c..7c3ec1c4e 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -704,7 +704,7 @@ def _resolve_dids(dids, unavailable, ignore_availability, all_states, resolve_ar
     return file_clause, dataset_clause, state_clause, files, constituents
 
 
-def _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, updated_after, session):
+def _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, ignore_availability, updated_after, session):
     """
     List file replicas for a list of datasets.
 
@@ -733,6 +733,9 @@ def _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, update
         order_by(models.DataIdentifierAssociation.child_scope,
                  models.DataIdentifierAssociation.child_name)
 
+    if not ignore_availability:
+        replica_query = replica_query.filter(models.RSE.availability.in_((4, 5, 6, 7)))
+
     if state_clause is not None:
         replica_query = replica_query.filter(and_(state_clause))
 
@@ -746,7 +749,7 @@ def _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, update
         yield replica
 
 
-def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, updated_after, session):
+def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, ignore_availability, updated_after, session):
     """
     List file replicas for a list of files.
 
@@ -759,6 +762,9 @@ def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, updat
                    or_(*replica_condition),
                    ]
 
+        if not ignore_availability:
+            filters.append(models.RSE.availability.in_((4, 5, 6, 7)))
+
         if state_clause is not None:
             filters.append(state_clause)
 
@@ -860,11 +866,11 @@ def get_multi_cache_prefix(cache_site, filename, logger=logging.log):
 def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                    schemes, files, rse_clause, rse_expression, client_location, domain,
                    sign_urls, signature_lifetime, constituents, resolve_parents,
-                   updated_after, filters,
+                   updated_after, filters, ignore_availability,
                    session):
 
-    files = [dataset_clause and _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, updated_after, session),
-             file_clause and _list_replicas_for_files(file_clause, state_clause, files, rse_clause, updated_after, session)]
+    files = [dataset_clause and _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, ignore_availability, updated_after, session),
+             file_clause and _list_replicas_for_files(file_clause, state_clause, files, rse_clause, ignore_availability, updated_after, session)]
 
     # we need to retain knowledge of the original domain selection by the user
     # in case we have to loop over replicas with a potential outgoing proxy
@@ -905,6 +911,7 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                 archive_result = list_replicas(dids=constituents['%s:%s' % (scope.internal, name)],
                                                schemes=schemes, client_location=client_location,
                                                domain=domain, sign_urls=sign_urls,
+                                               ignore_availability=ignore_availability,
                                                rse_expression=rse_expression,
                                                signature_lifetime=signature_lifetime,
                                                updated_after=updated_after,
@@ -1271,7 +1278,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     for f in _list_replicas(dataset_clause, file_clause, state_clause, pfns,
                             schemes, files, rse_clause, rse_expression, client_location, domain,
                             sign_urls, signature_lifetime, constituents, resolve_parents,
-                            updated_after, filter,
+                            updated_after, filter, ignore_availability,
                             session):
         yield f
 
diff --git a/lib/rucio/tests/test_archive.py b/lib/rucio/tests/test_archive.py
index 81a77e7e0..4f04c9078 100644
--- a/lib/rucio/tests/test_archive.py
+++ b/lib/rucio/tests/test_archive.py
@@ -26,6 +26,7 @@
 
 from rucio.common.utils import generate_uuid
 from rucio.core.replica import add_replicas, delete_replicas
+from rucio.core.rse import update_rse
 from rucio.core.did import attach_dids, get_metadata
 
 
@@ -132,6 +133,17 @@ def test_list_archive_contents_at_rse(rse_factory, mock_scope, root_account, did
     assert rse1 not in res
     assert rse2 in res
 
+    # if archive file is on a blacklisted RSE, it must not be listed
+    both_rses = rse1 + '|' + rse2
+    update_rse(rse1_id, {'availability_read': False})
+    update_rse(rse2_id, {'availability_read': False})
+    res = replica_client.list_replicas(dids=[{'scope': f['scope'], 'name': f['name']} for f in archived_file], metalink=True, rse_expression=both_rses, resolve_archives=True)
+    assert rse1 not in res
+    assert rse2 not in res
+    res = replica_client.list_replicas(dids=[{'scope': f['scope'], 'name': f['name']} for f in archived_file], metalink=True, rse_expression=both_rses, resolve_archives=True, unavailable=True)
+    assert rse1 in res
+    assert rse2 in res
+
 
 def test_archive_on_dataset_level(rse_factory, did_factory, root_account):
     rse_name, rse_id = rse_factory.make_xroot_rse()
diff --git a/lib/rucio/tests/test_replica.py b/lib/rucio/tests/test_replica.py
index bd745f016..0ddc71d80 100644
--- a/lib/rucio/tests/test_replica.py
+++ b/lib/rucio/tests/test_replica.py
@@ -902,6 +902,42 @@ def test_client_access_denied_on_delete_replicas(rse_factory, mock_scope, replic
         assert len(replicas) == 1
 
 
+def test_client_list_blacklisted_replicas(rse_factory, did_factory, replica_client, did_client):
+    """ REPLICA (CLIENT): Blacklisted replicas are filtered in list replicas"""
+
+    rse, _ = rse_factory.make_posix_rse()
+    file = did_factory.upload_test_file(rse)
+    dataset = did_factory.make_dataset()
+    container = did_factory.make_container()
+
+    # make all scopes external
+    file, dataset, container = ({'scope': did['scope'].external, 'name': did['name']} for did in (file, dataset, container))
+
+    did_client.add_files_to_dataset(files=[file], **dataset)
+    did_client.add_datasets_to_container(dsns=[dataset], **container)
+
+    # availability_write will not have any impact on listing replicas
+    did_factory.client.update_rse(rse, {'availability_write': False})
+    for did in (file, dataset, container):
+        replicas = list(replica_client.list_replicas(dids=[did]))
+        assert len(replicas) == 1
+        assert len(replicas[0]['rses']) == 1
+
+    # if availability_read is set to false, the replicas from the given rse will not be listed
+    did_factory.client.update_rse(rse, {'availability_read': False})
+    replicas = list(replica_client.list_replicas(dids=[file]))
+    assert len(replicas) == 1
+    assert not replicas[0]['rses'] and not replicas[0]['pfns']
+    for did in (dataset, container):
+        replicas = list(replica_client.list_replicas(dids=[did]))
+        assert len(replicas) == 0
+    # Explicitly requesting unavailable replicas will return them
+    for did in (file, dataset, container):
+        replicas = list(replica_client.list_replicas(dids=[did], unavailable=True))
+        assert len(replicas) == 1
+        assert len(replicas[0]['rses']) == 1
+
+
 @pytest.mark.dirty
 @pytest.mark.noparallel(reason='runs minos, which acts on all bad pfns')
 def test_client_add_temporary_unavailable_pfns(rse_factory, mock_scope, replica_client):
['lib/rucio/core/replica.py', 'lib/rucio/tests/test_archive.py', 'lib/rucio/tests/test_replica.py']