diff --git a/foreman/config/scenario.yaml b/foreman/config/scenario.yaml index 8348785..4de4667 100644 --- a/foreman/config/scenario.yaml +++ b/foreman/config/scenario.yaml @@ -5,17 +5,6 @@ hardware: lifecycle_nodes: - dummy_lifecycle_node -controllers: - joint_state_broadcaster: - requires: - - [kassow, inactive] - - [FrankaHardwareInterface, inactive] - - [dummy_lifecycle_node, active] - kassow_joint_trajectory_controller: - requires: [kassow, active] - franka_joint_trajectory_controller: - requires: [FrankaHardwareInterface, active] - goal_states: idle: diff --git a/foreman/foreman/adapters/component_state_monitor.py b/foreman/foreman/adapters/component_state_monitor.py index 1b8d332..946b3fe 100644 --- a/foreman/foreman/adapters/component_state_monitor.py +++ b/foreman/foreman/adapters/component_state_monitor.py @@ -1,6 +1,9 @@ +import threading from typing import Dict, List from controller_manager_msgs.msg import ControllerManagerActivity +from controller_manager_msgs.srv import ListControllers +from controller_manager_msgs.srv import ListHardwareComponents from lifecycle_msgs.msg import TransitionEvent from lifecycle_msgs.srv import GetState from rclpy.event_handler import QoSSubscriptionMatchedInfo @@ -14,6 +17,8 @@ from foreman.engine import ForemanEngine from foreman.types import Component from foreman.types import ComponentType +from foreman.types import ControllerDependencyRule +from foreman.types import HardwareRequirement from foreman.types import LifecycleState @@ -64,6 +69,24 @@ def __init__( f"{self._logger_prefix} Subscribed to /{controller_manager_name}/activity" ) + # Service clients used to query controller_manager when building dependency rules. + self._client_list_controllers = self._node.create_client( + ListControllers, + f'/{controller_manager_name}/list_controllers', + callback_group=self._node.callback_group_subscriber + ) + self._client_list_hardware_components = self._node.create_client( + ListHardwareComponents, + f'/{controller_manager_name}/list_hardware_components', + callback_group=self._node.callback_group_subscriber + ) + + # Latest controller_manager observations used to infer dependency rules. + # Updated asynchronously from controller_manager callbacks. + self._dependency_lock = threading.Lock() + self._latest_controllers = [] + self._latest_hardware_components = [] + # --- Lifecycle node monitoring --- self._lc_node_get_state_clients: Dict[str, object] = {} @@ -96,6 +119,88 @@ def __init__( f"{self._logger_prefix} Monitoring lifecycle nodes: {lifecycle_nodes}" ) + @staticmethod + def infer_dependency_rules(controller_states, hardware_components): + """Infer each controller's hardware dependencies from its required interfaces. + + A command interface requires its owning hardware ACTIVE; a state interface + requires it only INACTIVE. If a controller needs both from one hardware, ACTIVE wins. + """ + owner_of_interface = {} + # Build a mapping of interface names to their owning hardware components + for hardware in hardware_components: + for interface in list(hardware.command_interfaces) + list(hardware.state_interfaces): + owner_of_interface[interface.name] = hardware.name + + dependency_rules = [] + for controller in controller_states: + # Map required interfaces to hardware components and their required lifecycle states + required_hardware_state = {} + for interface in controller.required_command_interfaces: + hardware_name = owner_of_interface.get(interface) + if hardware_name: + required_hardware_state[hardware_name] = LifecycleState.ACTIVE + + for interface in controller.required_state_interfaces: + hardware_name = owner_of_interface.get(interface) + if hardware_name and hardware_name not in required_hardware_state: + required_hardware_state[hardware_name] = LifecycleState.INACTIVE + + dependency_rules.append(ControllerDependencyRule( + controller_name=controller.name, + required_hardware=[ + HardwareRequirement(name=name, state=state) + for name, state in required_hardware_state.items() + ], + )) + return dependency_rules + + def get_dependency_rules(self): + """ + Infer dependency rules from the latest controller_manager observations. + + The observations are refreshed asynchronously, so this method never blocks. + """ + with self._dependency_lock: + return self.infer_dependency_rules( + self._latest_controllers, self._latest_hardware_components) + + def _refresh_dependency_rules(self): + """Refresh dependency observations asynchronously. + + Query hardware first, then controllers, before rebuilding dependencies. + """ + if not (self._client_list_hardware_components.service_is_ready() + and self._client_list_controllers.service_is_ready()): + return + future = self._client_list_hardware_components.call_async( + ListHardwareComponents.Request()) + future.add_done_callback(self._on_hardware_components_response) + + def _on_hardware_components_response(self, future): + """Store the latest hardware observations and request controllers.""" + try: + hardware = future.result().component + except Exception as e: + self._node.get_logger().warning( + f"{self._logger_prefix} Failed to list hardware components: {e}") + return + with self._dependency_lock: + self._latest_hardware_components = hardware + ctrl_future = self._client_list_controllers.call_async(ListControllers.Request()) + ctrl_future.add_done_callback(self._on_controllers_response) + + def _on_controllers_response(self, future): + """Store the latest controller observations.""" + try: + controllers = future.result().controller + except Exception as e: + self._node.get_logger().warning( + f"{self._logger_prefix} Failed to list controllers: {e}") + return + with self._dependency_lock: + self._latest_controllers = controllers + # TODO: use matched event for this topic as well? That way we know if controller manager dies. # Minor. Currently we catch unexpected transitions in the engine (all components go to finalized) def _activity_callback(self, msg: ControllerManagerActivity): @@ -124,6 +229,8 @@ def _activity_callback(self, msg: ControllerManagerActivity): self._cm_components = components self._push_merged_state() + # Refresh dependency observations after receiving a new activity update. + self._refresh_dependency_rules() def _on_lifecycle_publisher_matched(self, name: str, info: QoSSubscriptionMatchedInfo): """DDS matched event: lifecycle node's transition_event publisher appeared or disappeared.""" diff --git a/foreman/foreman/engine.py b/foreman/foreman/engine.py index d888e1a..8107eb1 100644 --- a/foreman/foreman/engine.py +++ b/foreman/foreman/engine.py @@ -5,6 +5,8 @@ from foreman.planner import Planner from foreman.types import Component from foreman.types import ComponentType +from foreman.types import ControllerDependencyRule +from foreman.types import DependencyProvider from foreman.types import ErrorSnapshot from foreman.types import ForemanError from foreman.types import ForemanErrorCategory @@ -91,6 +93,10 @@ def abort_goal(self, error: ForemanError): self._last_issued_command = None self._locked_abort_transition() + def set_dependency_provider(self, dependency_provider: DependencyProvider): + """Set the source of dependency rules for the planner.""" + self._planner.set_dependency_provider(dependency_provider) + def get_next_transition(self) -> Optional[SystemTransitionCommand]: """Calculate the next step toward the goal.""" if not self._current_goal: @@ -251,8 +257,9 @@ def _locked_check_unsatisfiable_dependencies(self, goal: SystemGoal) -> List[str goal_infrastructure_states[comp.name] = comp.lifecycle_state errors = [] + current_dependency_rules = self._planner.get_current_rules() for ctrl_goal in goal.controller_goals: - rule = self._planner.rules.get(ctrl_goal.name) + rule = current_dependency_rules.get(ctrl_goal.name) if not rule: continue diff --git a/foreman/foreman/node.py b/foreman/foreman/node.py index 0194ef0..466bf7b 100644 --- a/foreman/foreman/node.py +++ b/foreman/foreman/node.py @@ -47,6 +47,8 @@ def __init__(self): controller_manager_name=controller_manager_name, lifecycle_nodes=self.foreman_config.lifecycle_nodes ) + # Planner queries dependency rules from the monitor when planning. + self.foreman_engine.set_dependency_provider(self.component_state_monitor) self.controller_manager_service_caller = adapters.ControllerManagerServiceCaller( node=self, controller_manager_name=controller_manager_name diff --git a/foreman/foreman/parser.py b/foreman/foreman/parser.py index 098a1c2..48ca5f9 100644 --- a/foreman/foreman/parser.py +++ b/foreman/foreman/parser.py @@ -8,7 +8,6 @@ from foreman.types import Component from foreman.types import ComponentType from foreman.types import ControllerDependencyRule -from foreman.types import HardwareRequirement from foreman.types import LifecycleState from foreman.types import SystemGoal @@ -44,49 +43,6 @@ def parse_state_string(state_str: str) -> LifecycleState: return state_mapping[normalized] -def parse_requires( - requires: List[str], hardware: List[str], lifecycle_nodes: List[str] = None -) -> List[HardwareRequirement]: - """Parse the 'requires' field into list of HardwareRequirement. - - Supports: - - [all, inactive] -> all hardware + lifecycle nodes must be at that state - - [component_name, active] -> specific hardware or lifecycle node must be at that state - """ - if lifecycle_nodes is None: - lifecycle_nodes = [] - - if not requires: - return [] - - # we can get either a single [component, state] entry - # or a list of [component, state] entries. - # here we normalize so we work with a list of [component, state] - if len(requires) == 2 and isinstance(requires[0], str): - requires_normalized = [requires] - else: - requires_normalized = requires - - reqs = [] - - for req in requires_normalized: - if not isinstance(req, list) or len(req) != 2: - raise ValueError(f"Invalid requirement format: {req}. Expected [target, state].") - - target = req[0] - state = parse_state_string(req[1]) - - if target == 'all': - reqs.extend([ - HardwareRequirement(name=name, state=state) - for name in hardware + lifecycle_nodes - ]) - else: - reqs.append(HardwareRequirement(name=target, state=state)) - - return reqs - - def parse_yaml_file(file_path: Path) -> ParsedScenario: """Parse a scenario YAML file into a ParsedScenario object.""" with open(file_path, 'r') as f: @@ -98,16 +54,8 @@ def parse_yaml_file(file_path: Path) -> ParsedScenario: hardware = data.get('hardware', []) lifecycle_nodes = data.get('lifecycle_nodes', []) - dependency_rules = [] - controllers = data.get('controllers', {}) - for ctrl_name, ctrl_config in controllers.items(): - requires = ctrl_config.get('requires', []) - reqs = parse_requires(requires, hardware, lifecycle_nodes) - - dependency_rules.append(ControllerDependencyRule( - controller_name=ctrl_name, - required_hardware=reqs - )) + # Dependencies are inferred at runtime from the controller_manager + dependency_rules: List[ControllerDependencyRule] = [] goals = {} goal_states = data.get('goal_states', {}) @@ -152,8 +100,6 @@ def parse_yaml_file(file_path: Path) -> ParsedScenario: metadata[key] = value tracked_components = set(hardware + lifecycle_nodes) - for rule in dependency_rules: - tracked_components.add(rule.controller_name) for goal in goals.values(): tracked_components.update(c.name for c in goal.hardware_goals) tracked_components.update(c.name for c in goal.controller_goals) diff --git a/foreman/foreman/planner.py b/foreman/foreman/planner.py index ffa6eb9..c5182ed 100644 --- a/foreman/foreman/planner.py +++ b/foreman/foreman/planner.py @@ -3,6 +3,7 @@ from foreman.types import Component from foreman.types import ComponentType from foreman.types import ControllerDependencyRule +from foreman.types import DependencyProvider from foreman.types import LifecycleState from foreman.types import SystemGoal from foreman.types import SystemState @@ -12,8 +13,21 @@ class Planner: """Plan the next single step towards the lifecycle state goal of the system.""" - def __init__(self, dependency_rules: List[ControllerDependencyRule]): - self.rules = {rule.controller_name: rule for rule in dependency_rules} + def __init__(self, dependency_rules: List[ControllerDependencyRule] = None, + dependency_provider: Optional[DependencyProvider] = None): + self._dependency_provider = dependency_provider + self.rules = {rule.controller_name: rule for rule in (dependency_rules or [])} + + def set_dependency_provider(self, dependency_provider: DependencyProvider): + """Set a provider for dependency rules, which can be queried at runtime.""" + self._dependency_provider = dependency_provider + + def get_current_rules(self): + """Return the current dependency rules keyed by controller name.""" + if self._dependency_provider is not None: + self.rules = {rule.controller_name: rule + for rule in self._dependency_provider.get_dependency_rules()} + return self.rules def get_next_transition( self, current_state: SystemState, goal: SystemGoal @@ -23,6 +37,9 @@ def get_next_transition( Priority: C deactivate > HW Down > HW Up > C cleanup > C config > C activate. """ + # Refresh dependency rules from the provider before computing the next transition. + self.get_current_rules() + cmds_hw_step_up = [] cmds_hw_step_down = [] cmds_ctrl_config = [] diff --git a/foreman/foreman/types.py b/foreman/foreman/types.py index 6afb49f..118911e 100644 --- a/foreman/foreman/types.py +++ b/foreman/foreman/types.py @@ -2,7 +2,7 @@ from dataclasses import field from enum import Enum from pathlib import Path -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Protocol from lifecycle_msgs.msg import State @@ -136,6 +136,12 @@ class ControllerDependencyRule: required_hardware: List[HardwareRequirement] +class DependencyProvider(Protocol): + """A source of controller dependency rules the planner can pull from.""" + + def get_dependency_rules(self) -> List['ControllerDependencyRule']: ... + + @dataclass class SystemGoal: """Named system goal state. Populated from YAML.""" diff --git a/foreman/test/test_component_state_monitor.py b/foreman/test/test_component_state_monitor.py new file mode 100644 index 0000000..3f37544 --- /dev/null +++ b/foreman/test/test_component_state_monitor.py @@ -0,0 +1,123 @@ +import unittest +from unittest import mock +from unittest.mock import Mock + +from controller_manager_msgs.msg import ControllerState +from controller_manager_msgs.msg import HardwareComponentState +from controller_manager_msgs.msg import HardwareInterface +from controller_manager_msgs.srv import ListControllers +from controller_manager_msgs.srv import ListHardwareComponents +import rclpy +from rclpy.callback_groups import MutuallyExclusiveCallbackGroup +from rclpy.callback_groups import ReentrantCallbackGroup + +from foreman.adapters.component_state_monitor import ComponentStateMonitor +from foreman.types import LifecycleState + + +def _controller_state(name, required_command=None, required_state=None): + """Build a ControllerState message like list_controllers returns.""" + return ControllerState( + name=name, + required_command_interfaces=required_command or [], + required_state_interfaces=required_state or [], + ) + + +def _hardware_component(name, command=None, state=None): + """Build a HardwareComponentState message like list_hardware_components returns.""" + return HardwareComponentState( + name=name, + command_interfaces=[HardwareInterface(name=n) for n in (command or [])], + state_interfaces=[HardwareInterface(name=n) for n in (state or [])], + ) + + +class TestInferDependencyRules(unittest.TestCase): + """Test the inference of dependency rules based on controller and hardware states.""" + + def test_command_interface_requires_hardware_active(self): + # A controller that writes a command interface needs its hardware ACTIVE. + controllers = [_controller_state( + "forward_position_controller", required_command=["joint1/position"])] + hardware = [_hardware_component("RRBot", command=["joint1/position"])] + rule = ComponentStateMonitor.infer_dependency_rules(controllers, hardware)[0] + self.assertEqual(rule.required_hardware[0].name, "RRBot") + self.assertEqual(rule.required_hardware[0].state, LifecycleState.ACTIVE) + + def test_state_interface_requires_hardware_inactive(self): + # A broadcaster only reads a state interface, so the hardware need only be INACTIVE. + controllers = [_controller_state("joint_state_broadcaster", + required_state=["joint1/position"])] + hardware = [_hardware_component("RRBot", state=["joint1/position"])] + rule = ComponentStateMonitor.infer_dependency_rules(controllers, hardware)[0] + self.assertEqual(rule.required_hardware[0].name, "RRBot") + self.assertEqual(rule.required_hardware[0].state, LifecycleState.INACTIVE) + + def test_needing_command_and_state_requires_active(self): + # If a controller needs both kinds from one hardware, the strict ACTIVE requirement wins. + controllers = [_controller_state("mixed", + required_command=["joint1/position"], + required_state=["joint1/velocity"])] + hardware = [_hardware_component("RRBot", + command=["joint1/position"], + state=["joint1/velocity"])] + rule = ComponentStateMonitor.infer_dependency_rules(controllers, hardware)[0] + self.assertEqual(rule.required_hardware[0].state, LifecycleState.ACTIVE) + + def test_owner_resolved_by_interface_name_not_prefix(self): + # rrbot_joint1/position is owned by whichever hardware exports it, not by rrbot_joint1. + controllers = [_controller_state("pos_ctrl", required_command=["rrbot_joint1/position"])] + hardware = [_hardware_component("RRBotSystemPositionOnly", + command=["rrbot_joint1/position"])] + rule = ComponentStateMonitor.infer_dependency_rules(controllers, hardware)[0] + self.assertEqual([h.name for h in rule.required_hardware], ["RRBotSystemPositionOnly"]) + + def test_interface_with_no_owning_hardware_gives_no_dependency(self): + # An interface no hardware exports (e.g. hardware not up yet) yields no dependency. + controllers = [_controller_state("lonely", required_command=["ghost/iface"])] + hardware = [_hardware_component("RRBot", command=["joint1/position"])] + rule = ComponentStateMonitor.infer_dependency_rules(controllers, hardware)[0] + self.assertEqual(rule.required_hardware, []) + + +class TestDependencyRulesQuery(unittest.TestCase): + """Tests dependency rules are inferred from the latest stored observations.""" + + @classmethod + def setUpClass(cls): + rclpy.init() + + @classmethod + def tearDownClass(cls): + rclpy.shutdown() + + def setUp(self): + self.node = rclpy.create_node("test_component_state_monitor") + self.node.callback_group_services = MutuallyExclusiveCallbackGroup() + self.node.callback_group_subscriber = ReentrantCallbackGroup() + self.addCleanup(self.node.destroy_node) + self.engine = Mock() + self.monitor = ComponentStateMonitor(self.node, self.engine, "test_controller_manager", []) + + def test_no_observations_yields_no_rules(self): + # Before any async refresh has stored observations, there are no rules. + self.assertEqual(self.monitor.get_dependency_rules(), []) + + def test_rules_inferred_from_latest_observations(self): + # Simulate the async refresh callbacks having stored the latest observations. + self.monitor._latest_hardware_components = [ + _hardware_component("RRBot", command=["joint1/position"])] + self.monitor._latest_controllers = [ + _controller_state("forward_position_controller", required_command=["joint1/position"])] + + rules = self.monitor.get_dependency_rules() + + self.assertEqual(len(rules), 1) + self.assertEqual(rules[0].controller_name, "forward_position_controller") + self.assertEqual(rules[0].required_hardware[0].name, "RRBot") + self.assertEqual(rules[0].required_hardware[0].state, LifecycleState.ACTIVE) + + +if __name__ == '__main__': + unittest.main() diff --git a/foreman/test/test_engine.py b/foreman/test/test_engine.py index 15bb366..8aa2911 100644 --- a/foreman/test/test_engine.py +++ b/foreman/test/test_engine.py @@ -6,8 +6,10 @@ from foreman.parser import ParsedScenario from foreman.types import Component from foreman.types import ComponentType +from foreman.types import ControllerDependencyRule from foreman.types import ForemanError from foreman.types import ForemanErrorCategory +from foreman.types import HardwareRequirement from foreman.types import LifecycleState from foreman.types import SystemGoal @@ -278,3 +280,79 @@ def test_goal_accepted_when_dependency_already_satisfied(dependency_config): response = engine.request_goal('active') assert response.success is True + + +@pytest.fixture +def inferred_rules_config(): + """Goal wants 'gripper' active, but the config declares no dependency rules.""" + + goal = SystemGoal('run', + controller_goals=[Component('gripper', ComponentType.CONTROLLER, LifecycleState.ACTIVE)]) + return ParsedScenario( + hardware=["hw1"], + dependency_rules=[], + goals={'run': goal}, + tracked_components={"hw1", "gripper"} + ) + + +class _FakeDependencyProvider: + """Simple dependency rule provider for planner tests.""" + + def __init__(self): + self.rules = [] + + def get_dependency_rules(self): + return self.rules + + +def test_pulled_rules_change_what_the_planner_allows(inferred_rules_config): + """Planner decisions change when the provider returns different rules.""" + + lock = threading.Lock() + engine = ForemanEngine(inferred_rules_config, lock) + provider = _FakeDependencyProvider() + engine.set_dependency_provider(provider) + + # hw1 is only INACTIVE, gripper is INACTIVE, and the source has no rules yet. + engine.set_system_state([ + Component('hw1', ComponentType.HARDWARE, LifecycleState.INACTIVE), + Component('gripper', ComponentType.CONTROLLER, LifecycleState.INACTIVE), + ]) + engine.request_goal('run') + + cmd = engine.get_next_transition() + assert cmd is not None + assert cmd.component.name == 'gripper' + assert cmd.goal_state == LifecycleState.ACTIVE + + # Update the provider to require hw1 ACTIVE before gripper activation. + provider.rules = [ControllerDependencyRule( + controller_name='gripper', + required_hardware=[HardwareRequirement('hw1', LifecycleState.ACTIVE)])] + + # The newly pulled rule blocks activation because hw1 is not ACTIVE. + assert engine.get_next_transition() is None + + +def test_controller_with_no_dependency_rule_is_satisfiable(inferred_rules_config): + """A controller named in a goal but absent from the pulled rules has no + requirements, so the goal is accepted instead of crashing on a missing rule.""" + lock = threading.Lock() + engine = ForemanEngine(inferred_rules_config, lock) + + provider = _FakeDependencyProvider() + # The provider knows a rule, but for a DIFFERENT controller, so 'gripper' has no rule. + provider.rules = [ControllerDependencyRule( + controller_name='new_controller', + required_hardware=[HardwareRequirement('hw1', LifecycleState.ACTIVE)])] + engine.set_dependency_provider(provider) + + engine.set_system_state([ + Component('hw1', ComponentType.HARDWARE, LifecycleState.INACTIVE), + Component('gripper', ComponentType.CONTROLLER, LifecycleState.INACTIVE), + ]) + + # 'gripper' has no rule, so no hardware requirements and goal must be accepted, not crash. + response = engine.request_goal('run') + assert response.success is True diff --git a/foreman/test/test_parser.py b/foreman/test/test_parser.py index a4aeaf0..da2dca7 100644 --- a/foreman/test/test_parser.py +++ b/foreman/test/test_parser.py @@ -6,8 +6,6 @@ from foreman.parser import ParsedScenario from foreman.types import Component from foreman.types import ComponentType -from foreman.types import ControllerDependencyRule -from foreman.types import HardwareRequirement from foreman.types import LifecycleState from foreman.types import SystemGoal @@ -36,40 +34,9 @@ def test_lifecycle_nodes_list(self, parsed_scenario): def test_metadata_empty(self, parsed_scenario): assert parsed_scenario.metadata == {} - -class TestDependencyRules: - """Tests for parsed dependency rules.""" - - def test_rules_count(self, parsed_scenario): - assert len(parsed_scenario.dependency_rules) == 3 - - def test_joint_state_broadcaster_rule(self, parsed_scenario): - rule = next(r for r in parsed_scenario.dependency_rules if r.controller_name == - "joint_state_broadcaster") - assert rule.controller_name == "joint_state_broadcaster" - assert len(rule.required_hardware) == 3 - reqs_by_name = {req.name: req for req in rule.required_hardware} - assert set(reqs_by_name.keys()) == { - "FrankaHardwareInterface", "kassow", "dummy_lifecycle_node"} - assert reqs_by_name["kassow"].state == LifecycleState.INACTIVE - assert reqs_by_name["FrankaHardwareInterface"].state == LifecycleState.INACTIVE - assert reqs_by_name["dummy_lifecycle_node"].state == LifecycleState.ACTIVE - - def test_kassow_jtc_rule(self, parsed_scenario): - rule = next(r for r in parsed_scenario.dependency_rules if r.controller_name == - "kassow_joint_trajectory_controller") - assert rule.controller_name == "kassow_joint_trajectory_controller" - assert len(rule.required_hardware) == 1 - assert rule.required_hardware[0].name == "kassow" - assert rule.required_hardware[0].state == LifecycleState.ACTIVE - - def test_franka_jtc_rule(self, parsed_scenario): - rule = next(r for r in parsed_scenario.dependency_rules if r.controller_name == - "franka_joint_trajectory_controller") - assert rule.controller_name == "franka_joint_trajectory_controller" - assert len(rule.required_hardware) == 1 - assert rule.required_hardware[0].name == "FrankaHardwareInterface" - assert rule.required_hardware[0].state == LifecycleState.ACTIVE + def test_dependency_rules_not_parsed(self, parsed_scenario): + """Dependencies are inferred at runtime, never parsed from YAML.""" + assert parsed_scenario.dependency_rules == [] class TestGoalStates: diff --git a/foreman/test/test_planner.py b/foreman/test/test_planner.py index acdfba8..db8ebe6 100644 --- a/foreman/test/test_planner.py +++ b/foreman/test/test_planner.py @@ -55,6 +55,22 @@ def apply_command(state: SystemState, cmd: SystemTransitionCommand): ) +def test_planner_pulls_rules_from_provider(basic_planner): + """Verify the planner uses rules returned by its provider.""" + class _FakeProvider: + def get_dependency_rules(self): + return [ControllerDependencyRule( + controller_name='new_ctrl', + required_hardware=[HardwareRequirement('new_hw', LifecycleState.ACTIVE)])] + + basic_planner.set_dependency_provider(_FakeProvider()) + rules = basic_planner.get_current_rules() + + # Rules now come from the provider instead of the initial configuration. + assert 'franka_jtc' not in rules + assert rules['new_ctrl'].required_hardware[0].name == 'new_hw' + + def test_scenario_1_standard_bring_up(basic_planner): state = SystemState(components={ 'franka_hw': Component('franka_hw', ComponentType.HARDWARE, LifecycleState.UNCONFIGURED),