fix: handle None state in StateMachine and None args in DowntimeCommand

- StateMachine.setState: guard against self.state being None before
  accessing self.states[self.state], preventing KeyError when the
  machine transitions from a valid state to None and back.

- DowntimeCommand._prepareCommand: change key-existence checks to
  value checks (if not self.args.get("name")) so that None values
  from decisionParams defaults are also caught.

- Add unit tests for both fixes and a new StateMachine test suite.

Author: fstagni

dirac-common/src/DIRACCommon/Core/Utilities/StateMachine.py

@@ -129,11 +129,11 @@ def setState(self, candidateState, noWarn=False, *, logger_warn=None):
         if not candidateState:
             self.state = candidateState
         elif candidateState in self.states:
-            if not self.states[self.state].stateMap:
+            if self.state is not None and not self.states[self.state].stateMap:
                 if not noWarn and logger_warn:
                     logger_warn("Final state, won't move", f"({self.state}, asked to move to {candidateState})")
                 return S_OK(self.state)
-            if candidateState not in self.states[self.state].stateMap and logger_warn:
+            if self.state is not None and candidateState not in self.states[self.state].stateMap and logger_warn:
                 logger_warn(f"Can't move from {self.state} to {candidateState}, choosing a good one")
             result = self.getNextState(candidateState)
             if not result["OK"]:

dirac-common/tests/Core/Utilities/test_StateMachine.py

@@ -0,0 +1,91 @@
+import pytest
+
+from DIRACCommon.Core.Utilities.StateMachine import State, StateMachine
+
+
+class TestState:
+    """Tests for the State class"""
+
+    def test_state_basic(self):
+        state = State(100)
+        assert state.level == 100
+        assert state.stateMap == []
+        assert state.default is None
+
+    def test_state_with_transitions(self):
+        state = State(0, ["State1", "State2"], defState="State1")
+        assert state.level == 0
+        assert state.stateMap == ["State1", "State2"]
+        assert state.default == "State1"
+
+    def test_transition_rule_in_map(self):
+        state = State(0, ["State1", "State2"], defState="State1")
+        assert state.transitionRule("State2") == "State2"
+
+    def test_transition_rule_not_in_map_with_default(self):
+        state = State(0, ["State1", "State2"], defState="State1")
+        assert state.transitionRule("UnknownState") == "State1"
+
+    def test_transition_rule_not_in_map_without_default(self):
+        state = State(0, ["State1", "State2"])
+        assert state.transitionRule("UnknownState") == "UnknownState"
+
+
+class TestStateMachine:
+    """Tests for the StateMachine class"""
+
+    def test_get_level_of_state(self):
+        sm = StateMachine()
+        assert sm.getLevelOfState("Nirvana") == 100
+        assert sm.getLevelOfState("UnknownState") == -1
+
+    def test_get_states(self):
+        sm = StateMachine()
+        assert sm.getStates() == ["Nirvana"]
+
+    def test_set_state_none_candidate(self):
+        sm = StateMachine(state="Nirvana")
+        result = sm.setState(None)
+        assert result["OK"] is True
+        assert result["Value"] is None
+
+    def test_set_state_same_state(self):
+        sm = StateMachine(state="Nirvana")
+        result = sm.setState("Nirvana")
+        assert result["OK"] is True
+        assert result["Value"] == "Nirvana"
+
+    def test_set_state_invalid_candidate(self):
+        sm = StateMachine(state="Nirvana")
+        result = sm.setState("InvalidState")
+        assert result["OK"] is False
+
+    def test_set_state_from_none_to_valid(self):
+        """Test transitioning from None state to a valid state"""
+        sm = StateMachine(state=None)
+        result = sm.setState("Nirvana")
+        assert result["OK"] is True
+        assert result["Value"] == "Nirvana"
+
+    def test_set_state_to_none_then_to_valid(self):
+        """Test setting state to None and then to a valid state - reproduces KeyError bug"""
+        sm = StateMachine(state="Nirvana")
+        # First transition to None
+        result = sm.setState(None)
+        assert result["OK"] is True
+        assert result["Value"] is None
+        # Then transition to a valid state - this should not raise KeyError
+        result = sm.setState("Nirvana")
+        assert result["OK"] is True
+        assert result["Value"] == "Nirvana"
+
+    def test_get_next_state_with_none_current(self):
+        sm = StateMachine(state=None)
+        result = sm.getNextState("Nirvana")
+        assert result["OK"] is True
+        assert result["Value"] == "Nirvana"
+
+    def test_get_next_state_invalid(self):
+        sm = StateMachine(state="Nirvana")
+        result = sm.getNextState("InvalidState")
+        assert result["OK"] is False

src/DIRAC/ResourceStatusSystem/Command/DowntimeCommand.py

