[uss_qualifier] Add resource environments

AGENTS.md

@@ -30,6 +30,7 @@ This document contains key context, nuances, and troubleshooting tips specifical
 - **Documentation Traceability**: All documented test checks must trace back to exactly one or more requirements using a specific bold format, and feature a severity emoji prefix (e.g., `## 🛑 Correct operational intent details check`). You must refer to `monitoring/uss_qualifier/scenarios/README.md` for specific markup details before modifying test steps.
 
 ## 4. Local Testing constraints
+- **Qualifier Working Directory**: Run resource-construction checks and qualifier resource tests from `monitoring/uss_qualifier` with the repository root on `PYTHONPATH`. External resource paths such as `file://./test_data/...` resolve relative to the current working directory, as they do in `run_locally.sh`.
 - **Docker Dependency**: Mock USS and DSS environments require active Docker containers. Standard testing commands are typically structured via bash scripts like `./monitoring/uss_qualifier/run_locally.sh <config>`. If container-building fails due to `Authentication` or package registry issues in the agent's environment, gracefully halt and ask the human user to run the script instead.
 
 ## 5. Continuous Improvement of this Guide

NEXT_RELEASE_NOTES.md

@@ -40,6 +40,8 @@ The release notes should contain at least the following sections:
 * For any uss_qualifier test configurations using any of the test suites `suites.astm.utm.f3548_21`, `suites.faa.uft.message_signing`, `suites.uspace.flight_auth`, or `suites.uspace.required_services`, the following resources must be changed:
     * `priority_preemption_flights` was previously an optional `resources.flight_planning.FlightIntentsResource`.  If this resource was previously provided to one of the above test suites, a new resource named `priority_preemption_flights_provider` must be provided instead.  This new resource must be a `resources.ResourceProvidingResource[resources.flight_planning.FlightIntentsResource]`, and one suitable concrete implementation is a `resources.geospatial.TriangularCascadeSoutheastResource`.  See [f3548_self_contained](./monitoring/uss_qualifier/configurations/dev/f3548_self_contained.yaml) for an example.
 
+* (uncommon) For any uss_qualifier test configurations directly using (i.e., not via the f3548_21 test suite) the scenario `scenarios.flight_planning.prep_planners.PrepareFlightPlannersScenario` or the scenario `scenarios.astm.utm.prep_planners.PrepareFlightPlanners`, the `flight_intents_provider` resource must be replaced with a `flight_intents_environments` that is a `resources.environments.ResourceEnvironmentsGenerator`; see [the f3548_21 test suite](./monitoring/uss_qualifier/suites/astm/utm/f3548_21.yaml) for an example.
+
 ## Optional migration tasks
 
 ## Important information

monitoring/uss_qualifier/action_generators/resource_environments.py

@@ -0,0 +1,80 @@
+from collections.abc import Iterator
+
+from implicitdict import ImplicitDict
+
+from monitoring.monitorlib.inspection import fullname
+from monitoring.uss_qualifier.action_generators.action_generator import ActionGenerator
+from monitoring.uss_qualifier.action_generators.documentation.definitions import (
+    PotentialGeneratedAction,
+)
+from monitoring.uss_qualifier.action_generators.documentation.documentation import (
+    list_potential_actions_for_action_declaration,
+)
+from monitoring.uss_qualifier.resources.definitions import ResourceID
+from monitoring.uss_qualifier.resources.environments import (
+    ResourceEnvironmentsGenerator,
+)
+from monitoring.uss_qualifier.resources.resource import (
+    MissingResourceError,
+    ResourceType,
+)
+from monitoring.uss_qualifier.suites.definitions import TestSuiteActionDeclaration
+from monitoring.uss_qualifier.suites.suite import TestSuiteAction
+
+
+class ResourceEnvironmentsActionGeneratorSpecification(ImplicitDict):
+    action_to_repeat: TestSuiteActionDeclaration
+    """Test suite action to run for each resource environment"""
+
+    resource_environments_source: ResourceID
+    """Resource providing different resource environments"""
+
+
+class ResourceEnvironmentsActionGenerator(
+    ActionGenerator[ResourceEnvironmentsActionGeneratorSpecification]
+):
+    _actions: list[TestSuiteAction]
+    _current_action: int
+
+    @classmethod
+    def list_potential_actions(
+        cls, specification: ResourceEnvironmentsActionGeneratorSpecification | None
+    ) -> list[PotentialGeneratedAction]:
+        if specification is None:
+            raise ValueError(f"{cls.__name__} requires a specification")
+        return list_potential_actions_for_action_declaration(
+            specification.action_to_repeat
+        )
+
+    @classmethod
+    def get_name(cls) -> str:
+        return "For each resource environment"
+
+    def __init__(
+        self,
+        specification: ResourceEnvironmentsActionGeneratorSpecification,
+        resources: dict[ResourceID, ResourceType],
+    ):
+        if specification.resource_environments_source not in resources:
+            raise MissingResourceError(
+                f"Resource ID {specification.resource_environments_source} specified as `resource_environments_source` was not present in the available resource pool",
+                specification.resource_environments_source,
+            )
+        environments_source = resources[specification.resource_environments_source]
+        if not isinstance(environments_source, ResourceEnvironmentsGenerator):
+            raise ValueError(
+                f"Expected resource ID {specification.resource_environments_source} to be a {fullname(ResourceEnvironmentsGenerator)} but it was a {fullname(environments_source.__class__)} instead"
+            )
+
+        self._actions = []
+        for resource_env in environments_source.get_environments():
+            modified_resources = resources | resource_env
+
+            self._actions.append(
+                TestSuiteAction(specification.action_to_repeat, modified_resources)
+            )
+
+        self._current_action = 0
+
+    def actions(self) -> Iterator[TestSuiteAction]:
+        yield from self._actions

