[uss_qualifier] Add badge concept

monitoring/monitorlib/schema_validation.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from typing import List, Dict
 
-import jsonpath_ng
+import bc_jsonpath_ng
 import jsonschema.validators
 import yaml
 
@@ -80,7 +80,7 @@ def validate(
     resolver = jsonschema.validators.RefResolver(
         base_uri=f"{Path(base_path).as_uri()}/", referrer=openapi_content
     )
-    schema_matches = jsonpath_ng.parse(object_path).find(openapi_content)
+    schema_matches = bc_jsonpath_ng.parse(object_path).find(openapi_content)
     if len(schema_matches) != 1:
         raise ValueError(
             f"Found {len(schema_matches)} matches to JSON path '{object_path}' within OpenAPI definition at {openapi_path} when expecting exactly 1 match"

monitoring/uss_qualifier/README.md

@@ -32,3 +32,11 @@ At this point, uss_qualifier can be run again with a different configuration tar
 * [Test scenarios](scenarios/README.md) (includes test case, test step, check breakdown)
 * [Test configurations](configurations/README.md)
 * [Test resources](resources/README.md)
+
+### Badges
+
+A test suite may define one or more "badges" and the criteria necessary for a participant to earn them.  Badges are intended to be used to indicate whether a participant has satisfied the requirements for an optional capability.
+
+For instance, ASTM F3411-22a network remote identification includes the concept of an "intent-based network participant" who provides information about their overall flight intent, but not real-time telemetry.  A NetRID Service Provider is not required to support intent-based network participants, but if they do choose to support them, then they must follow a few requirements.  In this case, an "intent-based network participant support" badge may be defined to be granted only if uss_qualifier confirms compliance with the requirements regarding intent-based network participants.  If compliance to the intent-based network participants cannot be confirmed, the NetRID Service Provider under test may still be fully compliant with the standard; they just elected not to support this particular optional capability and would therefore be standard-compliant without earning the "intent-based network participant support" badge.
+
+As another example, ASTM F3411-22a does not require a NetRID Display Provider to provide Display Application clients with the serial number of an identified aircraft.  However, a particular jurisdiction may accept F3411-22a remote identification as a means of compliance only if Display Applications are capable of providing the viewer with the aircraft serial number.  In this case, the ASTM F3411-22a test suite could define a "serial number" badge indicating that a Display Provider provides the correct aircraft serial number to its Display Application clients, and then the test suite for the jurisdiction may define a "qualified in jurisdiction" badge which is not granted unless the ASTM F3411-22a test suite's "serial number" badge is granted (among other criteria).

monitoring/uss_qualifier/fileio.py

@@ -3,7 +3,7 @@
 import os
 from typing import Tuple, Optional, Dict, List, Union
 
-import jsonpath_ng
+import bc_jsonpath_ng
 import requests
 import yaml
 
@@ -235,7 +235,7 @@ def _replace_refs(
     cache: Optional[Dict[str, dict]] = None,
 ) -> None:
     for path in ref_parent_paths:
-        parent = [m.value for m in jsonpath_ng.parse(path).find(content)]
+        parent = [m.value for m in bc_jsonpath_ng.parse(path).find(content)]
         if len(parent) != 1:
             raise RuntimeError(
                 f'Unexpectedly found {len(parent)} matches for $ref parent JSON Path "{path}"'
@@ -247,7 +247,7 @@ def _replace_refs(
                 ref_path, context_file_name, cache
             )
         else:
-            ref_json_path = jsonpath_ng.parse(
+            ref_json_path = bc_jsonpath_ng.parse(
                 ref_path.replace("#", "$").replace("/", ".")
             )
             ref_content = [m.value for m in ref_json_path.find(content)]
@@ -264,7 +264,8 @@ def _replace_refs(
             allof_parent_path = ".".join(path.split(".")[0:-1])
             if allof_parent_path + ".allOf" in allof_paths:
                 allof_parent_content = [
-                    m.value for m in jsonpath_ng.parse(allof_parent_path).find(content)
+                    m.value
+                    for m in bc_jsonpath_ng.parse(allof_parent_path).find(content)
                 ]
                 if len(allof_parent_content) != 1:
                     raise RuntimeError(

monitoring/uss_qualifier/reports/badge_definitions.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+from typing import Optional, List
+
+from implicitdict import ImplicitDict
+from monitoring.uss_qualifier.requirements.definitions import RequirementCollection
+
+
+BadgeID = str
+"""Identifier of a badge."""
+
+JSONPathExpression = str
+"""JsonPath expression; see https://pypi.org/project/jsonpath-ng/"""
+
+
+class SpecificCondition(ImplicitDict):
+    pass
+
+
+class AllConditions(SpecificCondition):
+    """Condition will only be satisfied when all specified conditions are satisfied."""
+
+    conditions: List[BadgeGrantCondition]
+
+
+class AnyCondition(SpecificCondition):
+    """Condition will be satisfied when any of the specified conditions are satisfied."""
+
+    conditions: List[BadgeGrantCondition]
+
+
+class RequirementsCheckedCondition(SpecificCondition):
+    """Condition will only be satisfied if at least one successful check exists for all specified requirements."""
+
+    checked: RequirementCollection
+    """Each requirement contained within this collection must be covered by at least one successful check."""
+
+
+class NoFailedChecksCondition(SpecificCondition):
+    """Condition will only be satisfied if there are no applicable failed checks.
+
+    For a badge granted to a participant, only checks including the participant's ID will be considered."""
+
+    pass
+
+
+class BadgeGrantedCondition(SpecificCondition):
+    """Condition will be satisfied when the specified badge is granted."""
+
+    badge_id: BadgeID
+    """Identifier of badge that must be granted for this condition to be satisifed."""
+
+    badge_location: Optional[JSONPathExpression]
+    """Location of report to inspect for the presence of the specified badge, relative to the report in which the badge
+    is defined.  Implicit default value is "$" (look for granted batch in the report in which the dependant badge is
+    defined).
+
+    If this location resolves to multiple TestSuiteReports, then the badge must be granted in all resolved reports in
+    order for this condition to be satisfied.  When this location resolves to artifacts that are not TestSuiteReports,
+    those artifacts will be ignored.
+
+    Note that badges are evaluated in the order they are defined.  So, if badge B defined in a particular location
+    depends on whether badge A in that same location is granted, badge A must be defined before badge B is defined.
+    Also note that badges are computed as test components are completed.  Since a parent test component (e.g., test
+    suite) is not complete until all of its child components are complete, a descendant test component's badge condition
+    cannot depend on whether an ancestor's (e.g., parent's) badge is granted.
+    """
+
+
+class BadgeGrantCondition(ImplicitDict):
+    """Specification of a single condition used to determine whether a badge should be granted.
+
+    Exactly one field must be specified."""
+
+    all_conditions: Optional[AllConditions]
+    any_conditions: Optional[AnyCondition]
+    no_failed_checks: Optional[NoFailedChecksCondition]
+    requirements_checked: Optional[RequirementsCheckedCondition]
+    badge_granted: Optional[BadgeGrantedCondition]
+
+
+class ParticipantBadgeDefinition(ImplicitDict):
+    id: BadgeID
+    """Identifier of this badge, unique at the level in which this badge is defined."""
+
+    name: str
+    """Human-readable name of the badge"""
+
+    description: str
+    """Human-readable description of the achievement granting of the badge indicates."""
+
+    grant_condition: BadgeGrantCondition
+    """Condition required in order to grant the badge."""

monitoring/uss_qualifier/reports/badges.py

@@ -0,0 +1,143 @@
+from typing import Dict, Callable, TypeVar, Type
+
+import bc_jsonpath_ng.ext
+
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
+from monitoring.uss_qualifier.reports.badge_definitions import (
+    AllConditions,
+    SpecificCondition,
+    AnyCondition,
+    NoFailedChecksCondition,
+    RequirementsCheckedCondition,
+    BadgeGrantedCondition,
+    BadgeGrantCondition,
+)
+from monitoring.uss_qualifier.reports.report import TestSuiteReport
+from monitoring.uss_qualifier.requirements.definitions import RequirementID
+from monitoring.uss_qualifier.requirements.documentation import (
+    resolve_requirements_collection,
+)
+
+SpecificConditionType = TypeVar("SpecificConditionType", bound=SpecificCondition)
+_badge_condition_evaluators: Dict[
+    Type, Callable[[SpecificConditionType, ParticipantID, TestSuiteReport], bool]
+] = {}
+
+
+def badge_condition_evaluator(condition_type: Type):
+    """Decorator to label a function that evaluates a specific condition for granting a badge.
+
+    Args:
+        condition_type: A Type that inherits from badge_definitions.SpecificCondition.
+    """
+
+    def register_evaluator(func):
+        _badge_condition_evaluators[condition_type] = func
+        return func
+
+    return register_evaluator
+
+
+def condition_satisfied_for_test_suite(
+    grant_condition: BadgeGrantCondition,
+    participant_id: ParticipantID,
+    report: TestSuiteReport,
+) -> bool:
+    """Determine if a condition for granting a badge is satisfied based on a Test Suite report.
+
+    Args:
+        grant_condition: Badge-granting condition to check.
+        participant_id: Participant for which the badge would be granted.
+        report: Test Suite report upon which the badge (and grant condition) are based.
+
+    Returns: True if the condition was satisfied, False if not.
+    """
+    populated_fields = [
+        field_name
+        for field_name in grant_condition
+        if grant_condition[field_name] is not None
+    ]
+    if not populated_fields:
+        raise ValueError(
+            "No specific condition specified for grant_condition in BadgeGrantCondition"
+        )
+    if len(populated_fields) > 1:
+        raise ValueError(
+            "Multiple conditions specified for grant_condition in BadgeGrantCondition: "
+            + ", ".join(populated_fields)
+        )
+    specific_condition = grant_condition[populated_fields[0]]
+    condition_evaluator = _badge_condition_evaluators.get(
+        type(specific_condition), None
+    )
+    if condition_evaluator is None:
+        raise RuntimeError(
+            f"Could not find evaluator for condition type {type(specific_condition).__name__}"
+        )
+    return condition_evaluator(specific_condition, participant_id, report)
+
+
+@badge_condition_evaluator(AllConditions)
+def evaluate_all_conditions_condition(
+    condition: AllConditions, participant_id: ParticipantID, report: TestSuiteReport
+) -> bool:
+    for subcondition in condition.conditions:
+        if not condition_satisfied_for_test_suite(subcondition, participant_id, report):
+            return False
+    return True
+
+
+@badge_condition_evaluator(AnyCondition)
+def evaluate_any_condition_condition(
+    condition: AnyCondition, participant_id: ParticipantID, report: TestSuiteReport
+) -> bool:
+    for subcondition in condition.conditions:
+        if condition_satisfied_for_test_suite(subcondition, participant_id, report):
+            return True
+    return False
+
+
+@badge_condition_evaluator(NoFailedChecksCondition)
+def evaluate_no_failed_checks_condition(
+    condition: NoFailedChecksCondition,
+    participant_id: ParticipantID,
+    report: TestSuiteReport,
+) -> bool:
+    for _ in report.query_failed_checks(participant_id):
+        return False
+    return True
+
+
+@badge_condition_evaluator(RequirementsCheckedCondition)
+def evaluate_requirements_checked_conditions(
+    condition: RequirementsCheckedCondition,
+    participant_id: ParticipantID,
+    report: TestSuiteReport,
+) -> bool:
+    req_checked: Dict[RequirementID, bool] = {
+        req_id: False for req_id in resolve_requirements_collection(condition.checked)
+    }
+    for passed_check in report.query_passed_checks(participant_id):
+        for req_id in passed_check.requirements:
+            if req_id in req_checked:
+                req_checked[req_id] = True
+    return all(req_checked.values())
+
+
+@badge_condition_evaluator(BadgeGrantedCondition)
+def evaluate_badge_granted_condition(
+    condition: BadgeGrantedCondition,
+    participant_id: ParticipantID,
+    report: TestSuiteReport,
+) -> bool:
+    path = condition.badge_location if "badge_location" in condition else "$"
+    matching_reports = bc_jsonpath_ng.ext.parse(path).find(report)
+    result = False
+    for matching_report in matching_reports:
+        if isinstance(matching_report.value, TestSuiteReport):
+            badges = matching_report.value.badges_granted.get(participant_id, set())
+            if condition.badge_id in badges:
+                result = True
+            else:
+                return False
+    return result

monitoring/uss_qualifier/reports/graphs.py

@@ -87,7 +87,7 @@ def _make_test_scenario_nodes(
 
     # Add failed checks and error below
     parent_node = scenario_node
-    for failed_check in report.get_all_failed_checks():
+    for failed_check in report.query_failed_checks():
         check_node = Node(
             name=namer.make_name(report.scenario_type + "FailedCheck"),
             label=failed_check.summary,