50578fdba8
Swaps the legacy OpcManager read path for the gateway's SubscriptionEngine over the shared OpcConnectionPool. The OPC client certificate now always comes from MinIO + Mongo (CertProvider, Basic256Sha256) — no more certs/ shipped in the Gitea repository, so a new version branch no longer breaks the reads. - io/slots.py: parse_slot accepts slot v2 (names[]) and converts the legacy v1 (tag_name + topics; cert_path => use_security, path dropped), so gateway and Orchestrator can be deployed in any order. - managers/server_reader.py: new per-server reader with the OpcManager interface the IngestorManager already drives (connect, create_subscription, subscribe, unsubscribe, check_cycles, check_opc_listenning, shutdown). - IngestorManager owns the read path: lazy-MinIO CertProvider, the pool and a read DataSink, built without touching the network. Slots from Redis are normalized on lease and refresh; a malformed slot is notified and the last good config is kept. - GatewayRuntime borrows the pool (writes/gRPC) instead of building it, and no longer closes it on stop: a Postgres/NATS failure can't stop the reads. - pool: pin/unpin per active subscription so release_idle never evicts a subscribed connection; public evict() for lost servers / changed config. - SubscriptionEngine pins on subscribe, unpins on unsubscribe, and drops a subscription locally even when the server-side delete fails. - DataSink: model_id only written when the name has one, keeping v1 documents identical to the legacy ones. - config: CERT_BUCKET default is the legacy streamlit-connectors bucket (D7). - Removes OpcManager, its tests and scripts/generate-and-push-opc-cert.sh; README/CLAUDE.md document the MinIO certificate and the slot v2 format. Tests: 218 unit tests pass; ruff check/format clean. Local end-to-end against an asyncua server: v1 slot with two topics -> one doc per collection, legacy doc shape, idle sweep keeps the subscription, lost-server shutdown and re-initialization work. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
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'},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_reading_without_model_id_keeps_legacy_doc():
|
|
"""Names converted from a slot v1 carry no model_id: the document must be the
|
|
same as the legacy OpcManager one (SIENTIAPDE-1974 NF2)."""
|
|
sink, dm = make_sink()
|
|
names = [{'collection': 'raw_x', 'tag_name': 'Counter'}]
|
|
|
|
await sink.publish_reading(
|
|
tag='ns=2;i=2', value=42, timestamp='2026-07-13 00:00:00+0000', names=names
|
|
)
|
|
|
|
collection, doc = dm.publish.await_args_list[0].args
|
|
assert collection == 'raw_x'
|
|
assert doc == {
|
|
'tag': 'ns=2;i=2',
|
|
'name': 'Counter',
|
|
'timestamp': '2026-07-13 00:00:00+0000',
|
|
'value': 42,
|
|
}
|
|
_, kwargs = sink.emit_metric.await_args
|
|
assert kwargs['tags']['model_id'] == ''
|