Configuration and Subsystems ------------------------------ .. important:: Before reading further, ask yourself: **do you actually need this?** Good test code is simple — and simplicity matters especially for tests, because tests are rarely tested themselves. If you are dealing with a small, fixed set of devices, the right choice is often just to define a few variables directly in your test code or fixture. Reach for these building blocks only when you genuinely have a configuration problem: many devices, several deployment targets, or Tango Resource Locators (**TRLs**) that change between environments. Conceptual definition ^^^^^^^^^^^^^^^^^^^^^^^^^^ These two building blocks address the problem of keeping *Tango Resource Locators* (**TRLs**, the string addresses that you use to connect to Tango devices - e.g., `sys/controller/1`) and other environment-specific details **out of your test logic**. - :py:class:`~ska_integration_test_harness.core.config.ITHConfig` and :py:class:`~ska_integration_test_harness.core.config.SubsystemConfig` handle the **static** side: loading a YAML configuration file, resolving values from environment variables, and providing structured access to the result. - :py:class:`~ska_integration_test_harness.core.subsystem.Subsystem` handles the **runtime** side: turning the static device locators from the configuration into live Tango device proxies, with lazy creation and caching. The two layers are designed to work together but can also be used independently. If you already have device TRLs available at runtime (e.g., from a fixture), you can instantiate a :py:class:`~ska_integration_test_harness.core.subsystem.Subsystem` directly without involving the configuration system at all. Configuration: ITHConfig and SubsystemConfig ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ :py:class:`~ska_integration_test_harness.core.config.ITHConfig` loads a YAML file and exposes its content as a structured Python object. The YAML format supports `Jinja2 `_ templates, so any value can be derived from an **environment variable** or computed dynamically at load time. This makes it straightforward to maintain a single configuration file that works across different deployment targets (e.g., ``mid`` vs ``low``, CI vs on-site). The predefined structure inside the YAML is minimal: a top-level ``subsystems`` list, where each entry maps a subsystem name to a set of device TRLs keyed by role (e.g., ``controller``, ``subarray_1``). Any additional fields — flags, target names, numeric parameters — are accepted as-is and accessible as attributes on the loaded object. :py:class:`~ska_integration_test_harness.core.config.SubsystemConfig` represents one entry in that list. Its two main utilities are :py:meth:`~ska_integration_test_harness.core.config.SubsystemConfig.device_trl`, which returns the TRL for a single named role, and :py:meth:`~ska_integration_test_harness.core.config.SubsystemConfig.devices_by_prefix`, which returns all roles sharing a common prefix (e.g., all ``subarray_*`` entries). Subsystem: device proxy management ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ :py:class:`~ska_integration_test_harness.core.subsystem.Subsystem` takes a dictionary of role-to-TRL mappings and provides two operations: :py:meth:`~ska_integration_test_harness.core.subsystem.Subsystem.get_device` (single proxy by role key) and :py:meth:`~ska_integration_test_harness.core.subsystem.Subsystem.get_device_keys` /:py:meth:`~ska_integration_test_harness.core.subsystem.Subsystem.get_devices` (all proxies, optionally filtered by prefix). Proxies are created on first access and cached thereafter, so repeated calls do not incur extra connection overhead. The class is designed to be subclassed. A typical pattern is to extend it with typed accessor methods for your specific subsystem (e.g., ``controller()``, ``subarray(n)``), keeping your test code free of raw string keys. The :py:meth:`~ska_integration_test_harness.core.subsystem.Subsystem.from_config` class method instantiates a ``Subsystem`` (or any subclass) directly from a :py:class:`~ska_integration_test_harness.core.config.SubsystemConfig`. A custom device proxy factory can be injected at construction time, which is useful for unit-testing the subsystem wrapper itself without a live Tango bus. Usage Example: Multi-target Telescope Wrapper ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ This example shows how to combine a YAML configuration file, a small set of ``Subsystem`` subclasses, and a ``Telescope`` wrapper to produce test code that is completely agnostic to the deployment target. **The configuration file.** The YAML file below drives two environment variables: ``TELESCOPE_SHORT`` (``"mid"`` or ``"low"``) and the optional ``AGNOSTIC_MODE`` flag. Jinja2 resolves them at load time, so the resulting ``ITHConfig`` object always reflects the actual deployment. Notably, the ``dishes`` subsystem appears only for Mid, and ``mccs_lmc`` appears only for Low — and only when ``AGNOSTIC_MODE`` is not set. The Python code never inspects ``target`` or ``agnostic_mode`` directly: it simply checks which subsystems are present. .. code-block:: jinja # Set via CI job variables (e.g. TELESCOPE_SHORT=mid or TELESCOPE_SHORT=low) target: {{ TELESCOPE_SHORT }} agnostic_mode: {{ AGNOSTIC_MODE | default("false") }} subsystems: - name: tmc devices: central_node: "{{ TELESCOPE_SHORT }}-tmc/central-node/0" subarray_1: "{{ TELESCOPE_SHORT }}-tmc/subarray/01" subarray_2: "{{ TELESCOPE_SHORT }}-tmc/subarray/02" subarray_3: "{{ TELESCOPE_SHORT }}-tmc/subarray/03" - name: csp_lmc devices: controller: "{{ TELESCOPE_SHORT }}-csp/control/0" subarray_1: "{{ TELESCOPE_SHORT }}-csp/subarray/01" subarray_2: "{{ TELESCOPE_SHORT }}-csp/subarray/02" subarray_3: "{{ TELESCOPE_SHORT }}-csp/subarray/03" - name: sdp_lmc devices: controller: "{{ TELESCOPE_SHORT }}-sdp/control/0" subarray_1: "{{ TELESCOPE_SHORT }}-sdp/subarray/01" subarray_2: "{{ TELESCOPE_SHORT }}-sdp/subarray/02" subarray_3: "{{ TELESCOPE_SHORT }}-sdp/subarray/03" {% if TELESCOPE_SHORT == "mid" and not (AGNOSTIC_MODE | default("false")) %} - name: dishes devices: dish_1: "mid-dish/dish-manager/SKA001" dish_36: "mid-dish/dish-manager/SKA036" dish_63: "mid-dish/dish-manager/SKA063" dish_100: "mid-dish/dish-manager/SKA100" {% elif TELESCOPE_SHORT == "low" and not (AGNOSTIC_MODE | default("false")) %} - name: mccs_lmc devices: controller: "low-mccs/control/control" subarray_1: "low-mccs/subarray/01" subarray_2: "low-mccs/subarray/02" subarray_3: "low-mccs/subarray/03" {% endif %} **Subsystem wrappers.** Rather than one class per subsystem type, we define a small hierarchy based on structure. ``SubarraySubsystem`` covers any subsystem that has a ``subarray_N`` naming convention (TMC, CSP, SDP, MCCS), adding typed accessors for subarray devices and their observation state. ``TMC`` overrides the controller accessor name (``central_node`` instead of ``controller``) to match the TMC naming convention. ``DishesControllers`` handles the dish-specific ``dish_N`` naming. .. code-block:: python import tango from ska_control_model import ObsState from ska_integration_test_harness.core.subsystem import Subsystem class SubarraySubsystem(Subsystem): """Any subsystem whose devices follow the 'subarray_N' convention.""" def subarray_device(self, index: int) -> tango.DeviceProxy: return self.get_device(f"subarray_{index}") def subarray_indices(self) -> list[int]: keys = self.get_device_keys(prefix="subarray_") return [int(k.split("_")[-1]) for k in keys] def observation_state(self, subarray_index: int) -> ObsState: return ObsState(self.subarray_device(subarray_index).obsState) def controller_device(self) -> tango.DeviceProxy: return self.get_device("controller") class TMC(SubarraySubsystem): """TMC wrapper — uses 'central_node' instead of 'controller'.""" def central_node(self) -> tango.DeviceProxy: return self.get_device("central_node") def telescope_state(self) -> tango.DevState: return self.central_node().telescopeState class DishesControllers(Subsystem): """Wrapper for the Mid dish-manager devices.""" def dish_numbers(self) -> list[int]: return [int(k.split("_")[-1]) for k in self.get_device_keys("dish_")] def controller_device(self, dish_number: int) -> tango.DeviceProxy: return self.get_device(f"dish_{dish_number}") **The Telescope wrapper.** The constructor reads the configuration and instantiates whichever subsystems are present. It never inspects ``target`` or ``agnostic_mode`` — that logic already lives in the YAML template. The two helpers ``subarray_subsystems()`` and ``subarray_indices()`` are the key methods that make test steps target-agnostic: they return *only the subsystems that were actually configured*, with no hard-coded list. .. code-block:: python from ska_integration_test_harness.core.config import ITHConfig class Telescope: def __init__(self, config: ITHConfig) -> None: # TMC is always required self.tmc = TMC.from_config(config.subsystem("tmc")) # The remaining subsystems are optional — present only if the # YAML template included them for the current target/mode. self.csp = ( SubarraySubsystem.from_config(config.subsystem("csp_lmc")) if config.has_subsystem("csp_lmc") else None ) self.sdp = ( SubarraySubsystem.from_config(config.subsystem("sdp_lmc")) if config.has_subsystem("sdp_lmc") else None ) self.mccs = ( SubarraySubsystem.from_config(config.subsystem("mccs_lmc")) if config.has_subsystem("mccs_lmc") else None ) self.dishes = ( DishesControllers.from_config(config.subsystem("dishes")) if config.has_subsystem("dishes") else None ) def subarray_subsystems( self, include_main: bool = True ) -> list[SubarraySubsystem]: """All subsystems that expose subarray devices. Returns only those actually configured — automatically excluding, for example, ``dishes`` (which has no subarrays) and any subsystem absent from the YAML for the current target. Setting ``include_main=False`` omits TMC, which is useful when you want to iterate over the *other* subsystems only. """ candidates = ( ([self.tmc] if include_main else []) + [self.csp, self.sdp, self.mccs] ) return [s for s in candidates if s is not None] def subarray_indices(self) -> list[int]: """Subarray indices as defined in the TMC configuration.""" return self.tmc.subarray_indices() **Generic, target-agnostic test steps.** With the wrapper in place, individual test steps can be written once and run unchanged against Mid, Low, or agnostic deployments. The event subscription step iterates over ``subarray_subsystems()`` and ``subarray_indices()`` to cover every subarray in every relevant subsystem — whatever the YAML says is configured: .. code-block:: python @pytest.fixture def subscribe_subarray_events( telescope: Telescope, event_tracer: TangoEventTracer, ) -> None: """Subscribe to obsState events for all subarrays in all subsystems.""" for subsystem in telescope.subarray_subsystems(include_main=True): for index in telescope.subarray_indices(): event_tracer.subscribe_event( subsystem.subarray_device(index), "obsState" ) The verification step then asserts the expected transition in TMC and in each of the other subsystems present in the configuration: .. code-block:: python @then(parsers.parse( "subarray {subarray_label} transitions to {to_state}" )) def subarray_transitions_to_obsstate( event_tracer: TangoEventTracer, telescope: Telescope, subarray_label: str, to_state: str, ) -> None: """Verify all relevant subsystems report the expected obsState.""" index = int(subarray_label) target = ObsState[to_state] # Verify the TMC subarray first assert_that(event_tracer).described_as( f"Subarray {index} reached obsState {to_state} in TMC." ).within_timeout(MEDIUM_TIMEOUT).has_change_event_occurred( device_name=telescope.tmc.subarray_device(index), attribute_name="obsState", attribute_value=target, ) # Then verify every other subarray subsystem that is configured. # This loop is identical for Mid (CSP + SDP + Dishes), # Low (CSP + SDP + MCCS), and agnostic mode (CSP + SDP only). # No ``if target == "mid"`` branching required. for subsystem in telescope.subarray_subsystems(include_main=False): assert_that(event_tracer).described_as( f"Subarray {index} reached obsState {to_state}" f" in {subsystem.subsystem_name}." ).within_timeout(MEDIUM_TIMEOUT).has_change_event_occurred( device_name=subsystem.subarray_device(index), attribute_name="obsState", attribute_value=target, ) The key observation is that **no test step contains any conditional logic about the deployment target**. The configuration file is the single place where Mid and Low differ. Once loaded, the ``Telescope`` object exposes a uniform interface, and all test code — subscriptions, assertions, state navigation — works identically across targets. When to use (and when not to) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ These building blocks are worth their weight when: - You have **many devices** or **dynamically numbered groups** (e.g., a variable number of subarrays) that would otherwise require scattered hardcoded TRLs. - You need to **run the same tests against different deployments** and want to switch targets by changing an environment variable rather than editing test code. - You want a **single place** to document and review which devices a test suite connects to. They are likely unnecessary when: - You have a handful of fixed devices — just define them as variables. - Your TRLs never change between runs. - Adding a YAML file and a loader would be more code than the problem it solves. The guiding principle is the same as for any test infrastructure: **keep it only as complex as the problem demands**. Test code that is hard to understand is also hard to trust. API reference ^^^^^^^^^^^^^^ - :py:mod:`ska_integration_test_harness.core.config` - :py:mod:`ska_integration_test_harness.core.subsystem`