ahp
b3243427b7
Complete implementation of the daena log/event collection service:
Core runtime:
- DaenaRuntime service lifecycle (start/stop/reload)
- Pipeline orchestrator with source/persist/dispatch flow
- LMDB persistence backend with key-prefix partitioning
- Plugin discovery via Python entry points
Pipeline:
- Source/Processor/Sink ABC interfaces with DeliveryResult
- DummySource for testing and development
- DummySink and HTTPSink implementations
- PassthroughProcessor
- Record dataclass for in-memory records
- StoredRecord model with RecordState tracking
REST API (FastAPI, v1):
- GET /system/{status,health,config}
- GET /plugins
- GET /queue, /queue/{pending,dead}
- GET /sinks
- Versioned under /api/v1 prefix
CLI (Typer):
- status, health, queue, sinks, plugins, config, version
Persistence (LMDB):
- Single-database key-prefix design (no named DBs)
- Index keys: idx:{state}:{ts}\x00{id}
- Record keys: rec:{id}
- Full CRUD: put/get/ack/nack/dead/update_state
- Pending/retry/dead-letter queries with state filtering
- delete_old for TTL-based retention
Deployment:
- systemd unit file (deploy/daena.service)
- Example YAML config (config/daena.yaml)
- Entry points: daena (CLI), daena-serve (uvicorn server)
Quality:
- 23 unit tests covering config, persistence, pipeline
- ruff linting and formatting pass clean
- Py3.13 compatible, uv-managed, pyenv-pinned
- GPG-signed commit on dev branch
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from daena.pipeline.base import DeliveryResult, Processor
|
|
from daena.pipeline.processors.passthrough import PassthroughProcessor
|
|
from daena.pipeline.record import Record
|
|
from daena.pipeline.sinks.dummy import DummySink
|
|
from daena.pipeline.sinks.http import HTTPSink
|
|
from daena.pipeline.sources.dummy import DummySource
|
|
|
|
|
|
def test_dummy_source_config() -> None:
|
|
source = DummySource(
|
|
name="test",
|
|
config={"interval": 1.0, "message": "hello"},
|
|
)
|
|
assert source.name == "test"
|
|
assert source._interval == 1.0
|
|
assert source._message == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dummy_sink_delivery() -> None:
|
|
sink = DummySink(name="test")
|
|
record = Record(source="test", record_type="test", body={"key": "value"})
|
|
result = await sink.deliver(record)
|
|
assert isinstance(result, DeliveryResult)
|
|
assert result.success is True
|
|
assert result.record_id == record.record_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passthrough_processor() -> None:
|
|
proc = PassthroughProcessor(name="test")
|
|
record = Record(source="test", record_type="test", body={"key": "value"})
|
|
result = await proc.process(record)
|
|
assert result is not None
|
|
assert result.record_id == record.record_id
|
|
assert result.body == {"key": "value"}
|
|
|
|
|
|
def test_http_sink_requires_url() -> None:
|
|
sink = HTTPSink(name="test", config={"url": "http://example.com/ingest"})
|
|
assert sink._url == "http://example.com/ingest"
|
|
assert sink._method == "POST"
|
|
|
|
|
|
def test_record_has_id() -> None:
|
|
r1 = Record(source="s1", record_type="t1", body={})
|
|
r2 = Record(source="s2", record_type="t2", body={})
|
|
assert r1.record_id != r2.record_id
|
|
assert r1.timestamp is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_processor_drop_record() -> None:
|
|
class DropProcessor(Processor):
|
|
async def process(self, record: Record) -> Record | None: # noqa: ARG002
|
|
return None
|
|
|
|
proc = DropProcessor(name="dropper")
|
|
record = Record(source="test", record_type="test", body={})
|
|
result = await proc.process(record)
|
|
assert result is None
|