[monitorlib] Move IDGenerator to monitorlib

monitoring/monitorlib/README.md

@@ -37,3 +37,20 @@ AuthAdapter's `__init__` constructor.  Both ordinal (e.g.,
 
 Use [`get_access_token.py`](../get_access_token.py) to retrieve an access token
 using a provided auth spec.
+
+## Test resource IDs
+
+[ids.py](ids.py) provides `register_resource_type` and `IDFactory` for creating
+recognizable UUIDv4-format test IDs.  Register each resource type with a code
+that is unique across all monitoring tools.  The comment above
+`register_resource_type` identifies the next available code; update it after
+allocating codes.  `IDFactory` encodes the owner, resource type code, and an
+instance index from 0 to 255.  The default index of 0 preserves legacy IDs.
+
+If a test resource (subscription, operational intent, etc.) is orphaned in a DSS,
+use [decode_id.py](decode_id.py) to identify its owner and origin.  From the
+repository root, run:
+
+```shell
+uv run python -m monitoring.monitorlib.decode_id <ID>
+```

monitoring/monitorlib/decode_id.py

@@ -0,0 +1,27 @@
+import argparse
+import sys
+
+import monitoring.uss_qualifier.scenarios
+from monitoring.monitorlib import ids
+from monitoring.monitorlib.inspection import import_submodules
+
+
+def parse_args(argv: list[str]):
+    parser = argparse.ArgumentParser(description="Decode a test ID")
+    parser.add_argument(
+        action="store", dest="id", type=str, metavar="ID", help="The ID to decode"
+    )
+    return parser.parse_args(argv)
+
+
+if __name__ == "__main__":
+    args = parse_args(sys.argv[1:])
+    import_submodules(monitoring.uss_qualifier.scenarios)
+    decoded = ids.IDFactory.decode_details(args.id)
+    id_code = decoded.resource_type
+    print(f"Owner: {decoded.owner_name}")
+    print(f"Instance index: {decoded.instance_index}")
+    if id_code in ids.resource_type_code_descriptions:
+        print(f"Resource type: {ids.resource_type_code_descriptions[id_code]}")
+    else:
+        print(f"Resource type: Unregistered ({id_code})")

monitoring/monitorlib/ids.py

@@ -1,5 +1,9 @@
 import hashlib
+import inspect
+import math
+import os
 import uuid
+from dataclasses import dataclass
 
 mac = uuid.getnode()
 
@@ -27,3 +31,220 @@ def make_id(code: str) -> uuid.UUID:
     digest[-2] = 0
     digest[-1] = 0
     return uuid.UUID(bytes=bytes(digest), version=4)
