update tests

.github/workflows/public-ci.yml

@@ -144,6 +144,19 @@ jobs:
     runs-on: linux-amd64-cpu16
     permissions:
       contents: read
+    services:
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      REDIS_HOST: localhost
+      REDIS_PORT: "6379"
     steps:
       - uses: *actions_checkout
         with:

src/tests/common/test_lock_redis.py

@@ -24,6 +24,7 @@
 import contextlib
 import os
 import time
+import uuid
 from collections.abc import AsyncIterator
 
 import pytest
@@ -39,14 +40,25 @@
 TTL_S = 30
 
 
+def _unique_key(label: str) -> str:
+    """Per-run lock key so parallel workers and reruns never share a Redis key."""
+    return f"wf-lock:test:{label}:{uuid.uuid4().hex}"
+
+
 @pytest.fixture
 async def redis_lock_backend(monkeypatch) -> AsyncIterator[None]:
-    """Point the lock helpers at a real Redis, or skip when none is available."""
-    host = os.environ.get("REDIS_HOST")
+    """Point the lock helpers at a real Redis, or skip when none is available.
+
+    ``REDIS_HOST`` / ``REDIS_PORT`` mark an explicitly configured Redis (CI). A
+    ping failure there must fail the test. ``LOCK_REDIS_TEST`` is the local
+    opt-in path and may skip when Redis cannot be reached.
+    """
+    configured_host = os.environ.get("REDIS_HOST")
     port = int(os.environ.get("REDIS_PORT", "6379"))
     container = None
+    host = configured_host
 
-    if not host:
+    if not configured_host:
         if not os.environ.get("LOCK_REDIS_TEST"):
             pytest.skip("Set REDIS_HOST or LOCK_REDIS_TEST=1 to run the Redis-backed lock test")
         try:
@@ -67,6 +79,9 @@ async def redis_lock_backend(monkeypatch) -> AsyncIterator[None]:
         await client.close()
         if container is not None:
             container.stop()
+        # Configured CI Redis must fail loudly; local opt-in may skip.
+        if configured_host:
+            raise
         pytest.skip(f"Redis not reachable for lock test: {exc}")
 
     # Force the Redis-backed path (bypass the local no-op).
@@ -100,10 +115,11 @@ def _kinds_in_time_order(events: list[tuple[str, str, float]]) -> list[str]:
 async def test_same_key_holders_are_serialized(redis_lock_backend):
     """Two concurrent holders of one key never overlap their critical sections."""
     events: list[tuple[str, str, float]] = []
+    key = _unique_key("same")
 
     await asyncio.gather(
-        _critical_section("wf-lock:ngc:pkey=0x0005", "run-a", events),
-        _critical_section("wf-lock:ngc:pkey=0x0005", "run-b", events),
+        _critical_section(key, "run-a", events),
+        _critical_section(key, "run-b", events),
     )
 
     assert _kinds_in_time_order(events) == ["enter", "exit", "enter", "exit"]
@@ -115,8 +131,8 @@ async def test_distinct_keys_run_concurrently(redis_lock_backend):
     events: list[tuple[str, str, float]] = []
 
     await asyncio.gather(
-        _critical_section("wf-lock:ngc:pkey=0x0005", "run-a", events),
-        _critical_section("wf-lock:ngc:pkey=0x0006", "run-b", events),
+        _critical_section(_unique_key("distinct-a"), "run-a", events),
+        _critical_section(_unique_key("distinct-b"), "run-b", events),
     )
 
     assert _kinds_in_time_order(events) == ["enter", "enter", "exit", "exit"]
@@ -125,7 +141,7 @@ async def test_distinct_keys_run_concurrently(redis_lock_backend):
 @pytest.mark.asyncio
 async def test_renew_extends_holder_and_release_frees_key(redis_lock_backend):
     """A holder can renew its own lock, and release lets another token take it."""
-    key = "wf-lock:ngc:pkey=0x0007"
+    key = _unique_key("renew")
 
     assert await acquire_lock(key, "run-a", timeout=TTL_S, blocking_timeout=5)
     # The holder can extend its own TTL.
@@ -142,7 +158,7 @@ async def test_renew_extends_holder_and_release_frees_key(redis_lock_backend):
 @pytest.mark.asyncio
 async def test_renew_fails_when_lock_lost_to_another_holder(redis_lock_backend):
     """Renewing a key owned by a different token reports the loss."""
