diff --git a/.github/workflows/public-ci.yml b/.github/workflows/public-ci.yml index aeef9acd..cdeeb7ee 100644 --- a/.github/workflows/public-ci.yml +++ b/.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: diff --git a/src/tests/common/test_lock_redis.py b/src/tests/common/test_lock_redis.py index bbb232a2..59c9c3c6 100644 --- a/src/tests/common/test_lock_redis.py +++ b/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). diff --git a/src/tests/temporal/ngc/workflows/test_ib_pkey_lock.py b/src/tests/temporal/ngc/workflows/test_ib_pkey_lock.py index d9eae1c3..1f18bdc4 100644 --- a/src/tests/temporal/ngc/workflows/test_ib_pkey_lock.py +++ b/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