Tests/Rule move_rule with invalid Source Replica Expression; Fix ru…

…cio#4845

Currently, a misleading error message is raised to the user upon
attempting to move a rule which has an invalid Source Replica Expression
(SRE). An invalid SRE is an RSE expression which evaluates to an empty
set, which can occur due to a variety of reasons.

The misleading error message informs the user that the provided `RSE
expression is considered invalid`, this is misleading, since the error
does not stem from the RSE expression provided to the `move_rule`
command but from the SRE attribute of the rule prior to moving/updating
it.

The solution is to introduce a new exception
`InvalidSourceReplicaExpression` and raise it within
`core/rule.py::add_rule`, should the attempt to parse the SRE with
`parse_expression` fail.

This commit also includes a test added to `test_rule.py::TestCore` which
simulates two ways in which the SRE might become invalid:
1) By deleting the only RSE the SRE points to.
2) By matching the RSE to the SRE via an RSE attribute but then deleting
   this RSE attribute from the RSE.
In both cases the SRE will evaluate to an empty set.

lib/rucio/common/exception.py

@@ -1180,3 +1180,14 @@ def __init__(self, enddatetime: Optional[str], *args, **kwargs):
         super(ServerSSLCertificateExpiredException, self).__init__(*args, **kwargs)
         self._message = f'Server SSL Certificate has expired on {enddatetime}, auth is no longer possible'
         self.error_code = 104
+
+
+class InvalidSourceReplicaExpression(RucioException):
+    """
+    Source Replica Expression Considered Invalid
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(InvalidSourceReplicaExpression, self).__init__(*args, **kwargs)
+        self._message = 'Provided Source Replica expression is considered invalid.'
+        self.error_code = 105

lib/rucio/core/rule.py

@@ -47,7 +47,8 @@
                                     ReplicationRuleCreationTemporaryFailed, InsufficientTargetRSEs, RucioException,
                                     InvalidRuleWeight, StagingAreaRuleRequiresLifetime, DuplicateRule,
                                     InvalidObject, RSEWriteBlocked, RuleReplaceFailed, RequestNotFound,
-                                    ManualRuleApprovalBlocked, UnsupportedOperation, UndefinedPolicy, InvalidValueForKey)
+                                    ManualRuleApprovalBlocked, UnsupportedOperation, UndefinedPolicy, InvalidValueForKey,
+                                    InvalidSourceReplicaExpression)
 from rucio.common.schema import validate_schema
 from rucio.common.types import InternalScope, InternalAccount
 from rucio.common.utils import str_to_date, sizefmt, chunks
@@ -151,7 +152,10 @@ def add_rule(dids, account, copies, rse_expression, grouping, weight, lifetime,
                         raise ManualRuleApprovalBlocked()
 
             if source_replica_expression:
-                source_rses = parse_expression(source_replica_expression, filter_={'vo': vo}, session=session)
+                try:
+                    source_rses = parse_expression(source_replica_expression, filter_={'vo': vo}, session=session)
+                except InvalidRSEExpression:
+                    raise InvalidSourceReplicaExpression
             else:
                 source_rses = []
 

lib/rucio/tests/test_rule.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from collections import namedtuple
 import json
 import random
 import string
@@ -26,7 +27,7 @@
 from rucio.common.config import config_get_bool
 from rucio.common.exception import (RucioException, RuleNotFound, AccessDenied, InsufficientAccountLimit, DuplicateRule, RSEWriteBlocked,
                                     RSEOverQuota, RuleReplaceFailed, ManualRuleApprovalBlocked, InputValidationError,
-                                    UnsupportedOperation, InvalidValueForKey)
+                                    UnsupportedOperation, InvalidValueForKey, InvalidSourceReplicaExpression)
 from rucio.common.policy import get_policy
 from rucio.common.schema import get_schema_value
 from rucio.common.types import InternalAccount, InternalScope
@@ -53,8 +54,10 @@
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from typing import Tuple
+    from rucio.tests.temp_factories import TemporaryDidFactory, TemporaryRSEFactory
 
 LOG = getLogger(__name__)
+RSE_namedtuple = namedtuple('RSE_namedtuple', ['name', 'id'])
 
 
 def create_files(nrfiles, scope, rse_id, bytes_=1):
@@ -153,6 +156,72 @@ def setup_class(request, vo, root_account, jdoe_account, rse_factory):
 @pytest.mark.usefixtures('setup_class')
 class TestCore:
 
+    @pytest.mark.parametrize(
+        "delete_rse", [True, False]
+    )
+    def test_move_rule_invalid_source_replica_expression(
+        self,
+        delete_rse: bool,
+        did_factory: "TemporaryDidFactory",
+        rse_factory: "TemporaryRSEFactory",
+        root_account: "InternalAccount"
+    ):
+        """
+        REPLICATION RULE (CORE): Move a rule using an invalid source replica expression.
+
+        Parametrised so that there are two different ways for the source replica
+        expression (SRE) to become invalid:
+        1) by deleting the source RSE
+        2) by referring to the source RSE via an RSE attribute and then deleting
+            said RSE attribute from the source RSE
+        in both cases the SRE is invalidated since it evaluates to an empty set.
+        """
+
+        from rucio.common.cache import make_region_memcached
+        from dogpile.cache.api import NO_VALUE
+        from hashlib import sha256
+        REGION = make_region_memcached(expiration_time=600)
+
+        # create RSEs A, B, C (structured into namedtuples)
+        rseA, rseB, rseC = (
+            RSE_namedtuple(*rse_factory.make_mock_rse()) for _ in range(3)
+        )
+        rsekey, rseval = tag_generator(), tag_generator()
+        add_rse_attribute(rseA.id, rsekey, rseval)
+
+        # create DID (dataset is easiest)
+        dataset = did_factory.make_dataset()
+
+        # create source replica expression
+        if delete_rse:
+            SRE = rseA.name
+        else:
+            SRE = f"{rsekey}={rseval}"
+
+        # add rule binding DID to RSE-B, source replica expression matches RSE-A
+        rule = {
+            "account": root_account,
+            "copies": 1,
+            "rse_expression": rseB.name,
+            "source_replica_expression": SRE
+        }
+        ruledict = add_rules([dataset], [rule])
+        rule_id = ruledict[(dataset["scope"], dataset["name"])][0]
+
+        # delete RSE-A (or otherwise unmatching from source replica expr)
+        if delete_rse:
+            from rucio.core.rse import del_rse
+            del_rse(rseA.id)
+        else:
+            del_rse_attribute(rseA.id, rsekey)
+
+        # must remove cached entries for the Source Replica Expression
+        REGION.set(sha256(SRE.encode()).hexdigest(), NO_VALUE)
+
+        # move-rule asserts correct error
+        with pytest.raises(InvalidSourceReplicaExpression):
+            move_rule(rule_id, rseC.name)
+
     def test_add_rule_to_file_ask_approval(self, vo, mock_scope, jdoe_account):
         """ REPLICATION RULE (CORE): Add a replication rule, asking approval"""
         rse = rse_name_generator()