monitoring/uss_qualifier/resources/README.md

@@ -32,6 +32,8 @@ Resources for a given test configuration are all declared in a single global res
 2. Every type of test resource must define a "resource specification", which is a serializable data type that fully defines how to create an instance of that resource type.
 3. Every type of test resource must define how to create an instance of the test resource from an instance of the resource specification.
 
+Dependency resource IDs may be suffixed with `?` to indicate that the dependency is optional.  Explicit constructor parameters must have a default value to accept optional dependencies.  Dependencies accepted through `**kwargs` are treated as having an implicit default of `None`.  Missing optional dependencies are omitted from the constructor call, so they do not appear in the `kwargs` dictionary.
+
 
 ## Resource-modifying resources
 

monitoring/uss_qualifier/resources/combinations.py

@@ -0,0 +1,89 @@
+from collections.abc import Iterator
+from dataclasses import dataclass
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.monitorlib.inspection import fullname
+from monitoring.uss_qualifier.resources.definitions import ResourceID
+from monitoring.uss_qualifier.resources.environments import (
+    ResourceEnvironmentSelectorResource,
+    ResourceEnvironmentsGenerator,
+)
+from monitoring.uss_qualifier.resources.plural import PluralResource
+from monitoring.uss_qualifier.resources.resource import Resource
+
+
+class ResourceCombinationsSpecification(ImplicitDict):
+    pass
+
+
+@dataclass
+class _RoleCombination:
+    role: ResourceID
+    resources: PluralResource
+    index: int
+
+
+class ResourceCombinationsResource(
+    ResourceEnvironmentsGenerator, Resource[ResourceCombinationsSpecification]
+):
+    """Resource that produces combinations of each dependent PluralResource according to ResourceIDs assigned by dependent resource keys.
+
+    For example, with a pre-existing dependent `flight_planners` FlightPlannersResource, a
+    ResourceCombinationsResource(tested_uss=flight_planners, control_uss=flight_planners) will produce each combination
+    of tested_uss=FP1, control_uss=FP2 for all FP1, FP2 combinations in flight_planners.
+    """
+
+    _resource_sources: dict[ResourceID, PluralResource]
+    _selector: Optional[ResourceEnvironmentSelectorResource]
+
+    def __init__(
+        self,
+        specification: ResourceCombinationsSpecification,
+        resource_origin: str,
+        combination_selector: Optional[ResourceEnvironmentSelectorResource] = None,
+        **dependencies,
+    ):
+        super().__init__(specification, resource_origin)
+        self._selector = combination_selector
+        self._resource_sources = dict()
+        for k, v in dependencies.items():
+            if not isinstance(v, PluralResource):
+                raise ValueError(
+                    f"Dependent resources for ResourceCombinationsResource must be PluralResources; dependent resource '{k}' was instead a {fullname(v.__class__)}"
+                )
+            if v.get_resource_instances_count() == 0:
+                # Can't fill this role because there are no resources to fill it with.
+                continue
+            self._resource_sources[ResourceID(k)] = v
+
+    def get_environments(self) -> Iterator[dict[ResourceID, Resource]]:
+        """Get the combinations of resources fulfilling each specified role."""
+        roles = [
+            _RoleCombination(role=k, resources=v, index=0)
+            for k, v in self._resource_sources.items()
+        ]
+
+        incremented = True
+        while incremented:
+            combination = {
+                role.role: role.resources.get_resource_instance(role.index)
+                for role in roles
+            }
+            if self._selector is None or self._selector.select_resource_environment(
+                combination
+            ):
+                yield combination
+
+            incremented = False
+            for r in range(len(roles)):
+                if (
+                    roles[r].index
+                    < roles[r].resources.get_resource_instances_count() - 1
+                ):
+                    roles[r].index += 1
+                    for r0 in range(r):
+                        roles[r0].index = 0
+                    incremented = True
+                    break

monitoring/uss_qualifier/resources/definitions.py

@@ -18,7 +18,7 @@ class ResourceDeclaration(ImplicitDict):
     """Type of resource, expressed as a Python class name qualified relative to this `resources` module (optionally with generic type arguments in brackets)"""
 
     dependencies: dict[ResourceID, ResourceID] = {}