@@ -157,15 +157,15 @@ def _prepareCommand(self):
         :returns: S_OK tuple ``(element, elementName, hours, gOCDBServiceType)`` or S_ERROR.
         """
 
-        if "name" not in self.args:
+        if not self.args.get("name"):
             return S_ERROR('"name" not found in self.args')
         elementName = self.args["name"]
 
-        if "element" not in self.args:
+        if not self.args.get("element"):
             return S_ERROR('"element" not found in self.args')
         element = self.args["element"]
 
-        if "elementType" not in self.args:
+        if not self.args.get("elementType"):
             return S_ERROR('"elementType" not found in self.args')
         elementType = self.args["elementType"]
 

src/DIRAC/ResourceStatusSystem/Command/test/Test_RSS_Command_GOCDBStatusCommand.py

@@ -325,6 +325,9 @@ def test_doCache(mocker):
     [
         ({"element": "X"}, None, False, None),
         ({"element": "Site", "name": "aSite", "elementType": "Z"}, S_OK(), True, None),
+        ({"element": "Site", "name": None, "elementType": "Z"}, None, False, None),
+        ({"element": "Site", "name": "aSite", "elementType": None}, None, False, None),
+        ({"element": None, "name": "aSite", "elementType": "Z"}, None, False, None),
         (
             {"element": "Resource", "name": "669 devel.edu.mk", "elementType": "Z"},
             {

src/DIRAC/ResourceStatusSystem/PolicySystem/test/Test_PolicySystem.py

@@ -3,6 +3,7 @@
 from unittest.mock import MagicMock
 
 from DIRAC import gLogger
+from DIRAC.ResourceStatusSystem.PolicySystem.PDP import PDP
 from DIRAC.ResourceStatusSystem.PolicySystem.PEP import PEP
 
 # from DIRAC.ResourceStatusSystem.PolicySystem.PDP          import PDP
@@ -274,20 +275,93 @@ def test_enforce(self):
 # # #############################################################################
 #
 # # class PolicyInvokerFailure(PolicySystemTestCase):
-#
+
 # #   def test_policyFail(self):
 # #     for granularity in ValidRes:
 # #       self.assertRaises(Exception, self.pi.evaluatePolicy)
 
 #############################################################################
 
+
+class PDPCombinePoliciesTest(PolicySystemTestCase):
+    """Test cases for PDP._combineSinglePolicyResults, especially edge cases with None status"""
+
+    def test_combineSinglePolicyResults_with_none_status(self):
+        """Test that _combineSinglePolicyResults handles None status without raising KeyError.
+
+        This reproduces the bug where decisionParams['status'] is None, causing
+        StateMachine.setState to raise KeyError when accessing self.states[None].
+        """
+        pdp = PDP()
+        pdp.setup(
+            {
+                "element": "Resource",
+                "name": "testResource",
+                "elementType": "ComputingElement",
+                "statusType": "ReadAccess",
+                "status": None,  # This is the problematic case
+                "reason": None,
+                "tokenOwner": None,
+            }
+        )
+
+        singlePolicyResults = [{"Status": "Active", "Reason": "TestReason", "Policy": {"name": "TestPolicy"}}]
+
+        # This should not raise KeyError
+        result = pdp._combineSinglePolicyResults(singlePolicyResults)
+        self.assertTrue(result["OK"], f"Expected OK=True, got: {result}")
+
+    def test_combineSinglePolicyResults_with_valid_status(self):
+        """Test that _combineSinglePolicyResults works correctly with a valid status"""
+        pdp = PDP()
+        pdp.setup(
+            {
+                "element": "Resource",
+                "name": "testResource",
+                "elementType": "ComputingElement",
+                "statusType": "ReadAccess",
+                "status": "Active",
+                "reason": None,
+                "tokenOwner": None,
+            }
+        )
+
+        singlePolicyResults = [{"Status": "Active", "Reason": "TestReason", "Policy": {"name": "TestPolicy"}}]
+
+        result = pdp._combineSinglePolicyResults(singlePolicyResults)
+        self.assertTrue(result["OK"], f"Expected OK=True, got: {result}")
+        self.assertEqual(result["Value"]["Status"], "Active")
+
+    def test_combineSinglePolicyResults_empty_results(self):
+        """Test that _combineSinglePolicyResults handles empty results"""
+        pdp = PDP()
+        pdp.setup(
+            {
+                "element": "Resource",
+                "name": "testResource",
+                "elementType": "ComputingElement",
+                "statusType": "ReadAccess",
+                "status": None,
+                "reason": None,
+                "tokenOwner": None,
+            }
+        )
+
+        result = pdp._combineSinglePolicyResults([])
+        self.assertTrue(result["OK"], f"Expected OK=True, got: {result}")
+        self.assertEqual(result["Value"]["Status"], "Unknown")
+
+
+#############################################################################
+
 if __name__ == "__main__":
     suite = unittest.defaultTestLoader.loadTestsFromTestCase(PolicySystemTestCase)
     #   suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PolicyBaseSuccess))
     #   suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PolicyBaseFailure))
     #  suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PolicyInvokerSuccess))
     #  suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PolicyInvokerFailure))
     suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PEPSuccess))
+    suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PDPCombinePoliciesTest))
     #   suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PEPFailure))
     #   suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PDPSuccess))
     #   suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(PDPFailure))