+
+
+ID_CHAR_SET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz_"
+MAX_OWNER_LENGTH = 12  # characters
+
+
+def bin_to_hex(bin_string):
+    return format(int(bin_string, 2), "02x")
+
+
+def bin_to_dec(bin_string):
+    return int(bin_string, 2)
+
+
+def hex_to_bin(hex_string):
+    return format(int(hex_string, 16), "08b")
+
+
+def dec_to_bin(num):
+    return format(num, "06b")
+
+
+def split_by(string_val, num=8):
+    """Splits a string into substrings with a length of given number."""
+    return [string_val[i : i + num] for i in range(0, len(string_val), num)]
+
+
+def get_ord_val(letter):
+    """Encodes new ord value for ascii letters."""
+    v = ID_CHAR_SET.find(letter)
+    if v == -1:
+        v = 63
+    return v
+
+
+def get_ascii_val_from_bit_value(num):
+    """Decodes new ord value to ascii letters."""
+    return ID_CHAR_SET[num] if num < 63 else "?"
+
+
+def encode_owner(owner_name: str) -> str:
+    """Encode an owner name as an 18-character hexadecimal string"""
+    bits = ""
+    if len(owner_name) > MAX_OWNER_LENGTH:
+        string_val = (
+            owner_name[: math.ceil(MAX_OWNER_LENGTH / 2)]
+            + owner_name[-math.floor(MAX_OWNER_LENGTH / 2) :]
+        )
+    elif len(owner_name) < MAX_OWNER_LENGTH:
+        string_val = ("?" * (MAX_OWNER_LENGTH - len(owner_name))) + owner_name
+    else:
+        string_val = owner_name
+    for letter in string_val:
+        ord_val = get_ord_val(letter)
+        bits += dec_to_bin(ord_val)
+    return "".join(bin_to_hex(s) for s in split_by(bits))
+
+
+def encode_resource_type_code(resource_type: int) -> str:
+    """Encode a number between 0 and 0xFFFF as a 4-character hexadecimal string"""
+    return format(resource_type, "04x")
+
+
+def decode_owner(owner_id: str) -> str:
+    """Decode an owner name, marking a truncated final character as unknown.
+
+    IDFactory preserves only the first 17 of the original 18 hexadecimal digits.
+    """
+    truncated = len(owner_id) == 17
+    if truncated:
+        owner_id += "0"
+    elif len(owner_id) != 18:
+        raise ValueError("Invalid owner id.")
+    hex_splits = split_by(owner_id, num=2)
+    bits = ""
+    for h in hex_splits:
+        bits += hex_to_bin(h)
+    test_owner = ""
+    for seq in split_by(bits, 6):
+        num = bin_to_dec(seq)
+        test_owner += get_ascii_val_from_bit_value(num)
+    if truncated:
+        test_owner = test_owner[:-1] + "?"
+    if test_owner[0] != "?":
+        return (
+            test_owner[: math.ceil(MAX_OWNER_LENGTH / 2)]
+            + ".."
+            + test_owner[-math.floor(MAX_OWNER_LENGTH / 2) :]
+        )
+    if truncated:
+        return test_owner[:-1].lstrip("?") + "?"
+    return test_owner.lstrip("?")
+
+
+def decode_resouce_type(id: str) -> int:
+    """Decode a number between 0 and 0xFFFF from a 4-character hex string"""
+    return int(id, 16)
+
+
+ResourceType = int
+resource_type_code_descriptions: dict[ResourceType, str] = {}
+
+
+# Next code: 406
+def register_resource_type(code: int, description: str) -> ResourceType:
+    """Register that the specified code refers to the described resource.
+
+    Args:
+      code: An integer that is globally-unique among all register_resource_type
+            calls across all monitoring tools, referring to this resource type.
+      description: Description of the resource created from this type.
+
+    Returns:
+      ResourceType that can be used to create an ID with <IDFactory>.make_id.
+    """
+    test_filename = inspect.stack()[1].filename
+    this_folder = os.path.dirname(os.path.abspath(__file__))
+    test = os.path.relpath(test_filename, this_folder)
+    full_description = f"{test}: {description}"
+    if code in resource_type_code_descriptions:
+        raise ValueError(
+            f'Resource type code {code} is already in use as "{resource_type_code_descriptions[code]}" so it cannot be used for "{full_description}"'
+        )
+    resource_type_code_descriptions[code] = full_description
+    return code
+
+
+@dataclass(frozen=True)
+class DecodedTestID:
+    owner_name: str
+    resource_type: ResourceType
+    instance_index: int
+
+
+class IDFactory:
+    """Creates UUIDv4 (as described in RFC4122) formatted IDs encoding the kind of ID and owner.
+
+    Format: 000HXXXX-YYYY-4LYY-8YYY-YYYYYYYY0000
+
+    Note that a UUID V4 needs to conform to the following regexp, which is why we
+    set the first digit of the fourth group to 8.
+
+    ^[0-9a-fA-F]{8}\\-[0-9a-fA-F]{4}\\-4[0-9a-fA-F]{3}\\-[8-b][0-9a-fA-F]{3}\\-[0-9a-fA-F]{12}$
+
+    XXXX encodes the kind of ID according to resource_type_code_descriptions.
+    YYYYYYYYYYYYYYYYY encodes the owner/creator of the resource having the ID and
+    consists of 11 complete characters and 2 bits of a twelfth character,
+    encoded as 6-bit groups.  Decoding marks the incomplete character with '?'.
+    HL encodes an instance index from 0 to 255 in hexadecimal.  Index 0 produces
+    the original ID format; other indices separate repeated invocations of a test.
+    """
+
+    owner_id: str
+    _instance_index: int
+
+    def __init__(self, test_owner: str, instance_index: int = 0):
+        self.validate_instance_index(instance_index)
+        self.owner_id = encode_owner(test_owner)
+        self._instance_index = instance_index
+
+    @staticmethod
+    def validate_instance_index(instance_index: int) -> None:
+        if type(instance_index) is not int or not 0 <= instance_index <= 255:
+            raise ValueError("instance_index must be an integer from 0 to 255")
+
+    @property
+    def instance_index(self) -> int:
+        return self._instance_index
+
+    def make_id(self, resource_type: ResourceType):
+        """Make a test ID with the specified resource type code"""
+        return f"000{self.instance_index >> 4:x}{encode_resource_type_code(resource_type)}-{self.owner_id[0:4]}-4{self.instance_index & 0xF:x}{self.owner_id[4:6]}-8{self.owner_id[6:9]}-{self.owner_id[9:17]}0000"
+
+    @classmethod
+    def decode(cls, id: str) -> tuple[str, ResourceType]:
+        """Decode an ID, retaining the original two-item return value."""
+        decoded = cls.decode_details(id)
+        return decoded.owner_name, decoded.resource_type
+
+    @classmethod
+    def decode_details(cls, id: str) -> DecodedTestID:
+        """Decode the owner, resource type, and instance index of a test ID."""
+        hex_digits = id.replace("-", "").lower()
+        if len(hex_digits) != 32:
+            raise ValueError(f"ID {id} has the wrong number of characters for a UUID")
+        if any(c not in "0123456789abcdef" for c in hex_digits):
+            raise ValueError(f"ID {id} contains non-hexadecimal characters")
+        if hex_digits[0:3] != "000" or hex_digits[-4:] != "0000":
+            raise ValueError(
+                f"ID {id} does not have the leading and trailing zeros indicating a test ID"
+            )
+        if hex_digits[12] != "4" or hex_digits[16] != "8":
+            raise ValueError(f"ID {id} is not formatted like a v4 test ID")
+        x = hex_digits[4:8]
+        y = hex_digits[8:12] + hex_digits[14:16] + hex_digits[17:28]
+        resource_type_code = ResourceType(int(x, 16))
+        owner_name = decode_owner(y)
+        return DecodedTestID(
+            owner_name=owner_name,
+            resource_type=resource_type_code,
+            instance_index=int(hex_digits[3] + hex_digits[13], 16),
+        )
+
+
+check_scd_write_subscription_id = register_resource_type(
+    344, "Subscription created by build/dev/check_scd_write.sh"
+)
+check_scd_write_constraint_id = register_resource_type(
+    345, "Constraint reference created by build/dev/check_scd_write.sh"
+)
+check_scd_write_operational_intent_id = register_resource_type(
+    346, "Operational intent reference created by build/dev/check_scd_write.sh"
+)
+
+unknown_resource_id = register_resource_type(
+    404, "<Unknown resource ID / Unknown purpose>"
+)

