diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py new file mode 100644 index 0000000..52b526c --- /dev/null +++ b/ingestor/io/data_sink.py @@ -0,0 +1,39 @@ +import ingestor.metrics as metrics + + +class DataSink: + """Gateway egress wrapper: fans a reading out to every configured name/collection. + + `postgres` is unused for now — task 12 adds `update_write_status` on top of it. + """ + + def __init__(self, data_manager, postgres, metadata: dict | None = None) -> None: + self.data_manager = data_manager + self.postgres = postgres + self.metadata = metadata + + async def emit_metric(self, **kwargs) -> None: + # ponytail: delegate to data_manager, it's already a SientiaMonitoring + await self.data_manager.emit_metric(**kwargs) + + async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None: + """Publish one MongoDB document per name entry and record its tag value metric.""" + for name in names: + doc = { + 'tag': tag, + 'name': name['tag_name'], + 'model_id': name['model_id'], + 'timestamp': timestamp, + 'value': value, + } + await self.data_manager.publish(name['collection'], doc) + await self.emit_metric( + metric_object=metrics.OPC_TAG_VALUE, + method='set', + value=value, + tags={ + 'pod_id': self.data_manager.pod_id, + 'tag_name': name['tag_name'], + 'model_id': name['model_id'], + }, + ) diff --git a/tests/unit/io/test_data_sink_publish.py b/tests/unit/io/test_data_sink_publish.py new file mode 100644 index 0000000..a4edfee --- /dev/null +++ b/tests/unit/io/test_data_sink_publish.py @@ -0,0 +1,69 @@ +from unittest.mock import AsyncMock + +import pytest + +import ingestor.metrics as metrics +from ingestor.io.data_sink import DataSink + + +def make_sink(data_manager=None): + dm = data_manager or AsyncMock() + sink = DataSink(data_manager=dm, postgres=AsyncMock()) + sink.emit_metric = AsyncMock() + return sink, dm + + +@pytest.mark.asyncio +async def test_publish_reading_fans_out_per_name(): + sink, dm = make_sink() + names = [ + {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'}, + {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'}, + ] + + await sink.publish_reading( + tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names + ) + + assert dm.publish.await_count == 2 + + +@pytest.mark.asyncio +async def test_publish_reading_builds_doc_and_routes_collection(): + sink, dm = make_sink() + names = [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'}] + + await sink.publish_reading( + tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names + ) + + collection, doc = dm.publish.await_args_list[0].args + assert collection == 'raw_x' + assert doc == { + 'tag': 'ns=2;i=2', + 'name': 'Counter', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + 'value': 42, + } + + +@pytest.mark.asyncio +async def test_publish_reading_emits_tag_value_metric_per_name(): + sink, dm = make_sink() + names = [ + {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'}, + {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'}, + ] + + await sink.publish_reading( + tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names + ) + + assert sink.emit_metric.await_count == 2 + sink.emit_metric.assert_any_await( + metric_object=metrics.OPC_TAG_VALUE, + method='set', + value=42, + tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'}, + )