-    """Mapping of dependency parameter (additional argument to concrete resource constructor) to `name` of resource to use (optionally suffixed with `?` to indicate an optional dependency when the constructor parameter has a default value)"""
+    """Mapping of dependency parameter (additional argument to concrete resource constructor) to `name` of resource to use (optionally suffixed with `?` to indicate an optional dependency when the constructor parameter has a default value or is accepted through **kwargs). Missing optional dependencies are omitted from the constructor call; dependencies accepted through **kwargs have an implicit default of None."""
 
     specification: dict = {}
     """Specification of resource; format is the SpecificationType that corresponds to the `resource_type`"""

monitoring/uss_qualifier/resources/environments.py

@@ -0,0 +1,126 @@
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.monitorlib.inspection import fullname
+from monitoring.uss_qualifier.resources.definitions import ResourceID
+from monitoring.uss_qualifier.resources.resource import (
+    Resource,
+    ResourceProvidingResource,
+    ValueResource,
+)
+
+
+class ResourceEnvironmentsGenerator(ABC):
+    @abstractmethod
+    def get_environments(self) -> Iterator[dict[ResourceID, Resource]]:
+        raise NotImplementedError()
+
+
+class ResourceEnvironmentSelectorSpecification(ImplicitDict):
+    """Specification defining how resource environments should be selected from the available/generated environments."""
+
+    pass
+
+
+class ResourceEnvironmentSelectorResource(
+    ValueResource[ResourceEnvironmentSelectorSpecification]
+):
+    def select_resource_environment(
+        self, resource_environment: dict[ResourceID, Resource]
+    ) -> bool:
+        """Returns True when the specified environment should be selected/used."""
+        return True
+
+
+class EnvironmentAugmentationResourceSpecification(ImplicitDict):
+    # In the future, this specification could be enhanced to affect how and whether providers are exercised.
+    pass
+
+
+class EnvironmentAugmentationResource(
+    ResourceEnvironmentsGenerator,
+    Resource[EnvironmentAugmentationResourceSpecification],
+):
+    """Presents as a ResourceEnvironmentsGenerator which augments the environments from an input `environments` resource
+    with additional resources named according to additional ResourceProvidingResource resource dependencies."""
+
+    _environment_generator: ResourceEnvironmentsGenerator
+    _providers: dict[ResourceID, ResourceProvidingResource]
+
+    def __init__(
+        self,
+        specification: EnvironmentAugmentationResourceSpecification,
+        resource_origin: str,
+        environments: ResourceEnvironmentsGenerator,
+        **providers,
+    ):
+        super().__init__(specification, resource_origin)
+
+        if not isinstance(environments, ResourceEnvironmentsGenerator):
+            raise ValueError(
+                f"The `provider` resource dependency for an {self.__class__.__name__} must be a ResourceEnvironmentsGenerator; found instead a {fullname(environments.__class__)}"
+            )
+        self._environment_generator = environments
+
+        self._providers = dict()
+        for k, v in providers.items():
+            if not isinstance(v, ResourceProvidingResource):
+                raise ValueError(
+                    f"All non-`provider` resource dependencies for an {self.__class__.__name__} must be ResourceProvidingResources, but the '{k}' resource dependency was instead a {fullname(environments.__class__)}"
+                )
+            self._providers[ResourceID(k)] = v
+
+    def get_environments(self) -> Iterator[dict[ResourceID, Resource]]:
+        for i, environment in enumerate(self._environment_generator.get_environments()):
+            augmentation = dict()
+            for resource_id, provider in self._providers.items():
+                augmentation[resource_id] = provider.provide_resource_for(
+                    index=i, environment=environment
+                )
+            yield environment | augmentation
+
+
+class ConcatenatedResourceEnvironmentsResourceSpecification(ImplicitDict):
+    environments_sequence: Optional[list[ResourceID]]
+    """Sequence in which each dependent ResourceEnvironmentsGenerator resource's environments are enumerated."""
+
+
+class ConcatenatedResourceEnvironmentsResource(
+    ResourceEnvironmentsGenerator,
+    Resource[ConcatenatedResourceEnvironmentsResourceSpecification],
+):
+    """Presents as a ResourceEnvironmentsGenerator which sequentially enumerates the environments from each dependent
+    resource in the specified sequence."""
+
+    _generators: list[ResourceEnvironmentsGenerator]
+
+    def __init__(
+        self,
+        specification: ConcatenatedResourceEnvironmentsResourceSpecification,
+        resource_origin: str,
+        **generators,
+    ):
+        super().__init__(specification, resource_origin)
+
+        for k, v in generators.items():
+            if not isinstance(v, ResourceEnvironmentsGenerator):
+                raise ValueError(
+                    f"All resource dependencies for a {fullname(self.__class__)} must be ResourceEnvironmentsGenerators, but the '{k}' resource dependency was instead a {fullname(v.__class__)}"
+                )
+
+        remaining_generators = generators.copy()
+        self._generators = []
+        if specification.environments_sequence:
+            for resource_id in specification.environments_sequence:
+                if resource_id in remaining_generators:
+                    self._generators.append(remaining_generators.pop(resource_id))
+
+        for generator in remaining_generators.values():
+            self._generators.append(generator)
+
+    def get_environments(self) -> Iterator[dict[ResourceID, Resource]]:
+        for generator in self._generators:
+            yield from generator.get_environments()