monitoring/prober/README.md

@@ -143,7 +143,7 @@ Specifically:
    obtained using the `ids` fixture providing a registered resource type code.
    E.g.:
    ```python
-   from monitoring.prober.infrastructure import register_resource_type
+   from monitoring.monitorlib.ids import register_resource_type
 
    [...]
    CONFLICTING_OI_TYPE = register_resource_type(103, "OI that conflicts with first OI")
@@ -153,9 +153,9 @@ Specifically:
    def test_create_conflict(ids):
        conflicting_id = ids(CONFLICTING_OI_TYPE)
    ```
-   The resource type code must be unique among all prober tests. To determine
+   The resource type code must be unique across all monitoring tools. To determine
    an appropriate resource type code for a new resource, see the comment above
-   `register_resource_type` in [infrastructure.py](infrastructure.py), and then
+   `register_resource_type` in [ids.py](../monitorlib/ids.py), and then
    make sure to update that comment after adding new resource type codes.
 1. Tests should be designed to pass even with pre-existing resources which may
    even be in the same area/time range/etc as the test resources.  Toward this
@@ -165,6 +165,5 @@ Specifically:
 The guidelines above should allow the prober to be used on production DSS
 instances without any nominal adverse effects.
 
-If an orphaned resource with a test ID is found in a DSS, the owner and origin
-of that resource can be determined from the ID with
-[decode_id.py](decode_id.py).
+For decoding orphaned test resource IDs, see the
+[monitorlib ID documentation](../monitorlib/README.md#test-resource-ids).

monitoring/prober/conftest.py

@@ -7,17 +7,13 @@
 import uas_standards.astm.f3548.v21.constants as v21_constants
 
 from monitoring.monitorlib import auth, scd
+from monitoring.monitorlib.ids import IDFactory, ResourceType
 from monitoring.monitorlib.infrastructure import (
     AsyncUTMTestSession,
     UTMClientSession,
     utm_client_session_factory,
 )
-from monitoring.prober.infrastructure import (
-    IDFactory,
-    ResourceType,
-    VersionString,
-    add_test_result,
-)
+from monitoring.prober.infrastructure import VersionString, add_test_result
 
 OPT_RID_AUTH = "rid_auth"
 OPT_RID_V2_AUTH = "rid_v2_auth"
@@ -252,7 +248,7 @@ def ids(pytestconfig, subscriber) -> Callable[[ResourceType], str]:
     This fixture is a function that accepts a ResourceType as the argument and
     returns a UUIDv4-format string containing an ID for that resource.
 
-    See register_resource_type in infrastructure.py for how to create
+    See register_resource_type in monitoring.monitorlib.ids for how to create
     ResourceTypes to provide to this fixture, and also the "Resources" section of
     the README.
     """