Commit fdb22321 authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

fix(test): make JetStream publisher test repeatable 

Reuse and delete the work-queue consumer instead of leaking a durable on every run. Purge only the deploy subject before publishing so stale test messages cannot be consumed, while leaving other OOP_TASKS subjects untouched. Assert the specified 24-hour max_age and work-queue retention.
parent 4284b939

tests/integration/test_nats_publisher.py

+33 −11
Original line number Diff line number Diff line 
import json

import os

from contextlib import suppress

from datetime import datetime, timezone

from uuid import uuid4



import pytest

from nats.js.api import RetentionPolicy



from federation_manager.adapters.databus.nats_adapter import NatsCommandPublisher

from federation_manager.contracts.srm import (

    SUBJECT_DEPLOY,

    TASK_STREAM,

    TASK_STREAM_MAX_AGE_SECONDS,

    DeployPayloadV1,

    DeployTargetV1,

    SrmServiceDeployV1,
@@ -37,11 +40,15 @@ async def test_deploy_command_round_trips_through_jetstream() -> None: 
        deploy=DeployPayloadV1(instance_name="video-es"),

    )



    # work-queue retention only drops acked messages, so start from a known-empty subject

    await publisher._require_js()._jsm.purge_stream(TASK_STREAM, subject=SUBJECT_DEPLOY)

    await publisher.publish(SUBJECT_DEPLOY, cmd.model_dump(mode="json"))



    sub = await publisher._require_js().pull_subscribe(

        SUBJECT_DEPLOY, durable=f"test-{op_id.hex[:8]}", stream=TASK_STREAM

    )

    # OOP_TASKS is a work queue: one consumer per subject, so reuse and drop a fixed durable.

    js = publisher._require_js()

    durable = "fm-test-deploy"

    try:

        sub = await js.pull_subscribe(SUBJECT_DEPLOY, durable=durable, stream=TASK_STREAM)

        msgs = await sub.fetch(1, timeout=5)

        await msgs[0].ack()
@@ -50,5 +57,20 @@ async def test_deploy_command_round_trips_through_jetstream() -> None: 
        assert received.source == "federation"

        assert received.targets[0].zone_id == zone

        assert received.service_specification_id == spec

    finally:

        with suppress(Exception):

            await js.delete_consumer(TASK_STREAM, durable)

        await publisher.close()





async def test_task_stream_carries_the_platform_age_limit() -> None:

    publisher = NatsCommandPublisher(URL)

    await publisher.connect()

    await publisher.ensure_task_stream()



    info = await publisher._require_js()._jsm.stream_info(TASK_STREAM)



    assert info.config.max_age == TASK_STREAM_MAX_AGE_SECONDS

    assert info.config.retention == RetentionPolicy.WORK_QUEUE



    await publisher.close()