LegacyHUB/tests/test_dispatch_inbox_consumer.py

"""C2: consume QMS §7 asset events from the dispatch service inbox.

AssetRegistered → build AssetManifestEnvelope from the event payload and run
knowledge-ingest; accepted/duplicate/rejected → confirm. AssetApproved and
unknown types → confirm (they carry no new work). Transient ingest failures →
NO confirm and stop: the inbox returns the same head message until confirmed,
so continuing would spin; the next cron run retries.
"""

from __future__ import annotations

import pytest

from app.api.schemas import AssetManifestEnvelope
from app.integrations.dispatch_inbox_consumer import (
    TransientIngestError,
    consume_inbox,
    manifest_envelope_from_event,
)


def _event(**overrides):
    event = {
        "event_type": "AssetRegistered",
        "event_version": "1.0",
        "owner_module": "qms",
        "asset_id": "512d5f53-6edf-58f3-93aa-2a387f38fcc0",
        "owner_record_type": "release_candidate",
        "owner_record_id": "rel-2026-001",
        "asset_kind": "document",
        "content_type": "application/pdf",
        "size_bytes": 619,
        "sha256": "f" * 64,
        "security": {"gate_status": "approved"},
        "manifest_ref": {
            "provider": "s3",
            "bucket": "teamhub-qmshub-releases",
            "object_key": "qmshub/2026/07/12/512d5f53-6edf-58f3-93aa-2a387f38fcc0/original/"
                          + "f" * 64 + ".pdf",
            "sha256": "f" * 64,
        },
        "actor": "qa@x.com",
    }
    event.update(overrides)
    return event


def _message(message_type="AssetRegistered", event=None):
    return {
        "message_id": "11111111-2222-3333-4444-555555555555",
        "message_type": message_type,
        "envelope": {
            "message_type": message_type,
            "participant_code": "qms",
            "card": {"event": event if event is not None else _event(event_type=message_type)},
        },
    }


# --- payload -> AssetManifestEnvelope ---

def test_manifest_envelope_from_event_is_valid_against_schema():
    envelope = manifest_envelope_from_event(_event())
    model = AssetManifestEnvelope.model_validate(envelope)
    assert model.manifest.asset.asset_id == "512d5f53-6edf-58f3-93aa-2a387f38fcc0"
    assert model.manifest.asset.owner_module == "qms"
    assert model.manifest.asset.sha256 == "f" * 64
    assert model.manifest.storage.bucket == "teamhub-qmshub-releases"
    assert model.manifest.security.gate_status == "approved"
    assert model.manifest.retention.policy_id


def test_manifest_envelope_rejects_incomplete_event():
    broken = _event()
    del broken["manifest_ref"]
    with pytest.raises(ValueError):
        manifest_envelope_from_event(broken)


# --- orchestration ---

class Recorder:
    def __init__(self, messages, ingest_results=None):
        self._messages = list(messages)
        self.ingested = []
        self.confirmed = []
        self._ingest_results = list(ingest_results or [])

    def fetch(self):
        return self._messages.pop(0) if self._messages else None

    def ingest(self, envelope):
        self.ingested.append(envelope)
        if self._ingest_results:
            result = self._ingest_results.pop(0)
            if isinstance(result, Exception):
                raise result
            return result
        return "accepted"

    def confirm(self, message_id):
        self.confirmed.append(message_id)


def test_registered_is_ingested_and_confirmed():
    rec = Recorder([_message("AssetRegistered")])
    summary = consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert len(rec.ingested) == 1 and len(rec.confirmed) == 1
    assert summary == {"ingested": 1, "confirmed": 1, "skipped": 0}


def test_approved_is_confirmed_without_ingest():
    rec = Recorder([_message("AssetApproved")])
    summary = consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert rec.ingested == [] and len(rec.confirmed) == 1
    assert summary["skipped"] == 1


def test_unknown_type_is_confirmed_to_avoid_head_of_line_blocking():
    rec = Recorder([_message("SomethingElse")])
    consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert rec.ingested == [] and len(rec.confirmed) == 1


def test_duplicate_and_rejected_still_confirm():
    rec = Recorder(
        [_message("AssetRegistered"), _message("AssetRegistered")],
        ingest_results=["duplicate", "rejected"],
    )
    summary = consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert len(rec.confirmed) == 2
    assert summary["ingested"] == 2


def test_transient_ingest_error_stops_without_confirm():
    rec = Recorder(
        [_message("AssetRegistered"), _message("AssetApproved")],
        ingest_results=[TransientIngestError("minio down")],
    )
    summary = consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert rec.confirmed == []          # head message left pending for retry
    assert summary["ingested"] == 0
    assert summary.get("stopped") == "transient_error"


def test_malformed_registered_event_is_confirmed_as_poison():
    """A permanently unmappable event must not block the queue forever."""
    bad = _message("AssetRegistered", event={"event_type": "AssetRegistered"})
    rec = Recorder([bad])
    summary = consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm)
    assert rec.ingested == []
    assert len(rec.confirmed) == 1
    assert summary["skipped"] == 1


def test_empty_inbox_returns_zero_summary():
    rec = Recorder([])
    assert consume_inbox(fetch=rec.fetch, ingest=rec.ingest, confirm=rec.confirm) == {
        "ingested": 0, "confirmed": 0, "skipped": 0,
    }