-    key = "wf-lock:ngc:pkey=0x0008"
+    key = _unique_key("lost")
 
     assert await acquire_lock(key, "run-b", timeout=TTL_S, blocking_timeout=5)
     # run-a never held it, so it cannot renew (and must not steal a held lock).

src/tests/temporal/ngc/workflows/test_ib_pkey_lock.py

@@ -17,6 +17,9 @@
 import pytest
 from pydantic import BaseModel
 
+from nv_config_manager.temporal.common.lock import WorkflowLockSpec, build_workflow_lock_key
+from nv_config_manager.temporal.common.mixins.metadata import WorkflowMetadataMixin
+from nv_config_manager.temporal.ngc.workflows import REGISTERED_WORKFLOWS
 from nv_config_manager.temporal.ngc.workflows._ib_pkey_lock import UFMHostLockMixin
 
 
@@ -38,3 +41,69 @@ async def test_canonicalizes_host_before_run(mocker):
 
     assert result.host == "10.0.0.5"
     assert result.pkey == "0x0100"
+
+
+@pytest.mark.asyncio
+async def test_host_spellings_collapse_to_one_lock_key(mocker):
+    """Name and IP that resolve to the same device produce one lock key.
+
+    Guards the end-to-end contract the lock relies on: canonicalization at the API
+    boundary plus deterministic key construction must serialize equivalent hosts.
+    """
+    mocker.patch(
+        "nv_config_manager.temporal.ngc.activities.ib_nautobot.canonicalize_ufm_host",
+        new=mocker.AsyncMock(return_value="10.0.0.5"),
+    )
+    spec = WorkflowLockSpec(key_fields=["host", "pkey"])
+
+    by_name = await UFMHostLockMixin.canonicalize_input(_HostInput(host="ufm01"))
+    by_ip = await UFMHostLockMixin.canonicalize_input(_HostInput(host="10.0.0.5"))
+
+    key_from_name = build_workflow_lock_key(
+        spec, workflow_name="IBPKeyMemberAdd", namespace="ngc", workflow_input=by_name
+    )
+    key_from_ip = build_workflow_lock_key(
+        spec, workflow_name="IBPKeyMemberAdd", namespace="ngc", workflow_input=by_ip
+    )
+
+    assert key_from_name == key_from_ip == "wf-lock:ngc:host=10.0.0.5:pkey=0x0100"
+
+
+def _host_keyed_locked_workflows() -> list[type]:
+    """Registered workflows whose lock key includes the host field."""
+    keyed: list[type] = []
+    for workflow_class in REGISTERED_WORKFLOWS:
+        spec = getattr(workflow_class, "workflow_lock", None)
+        if spec is not None and "host" in spec.key_fields:
+            keyed.append(workflow_class)
+    return keyed
+
+
+@pytest.mark.parametrize("workflow_class", _host_keyed_locked_workflows())
+@pytest.mark.asyncio
+async def test_host_keyed_lock_requires_canonicalization(workflow_class, mocker):
+    """A workflow that locks on ``host`` must collapse name and IP to one value.
+
+    Inheritance alone is not enough: a no-op override would still let equivalent
+    hosts map to different lock keys. Assert ``canonicalize_input`` produces the
+    same host for both spellings on every registered host-keyed workflow.
+    """
+    assert (
+        workflow_class.canonicalize_input.__func__
+        is not WorkflowMetadataMixin.canonicalize_input.__func__
+    ), (
+        f"{workflow_class.__name__} locks on 'host' but does not override "
+        "canonicalize_input; its lock key would not be canonical"
+    )
+
+    canonical_host = "10.0.0.5"
+    mocker.patch(
+        "nv_config_manager.temporal.ngc.activities.ib_nautobot.canonicalize_ufm_host",
+        new=mocker.AsyncMock(return_value=canonical_host),
+    )
+    input_class = workflow_class.get_workflow_input_class()
+    by_name = input_class(host="ufm01", pkey="0x0100", guids=["0000:0000:0000:0001"])
+    by_ip = input_class(host=canonical_host, pkey="0x0100", guids=["0000:0000:0000:0001"])
+
+    assert (await workflow_class.canonicalize_input(by_name)).host == canonical_host
+    assert (await workflow_class.canonicalize_input(by_ip)).host == canonical_host