daena/tests/test_dispatch.py

from __future__ import annotations

import asyncio
from pathlib import Path

import pytest
import yaml

from daena.config import (
    DaenaConfig,
    HealthConfig,
    PipelineConfig,
    ProcessorConfig,
    SinkConfig,
    SinkRetryConfig,
    SourceConfig,
    load_config,
)
from daena.core.runtime import DaenaRuntime
from daena.persistence.models import RecordState


def _runtime(tmp_path: Path, **overrides: object) -> DaenaRuntime:
    config = DaenaConfig()
    config.persistence.lmdb.path = str(tmp_path / "lmdb")
    config.pipeline = PipelineConfig(batch_size=10, flush_interval=0.05, max_queue_size=10_000)
    config.retention.interval_seconds = 3600
    for key, value in overrides.items():
        setattr(config, key, value)
    return DaenaRuntime(config)


@pytest.mark.asyncio
async def test_retry_does_not_block_until_success(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.sources = [
        SourceConfig(
            name="once",
            type="dummy",
            options={"interval": 0.05, "count": 1, "message": "retry"},
        )
    ]
    runtime.config.sinks = [
        SinkConfig(
            name="out",
            type="dummy",
            retry=SinkRetryConfig(max_attempts=3, min_delay=0, max_delay=0, multiplier=1),
            options={"fail_times": 1},
        )
    ]
    await runtime.start()
    delivered = 0
    for _ in range(40):
        assert runtime.backend is not None
        delivered = runtime.backend.count(RecordState.delivered)
        if delivered == 1:
            break
        await asyncio.sleep(0.05)
    dead = runtime.backend.count(RecordState.dead) if runtime.backend else -1
    await runtime.stop()
    assert delivered == 1
    assert dead == 0


@pytest.mark.asyncio
async def test_sinks_keep_independent_delivery_state(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.sources = [
        SourceConfig(name="once", type="dummy", options={"interval": 0.05, "count": 1})
    ]
    runtime.config.sinks = [
        SinkConfig(name="good", type="dummy"),
        SinkConfig(name="bad", type="dummy", options={"fail_times": 5, "permanent": True}),
    ]
    await runtime.start()
    stored = None
    for _ in range(40):
        assert runtime.backend is not None
        records = runtime.backend.list_by_state(RecordState.dead, limit=10)
        if records and records[0].deliveries["good"].state == RecordState.delivered:
            stored = records[0]
            break
        await asyncio.sleep(0.05)
    await runtime.stop()
    assert stored is not None
    assert stored.deliveries["good"].state == RecordState.delivered
    assert stored.deliveries["bad"].state == RecordState.dead


@pytest.mark.asyncio
async def test_backpressure_stops_ingest(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.pipeline = PipelineConfig(batch_size=10, flush_interval=0.05, max_queue_size=2)
    runtime.config.sources = [
        SourceConfig(name="fast", type="dummy", options={"interval": 0.01, "message": "bp"})
    ]
    runtime.config.sinks = [SinkConfig(name="slow", type="dummy", options={"delay": 30})]
    await runtime.start()
    await asyncio.sleep(0.4)
    assert runtime.backend is not None
    stored = runtime.backend.count()
    await runtime.stop()
    assert stored <= 2


@pytest.mark.asyncio
async def test_disabled_source_emits_nothing(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.sources = [
        SourceConfig(name="off", type="dummy", enabled=False, options={"interval": 0.05})
    ]
    runtime.config.sinks = [SinkConfig(name="out", type="dummy")]
    await runtime.start()
    await asyncio.sleep(0.3)
    assert runtime.backend is not None
    total = runtime.backend.count()
    await runtime.stop()
    assert total == 0


@pytest.mark.asyncio
async def test_reload_rereads_file(tmp_path: Path) -> None:
    path = tmp_path / "daena.yaml"
    data = {
        "persistence": {"lmdb": {"path": str(tmp_path / "lmdb")}},
        "pipeline": {"batch_size": 10, "flush_interval": 0.05, "max_queue_size": 100},
        "retention": {"interval_seconds": 3600},
        "sources": [{"name": "live", "type": "dummy", "enabled": True, "options": {"interval": 1}}],
        "sinks": [{"name": "out", "type": "dummy", "enabled": True}],
    }
    path.write_text(yaml.dump(data), encoding="utf-8")
    runtime = DaenaRuntime(load_config(path), config_path=path)
    await runtime.start()
    data["sources"][0]["enabled"] = False
    path.write_text(yaml.dump(data), encoding="utf-8")
    await runtime.reload()
    assert runtime.config.sources[0].enabled is False
    assert runtime.running is True
    await runtime.stop()


@pytest.mark.asyncio
async def test_health_degrades_when_delivery_stalls(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.health = HealthConfig(max_pending_age_seconds=0)
    runtime.config.sources = [
        SourceConfig(name="fast", type="dummy", options={"interval": 0.05, "count": 1})
    ]
    runtime.config.sinks = [SinkConfig(name="slow", type="dummy", options={"delay": 30})]
    await runtime.start()
    await asyncio.sleep(1.2)
    report = runtime.health()
    await runtime.stop()
    assert report.status == "degraded"
    assert report.healthy is False
    assert report.oldest_pending_age_seconds >= 1


@pytest.mark.asyncio
async def test_filter_drops_record_before_persist(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    runtime.config.sources = [
        SourceConfig(
            name="once",
            type="dummy",
            options={"interval": 0.05, "count": 1, "message": "drop-me"},
        )
    ]
    runtime.config.processors = [
        ProcessorConfig(
            name="drop",
            type="filter",
            options={"drop": {"body.message": "drop-me #1"}},
        )
    ]
    runtime.config.sinks = [SinkConfig(name="out", type="dummy")]
    await runtime.start()
    await asyncio.sleep(0.3)
    assert runtime.backend is not None
    total = runtime.backend.count()
    await runtime.stop()
    assert total == 0