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.

  • ITHConfig and SubsystemConfig handle the static side: loading a YAML configuration file, resolving values from environment variables, and providing structured access to the result.

  • 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 Subsystem directly without involving the configuration system at all.

Configuration: ITHConfig and SubsystemConfig

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.

SubsystemConfig represents one entry in that list. Its two main utilities are device_trl(), which returns the TRL for a single named role, and devices_by_prefix(), which returns all roles sharing a common prefix (e.g., all subarray_* entries).

Subsystem: device proxy management

Subsystem takes a dictionary of role-to-TRL mappings and provides two operations: get_device() (single proxy by role key) and get_device_keys() /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 from_config() class method instantiates a Subsystem (or any subclass) directly from a 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.

# 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.

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.

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:

@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:

@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