Extend tests to support new components_json field.

src/narrativelog/testutils.py

@@ -366,6 +366,12 @@ def random_message() -> MessageDictT:
     elif random_value > 0.25:
         date_begin = random_date()
 
+    test_components_json = {
+        "systems": random_strings(TEST_SYSTEMS),
+        "subsystems": random_strings(TEST_SUBSYSTEMS),
+        "components": random_strings(TEST_COMPONENTS),
+    }
+
     message = dict(
         id=None,
         site_id=TEST_SITE_ID,
@@ -396,6 +402,8 @@ def random_message() -> MessageDictT:
         primary_hardware_components=random_strings(
             TEST_PRIMARY_HARDWARE_COMPONENTS
         ),
+        # Added 2024-12-16
+        components_json=test_components_json,
         # Added 2023-10-24
         category=random_str(nchar=CATEGORY_LEN),
         time_lost_type=random.choice(["fault", "weather"]),
@@ -502,11 +510,14 @@ async def create_test_database(
             pruned_message = message.copy()
             del pruned_message["is_valid"]
             # Do not insert "components",
-            # "primary_software_components", or "primary_hardware_components"
+            # "primary_software_components",
+            # "primary_hardware_components",
+            # or "components_json"
             # because they are in a separate table.
             del pruned_message["components"]
             del pruned_message["primary_software_components"]
             del pruned_message["primary_hardware_components"]
+            del pruned_message["components_json"]
 
             # Insert the message
             result_message = await connection.execute(
@@ -529,6 +540,7 @@ async def create_test_database(
                     primary_hardware_components=message[
                         "primary_hardware_components"
                     ],
+                    components_json=message["components_json"],
                     message_id=data_message.id,
                 )
                 .returning(literal_column("*"))

tests/test_add_message.py

@@ -7,6 +7,7 @@
 
 from narrativelog.testutils import (
     TEST_COMPONENTS,
+    TEST_COMPONENTS_JSON,
     TEST_CSCS,
     TEST_PRIMARY_HARDWARE_COMPONENTS,
     TEST_PRIMARY_SOFTWARE_COMPONENTS,
@@ -90,6 +91,7 @@ async def test_add_message(self) -> None:
             add_args_full["primary_hardware_components"] = random_strings(
                 TEST_PRIMARY_HARDWARE_COMPONENTS
             )
+            add_args_full["components_json"] = TEST_COMPONENTS_JSON
             add_args_full["category"] = "test"
             add_args_full["time_lost_type"] = random.choice(
                 ["fault", "weather"]

tests/test_edit_message.py

@@ -85,6 +85,11 @@ async def test_edit_message(self) -> None:
                     "new primary_hardware_component 1",
                     "new primary_hardware_component 2",
                 ],
+                components_json={
+                    "systems": ["new system 1", "new system 2"],
+                    "subsystems": ["new subsystem 1", "new subsystem 2"],
+                    "components": ["new component 1", "new component 2"],
+                },
                 category="New category",
                 time_lost_type=random.choice(["fault", "weather"]),
                 date_begin="2024-01-01T00:00:00",

tests/test_find_messages.py

@@ -1,6 +1,7 @@
 import collections.abc
 import http
 import itertools
+import json
 import random
 import typing
 import unittest
@@ -341,6 +342,36 @@ def test_contains(
 
                 find_args_predicates.append(({field: value}, test_contains))
 
+            # "Contains" arguments for JSON fields: these specify a
+            # JSON path to match.
+            for field in ("components_json",):
+                # Scramble the messages and use the first
+                # message with at least a key with two values
+                messages_to_search = random.sample(messages, len(messages))
+                for message in messages_to_search:
+                    components_json = message[field]
+                    for key in components_json:
+                        if len(components_json[key]) >= 2:
+                            first_two_values = components_json[key][0:2]
+                            values = [(key, val) for val in first_two_values]
+                            path = json.dumps({key: first_two_values})
+                            break
+
+                @doc_str(f"{path!r} in message[{field!r}]")
+                def test_contains_path(
+                    message: MessageDictT,
+                    field: str = field,
+                    values: list[tuple[str, str]] = values,
+                ) -> bool:
+                    matches = [
+                        val in message[field][key] for key, val in values
+                    ]
+                    return any(matches)
+
+                find_args_predicates.append(
+                    ({"components_path": path}, test_contains_path)
+                )
+
             # has_<field> arguments (for fields that may be null).
             for field in (
                 "date_begin",