diff --git a/ingestor/adapters/__init__.py b/ingestor/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py new file mode 100644 index 0000000..ca2d5c2 --- /dev/null +++ b/ingestor/adapters/grpc_server.py @@ -0,0 +1,68 @@ +"""gRPC input adapter for on-demand tag reads (contract C4). + +Thin adapter: unpack the request, call ActiveReadEngine.read_now, map the +result onto ReadTagResponse. Internal-only for now -- no auth/TLS, per the +feature owner (no consumer exists yet, likely a frontend later). +""" + +import grpc +from sientia_do.observability.logger import Logger + +from ingestor.engines.active_read import ActiveReadEngine +from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc + + +class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer): + def __init__(self, active_read: ActiveReadEngine, logger: Logger, slot_lookup=None): + self.active_read = active_read + self.logger = logger + # ponytail: slot_lookup wiring (address -> names from the current slot) + # lands in Task 18 (app wiring); optional callable keeps this servicer usable now. + self.slot_lookup = slot_lookup + + async def ReadTag(self, request, context): + names = None + if self.slot_lookup is not None: + names = self.slot_lookup(request.server_id, request.addr) + try: + result = await self.active_read.read_now(request.server_id, request.addr, names=names) + except Exception as e: + self.logger.error(f'ReadTag failed for {request.server_id}:{request.addr}: {e}') + await context.abort(grpc.StatusCode.INTERNAL, f'read failed: {e}') + return + return opc_gateway_pb2.ReadTagResponse( + value=str(result['value']), + opc_status_code=result['opc_status_code'] or '', + source_timestamp=result['source_timestamp'] or '', + server_timestamp=result['server_timestamp'] or '', + persisted=result['persisted'], + ) + + +class GrpcServer: + def __init__( + self, + active_read: ActiveReadEngine, + logger: Logger, + port: int = 50051, + slot_lookup=None, + ): + self.active_read = active_read + self.logger = logger + self.port = port + self.slot_lookup = slot_lookup + self.server: grpc.aio.Server | None = None + + async def start(self) -> None: + self.server = grpc.aio.server() + opc_gateway_pb2_grpc.add_OpcGatewayServicer_to_server( + ReadTagServicer(self.active_read, self.logger, self.slot_lookup), self.server + ) + self.server.add_insecure_port(f'[::]:{self.port}') + await self.server.start() + self.logger.info(f'GrpcServer started on port {self.port}') + + async def stop(self) -> None: + if self.server is None: + return + await self.server.stop(grace=5) diff --git a/ingestor/adapters/nats_consumer.py b/ingestor/adapters/nats_consumer.py new file mode 100644 index 0000000..97ed2fe --- /dev/null +++ b/ingestor/adapters/nats_consumer.py @@ -0,0 +1,55 @@ +"""NATS input adapter for OPC write commands (contract C2). + +Thin adapter: decode the message JSON and hand the dict to WriteEngine, +which is already raise-safe against malformed commands. No metrics of its +own — the engine records write metrics; a malformed-JSON drop is logged +here. No reconnect logic — nats-py handles reconnection internally. +""" + +import json + +import nats +from nats.aio.client import Client +from nats.aio.subscription import Subscription +from sientia_do.observability.logger import Logger + +from ingestor.engines.write import WriteEngine + + +class NatsWriteConsumer: + def __init__( + self, + nats_url: str, + write_engine: WriteEngine, + logger: Logger, + subject: str = 'opc.write', + queue: str = 'gateway-writers', + ): + self.nats_url = nats_url + self.write_engine = write_engine + self.logger = logger + self.subject = subject + self.queue = queue + self.nc: Client | None = None + self.sub: Subscription | None = None + + async def start(self) -> None: + self.nc = await nats.connect(servers=self.nats_url) + self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle) + self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}') + + async def _handle(self, msg) -> None: + try: + payload = json.loads(msg.data) + except ValueError as e: # includes JSONDecodeError and UnicodeDecodeError + preview = msg.data[:200] + self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}') + return + await self.write_engine.write(payload) + + async def stop(self) -> None: + if self.nc is None: + return + if self.sub is not None: + await self.sub.unsubscribe() + await self.nc.drain() diff --git a/ingestor/converters.py b/ingestor/converters.py index f785136..d8b5951 100644 --- a/ingestor/converters.py +++ b/ingestor/converters.py @@ -25,6 +25,8 @@ def to_typed_value(raw, data_type: str): data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'. Raises ConversionError on any invalid input (never a bare TypeError). """ + if not isinstance(data_type, str): + raise ConversionError(f'data_type must be a string, got {type(data_type).__name__}') dtype = data_type.strip().lower() try: if dtype in ('bool', 'boolean'): diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py new file mode 100644 index 0000000..cb2cacd --- /dev/null +++ b/ingestor/engines/active_read.py @@ -0,0 +1,132 @@ +"""Active OPC reads (F3): on-demand (`read_now`, used by the gRPC server - +Task 16) and scheduled (`scheduled_tick`, driven by the app main loop each +collect cycle). Emits metrics only -- unlike WriteEngine there is no +notification path here.""" + +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE + +import ingestor.metrics as metrics +from ingestor.connection.clock import MonotonicClock +from ingestor.connection.pool import OpcConnectionPool +from ingestor.io.data_sink import DataSink + + +class ActiveReadEngine(SientiaMonitoring): + """On-demand + scheduled OPC reads, sharing the pool/sink with WriteEngine + and SubscriptionEngine.""" + + def __init__( + self, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ): + self.pool = pool + self.sink = sink + self.clock = MonotonicClock() + self._ticks: dict[tuple[str, str], int] = {} + + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + async def read_now( + self, + server_id: str, + addr: str, + names: list[dict] | None = None, + server_cfg: dict | None = None, + ) -> dict: + """Read `addr` on `server_id` right now; persist via sink only if `names` given.""" + tags = {'pod_id': self.pod_id, 'server_name': server_id} + start = self.clock.elapsed() + try: + client = await self.pool.get(server_id, server_cfg or {}) + dv = await client.get_node(addr).read_data_value() + except Exception as e: + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'error'} + ) + self.logger.error(f'Active read failed for {server_id}:{addr}: {e}') + raise + + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_LATENCY, + method='observe', + value=self.clock.elapsed() - start, + tags=tags, + ) + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'success'} + ) + return await self._build_result(dv, addr, names, tags) + + async def _build_result(self, dv, addr: str, names: list[dict] | None, tags: dict) -> dict: + value, status_name, source_ts, server_ts = self._extract(dv) + await self._record_read_lag(source_ts, tags) + source_str = self._format_ts(source_ts) + persisted = bool(names) + if names: + await self.sink.publish_reading( + tag=addr, value=value, timestamp=source_str, names=names + ) + return { + 'value': value, + 'opc_status_code': status_name, + 'source_timestamp': source_str, + 'server_timestamp': self._format_ts(server_ts), + 'persisted': persisted, + } + + @staticmethod + def _extract(dv): + status_name = dv.StatusCode.name if dv.StatusCode is not None else None + return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp + + @staticmethod + def _format_ts(ts) -> str | None: + if ts is None: + return None + return ts.replace(tzinfo=OPC_TIMEZONE).strftime(DATETIME_FORMAT_WITH_TZ) + + async def _record_read_lag(self, source_ts, tags: dict) -> None: + if source_ts is None: + return + lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp() + await self.emit_metric( + metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags + ) + + async def scheduled_tick(self, slot: dict) -> None: + """Sweep every server/tag in a parsed slot, reading tags due for an active read.""" + for server in slot.values(): + server_id = server['server_id'] + server_cfg = {k: v for k, v in server.items() if k != 'tags'} + for addr, tag in server['tags'].items(): + await self._tick_tag(server_id, addr, tag, server_cfg) + + async def _tick_tag(self, server_id: str, addr: str, tag: dict, server_cfg: dict) -> None: + active_read = tag.get('active_read') or {} + if not active_read.get('enabled'): + return + if not self._should_read(server_id, addr, active_read.get('cycle_multiple', 1)): + return + try: + await self.read_now(server_id, addr, names=tag['names'], server_cfg=server_cfg) + except Exception as e: + self.logger.error(f'Scheduled active read failed for {server_id}:{addr}: {e}') + + def _should_read(self, server_id: str, addr: str, cycle_multiple: int) -> bool: + key = (server_id, addr) + count = self._ticks.get(key, 0) + 1 + self._ticks[key] = count + return count % max(cycle_multiple, 1) == 0 diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py new file mode 100644 index 0000000..675cb74 --- /dev/null +++ b/ingestor/engines/write.py @@ -0,0 +1,195 @@ +"""OPC write execution (F2 capstone) from a NATS C2 command dict. + +Fixes 1879 correctness gaps: safe value coercion (never raises on bad +input), StatusCode/SourceTimestamp fidelity on the DataValue actually sent +to the server, a rate-limit floor between writes, and OPC error typing. +""" + +import asyncio +from dataclasses import dataclass + +from asyncua import ua +from asyncua.ua.uaerrors import UaStatusCodeError +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring + +import ingestor.metrics as metrics +from ingestor.connection.clock import MonotonicClock +from ingestor.connection.pool import OpcConnectionPool +from ingestor.converters import ConversionError, to_typed_value +from ingestor.io.data_sink import DataSink + +_VARIANT_TYPES = { + 'float': ua.VariantType.Float, + 'double': ua.VariantType.Double, + 'real': ua.VariantType.Double, + 'int': ua.VariantType.Int32, + 'integer': ua.VariantType.Int32, + 'bool': ua.VariantType.Boolean, + 'boolean': ua.VariantType.Boolean, + 'str': ua.VariantType.String, + 'string': ua.VariantType.String, +} + +# ponytail: same reconnect-worthy Bad*-status subset as laborious's +# opc_repository.py (different repo, not importable here); keep in sync by +# hand or promote to a shared lib if it drifts. +_SESSION_BAD_NAMES = frozenset( + { + 'BadSessionIdInvalid', + 'BadSessionClosed', + 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', + 'BadSecureChannelClosed', + 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', + 'BadServerNotConnected', + 'BadConnectionClosed', + 'BadConnectionRejected', + 'BadNotConnected', + 'BadTimeout', + 'BadRequestTimeout', + 'BadTooManySessions', + } +) + + +@dataclass +class WriteResult: + ok: bool + status: str # 'ok' | 'error' | 'session_bad' + opc_status_code: str + message: str + + +class WriteEngine(SientiaMonitoring): + """Executes one OPC write from a NATS C2 command.""" + + def __init__( + self, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + write_timestamp: bool = True, + write_quality: bool = True, + min_pause_s: float = 0.0, + ): + self.pool = pool + self.sink = sink + self.write_timestamp = write_timestamp + self.write_quality = write_quality + self.min_pause_s = min_pause_s + self.clock = MonotonicClock() + self._last_write_at: float | None = None + + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + async def write(self, cmd: dict) -> WriteResult: + await self._rate_limit() + + missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd] + if missing: + return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error') + + try: + typed = to_typed_value(cmd['value'], cmd['data_type']) + except ConversionError as e: + return await self._emit_error(cmd, 'conversion', str(e), 'error') + + dv = self._build_datavalue(cmd, typed) + + start = self.clock.elapsed() + try: + client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd)) + node = client.get_node(cmd['addr']) + await node.write_value(dv) + except Exception as e: + status = 'session_bad' if self._is_session_bad(e) else 'error' + return await self._emit_error(cmd, type(e).__name__, str(e), status) + + return await self._emit_success(cmd, start) + + async def _rate_limit(self) -> None: + now = self.clock.elapsed() + if self.min_pause_s > 0 and self._last_write_at is not None: + remaining = self.min_pause_s - (now - self._last_write_at) + if remaining > 0: + await asyncio.sleep(remaining) + now = self.clock.elapsed() + self._last_write_at = now + + def _server_cfg(self, cmd: dict) -> dict: + # ponytail: minimal cfg built straight from the C2 message. Resolving + # the full connector config (url/security) from the connectors DB by + # server_id is F4 wiring, not this task — pool is mocked in tests. + return { + 'url': cmd.get('url'), + 'use_security': cmd.get('use_security', False), + 'server_uri': cmd.get('server_uri'), + } + + def _variant_type(self, data_type: str) -> ua.VariantType: + # direct index: unknown dtypes already became ConversionError in to_typed_value + return _VARIANT_TYPES[data_type.strip().lower()] + + def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue: + variant = ua.Variant(typed_value, self._variant_type(cmd['data_type'])) + write_quality = cmd.get('write_quality', self.write_quality) + write_timestamp = cmd.get('write_timestamp', self.write_timestamp) + # NOTE: DataValue.StatusCode default_factory populates a Good status + # even when omitted, so "off" must pass StatusCode=None explicitly + # or the flag silently no-ops. + status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None + source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None + return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts) + + def _is_session_bad(self, exc: Exception) -> bool: + return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES + + async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str): + tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')} + await self.emit_metric( + metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type} + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'}, + ) + result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message) + await self.sink.update_write_status(cmd, result) + return result + + async def _emit_success(self, cmd: dict, start: float): + tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')} + await self.emit_metric( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'}, + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_LATENCY, + method='observe', + value=self.clock.elapsed() - start, + tags=tags, + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL, + method='set', + value=1, + tags={ + 'pod_id': self.pod_id, + 'model_id': cmd.get('model_id', 'unknown'), + 'status': 'ok', + }, + ) + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + await self.sink.update_write_status(cmd, result) + return result diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 52b526c..633e1df 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -1,16 +1,20 @@ +import re + +from sqlalchemy import text + import ingestor.metrics as metrics +# fullmatch (no ^/$ anchors) so a trailing newline can't slip through — `$` would +# match before a trailing '\n', which is not what we want for a SQL identifier. +_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII) + class DataSink: - """Gateway egress wrapper: fans a reading out to every configured name/collection. + """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: + def __init__(self, data_manager, postgres) -> 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 @@ -37,3 +41,34 @@ class DataSink: 'model_id': name['model_id'], }, ) + + async def update_write_status(self, cmd: dict, result) -> None: # NOSONAR + """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id. + + schema/table_name are per-model SQL identifiers, not values — they can't be + bound params, so they're checked against an allowlist before interpolation. + status/timestamp/model_id ARE bound params, never f-stringed. + """ + schema = self._validate_identifier(cmd['schema']) + table = self._validate_identifier(cmd['table_name']) + # schema/table are allowlist-validated identifiers above; status/timestamp/model_id + # are bound params below, never interpolated — not a real injection vector. + stmt = text( + f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 # nosec B608 + 'WHERE timestamp = :timestamp AND model_id = :model_id' + ) + params = { + 'status': result.status, + 'timestamp': cmd['timestamp'], + 'model_id': cmd['model_id'], + } + # ponytail: sync engine call inside async def — same pattern sientia_do's own + # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here. + with self.postgres.engine.begin() as connection: + connection.execute(stmt, params) + + @staticmethod + def _validate_identifier(name: str) -> str: + if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name): + raise ValueError(f'invalid SQL identifier: {name!r}') + return name diff --git a/ingestor/proto/__init__.py b/ingestor/proto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/proto/opc_gateway.proto b/ingestor/proto/opc_gateway.proto new file mode 100644 index 0000000..6def8d3 --- /dev/null +++ b/ingestor/proto/opc_gateway.proto @@ -0,0 +1,18 @@ +syntax = "proto3"; +package opcgateway; + +service OpcGateway { + rpc ReadTag(ReadTagRequest) returns (ReadTagResponse); +} +message ReadTagRequest { + string server_id = 1; + string addr = 2; + string data_type = 3; +} +message ReadTagResponse { + string value = 1; + string opc_status_code = 2; + string source_timestamp = 3; + string server_timestamp = 4; + bool persisted = 5; +} diff --git a/ingestor/proto/opc_gateway_pb2.py b/ingestor/proto/opc_gateway_pb2.py new file mode 100644 index 0000000..0a5f08c --- /dev/null +++ b/ingestor/proto/opc_gateway_pb2.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: opc_gateway.proto +# Protobuf Python Version: 7.35.0 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 7, + 35, + 0, + '', + 'opc_gateway.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x11opc_gateway.proto\x12\nopcgateway\"D\n\x0eReadTagRequest\x12\x11\n\tserver_id\x18\x01 \x01(\t\x12\x0c\n\x04\x61\x64\x64r\x18\x02 \x01(\t\x12\x11\n\tdata_type\x18\x03 \x01(\t\"\x80\x01\n\x0fReadTagResponse\x12\r\n\x05value\x18\x01 \x01(\t\x12\x17\n\x0fopc_status_code\x18\x02 \x01(\t\x12\x18\n\x10source_timestamp\x18\x03 \x01(\t\x12\x18\n\x10server_timestamp\x18\x04 \x01(\t\x12\x11\n\tpersisted\x18\x05 \x01(\x08\x32P\n\nOpcGateway\x12\x42\n\x07ReadTag\x12\x1a.opcgateway.ReadTagRequest\x1a\x1b.opcgateway.ReadTagResponseb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'opc_gateway_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_READTAGREQUEST']._serialized_start=33 + _globals['_READTAGREQUEST']._serialized_end=101 + _globals['_READTAGRESPONSE']._serialized_start=104 + _globals['_READTAGRESPONSE']._serialized_end=232 + _globals['_OPCGATEWAY']._serialized_start=234 + _globals['_OPCGATEWAY']._serialized_end=314 +# @@protoc_insertion_point(module_scope) diff --git a/ingestor/proto/opc_gateway_pb2_grpc.py b/ingestor/proto/opc_gateway_pb2_grpc.py new file mode 100644 index 0000000..1f8ac2d --- /dev/null +++ b/ingestor/proto/opc_gateway_pb2_grpc.py @@ -0,0 +1,97 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc +import warnings + +from . import opc_gateway_pb2 as opc__gateway__pb2 + +GRPC_GENERATED_VERSION = '1.82.1' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False + +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f'The grpc package installed is at version {GRPC_VERSION},' + + ' but the generated code in opc_gateway_pb2_grpc.py depends on' + + f' grpcio>={GRPC_GENERATED_VERSION}.' + + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.' + ) + + +class OpcGatewayStub: + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.ReadTag = channel.unary_unary( + '/opcgateway.OpcGateway/ReadTag', + request_serializer=opc__gateway__pb2.ReadTagRequest.SerializeToString, + response_deserializer=opc__gateway__pb2.ReadTagResponse.FromString, + _registered_method=True) + + +class OpcGatewayServicer: + """Missing associated documentation comment in .proto file.""" + + def ReadTag(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_OpcGatewayServicer_to_server(servicer, server): + rpc_method_handlers = { + 'ReadTag': grpc.unary_unary_rpc_method_handler( + servicer.ReadTag, + request_deserializer=opc__gateway__pb2.ReadTagRequest.FromString, + response_serializer=opc__gateway__pb2.ReadTagResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'opcgateway.OpcGateway', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('opcgateway.OpcGateway', rpc_method_handlers) + + + # This class is part of an EXPERIMENTAL API. +class OpcGateway: + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def ReadTag(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/opcgateway.OpcGateway/ReadTag', + opc__gateway__pb2.ReadTagRequest.SerializeToString, + opc__gateway__pb2.ReadTagResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/pyproject.toml b/pyproject.toml index 8a7dbab..fd3e168 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ exclude = [ "*.pyc", ".pytest_cache", "htmlcov", + "ingestor/proto/*_pb2*.py", ] [tool.ruff.lint] @@ -80,6 +81,9 @@ warn_unused_ignores = false warn_no_return = true strict_equality = true ignore_missing_imports = true +# Generated protoc/grpcio-tools output - not a checking target, dynamic attrs +# mypy can't see (message classes built at runtime from DESCRIPTOR). +exclude = ["^ingestor/proto/"] # Ignore missing imports for external packages [[tool.mypy.overrides]] @@ -106,6 +110,14 @@ ignore_missing_imports = true module = "pandas.*" ignore_missing_imports = true +# Generated protoc/grpcio-tools output - not hand-maintained, dynamic attrs +# mypy can't see (message classes built at runtime from DESCRIPTOR). Skip +# following it entirely so importers see Any instead of false attr-defined errors. +[[tool.mypy.overrides]] +module = "ingestor.proto.*" +ignore_errors = true +follow_imports = "skip" + [tool.pytest.ini_options] testpaths = ["tests"] python_files = ["test_*.py"] diff --git a/sonar-project.properties b/sonar-project.properties index fc43878..5c832c5 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -8,3 +8,7 @@ sonar.python.coverage.reportPaths=coverage.xml sonar.python.xunit.reportPath=pytest.xml sonar.python.version=3.11 sonar.coverage.exclusions=ingestor/app.py +# Generated protoc/grpcio-tools output - not hand-maintained; renaming its +# ReadTag/etc. methods to satisfy naming-convention rules would break gRPC +# wire compatibility (grpcio-tools dictates these names, not style choice). +sonar.exclusions=ingestor/proto/** diff --git a/tests/unit/adapters/__init__.py b/tests/unit/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/adapters/test_grpc_server.py b/tests/unit/adapters/test_grpc_server.py new file mode 100644 index 0000000..a5115f6 --- /dev/null +++ b/tests/unit/adapters/test_grpc_server.py @@ -0,0 +1,97 @@ +from unittest.mock import AsyncMock, MagicMock + +import grpc +import pytest + +from ingestor.adapters.grpc_server import GrpcServer, ReadTagServicer +from ingestor.proto.opc_gateway_pb2 import ReadTagRequest + + +def make_request(server_id='srv1', addr='ns=2;i=5'): + return ReadTagRequest(server_id=server_id, addr=addr, data_type='float') + + +@pytest.mark.asyncio +async def test_read_tag_maps_full_result_onto_response(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1.5, + 'opc_status_code': 'Good', + 'source_timestamp': '2026-07-15T10:00:00-03:00', + 'server_timestamp': '2026-07-15T10:00:01-03:00', + 'persisted': False, + } + servicer = ReadTagServicer(active_read, MagicMock()) + + response = await servicer.ReadTag(make_request(), AsyncMock()) + + assert response.value == '1.5' + assert response.opc_status_code == 'Good' + assert response.source_timestamp == '2026-07-15T10:00:00-03:00' + assert response.server_timestamp == '2026-07-15T10:00:01-03:00' + assert response.persisted is False + + +@pytest.mark.asyncio +async def test_read_tag_none_timestamps_become_empty_string(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1.5, + 'opc_status_code': None, + 'source_timestamp': None, + 'server_timestamp': None, + 'persisted': False, + } + servicer = ReadTagServicer(active_read, MagicMock()) + + response = await servicer.ReadTag(make_request(), AsyncMock()) + + assert response.opc_status_code == '' + assert response.source_timestamp == '' + assert response.server_timestamp == '' + + +@pytest.mark.asyncio +async def test_read_tag_aborts_with_internal_on_engine_failure(): + active_read = AsyncMock() + active_read.read_now.side_effect = RuntimeError('boom') + logger = MagicMock() + servicer = ReadTagServicer(active_read, logger) + context = AsyncMock() + context.abort.side_effect = grpc.RpcError('aborted') + + request = make_request() + with pytest.raises(grpc.RpcError): + await servicer.ReadTag(request, context) + + context.abort.assert_awaited_once() + args = context.abort.call_args.args + assert args[0] == grpc.StatusCode.INTERNAL + logger.error.assert_called_once() + + +@pytest.mark.asyncio +async def test_read_tag_uses_slot_lookup_names_when_provided(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1, + 'opc_status_code': 'Good', + 'source_timestamp': None, + 'server_timestamp': None, + 'persisted': True, + } + names = [{'collection': 'c1', 'name': 't1'}] + slot_lookup = MagicMock(return_value=names) + servicer = ReadTagServicer(active_read, MagicMock(), slot_lookup=slot_lookup) + + await servicer.ReadTag(make_request(server_id='srv1', addr='ns=2;i=5'), AsyncMock()) + + slot_lookup.assert_called_once_with('srv1', 'ns=2;i=5') + active_read.read_now.assert_awaited_once_with('srv1', 'ns=2;i=5', names=names) + + +@pytest.mark.asyncio +async def test_grpc_server_stop_before_start_does_not_raise(): + server = GrpcServer(active_read=AsyncMock(), logger=MagicMock()) + + await server.stop() # no raise diff --git a/tests/unit/adapters/test_nats_consumer.py b/tests/unit/adapters/test_nats_consumer.py new file mode 100644 index 0000000..7410dab --- /dev/null +++ b/tests/unit/adapters/test_nats_consumer.py @@ -0,0 +1,76 @@ +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from ingestor.adapters.nats_consumer import NatsWriteConsumer + + +def make_consumer(**kwargs): + return NatsWriteConsumer( + nats_url='nats://localhost:4222', + write_engine=AsyncMock(), + logger=MagicMock(), + **kwargs, + ) + + +@pytest.mark.asyncio +async def test_handle_valid_json_dispatches_to_write_engine(): + consumer = make_consumer() + payload = {'server_id': 'srv1', 'addr': 'ns=2;i=5', 'value': 1.0, 'data_type': 'float'} + msg = MagicMock(data=json.dumps(payload).encode()) + + await consumer._handle(msg) + + consumer.write_engine.write.assert_awaited_once_with(payload) + + +@pytest.mark.asyncio +async def test_handle_invalid_json_drops_message_and_logs_error(): + consumer = make_consumer() + msg = MagicMock(data=b'not-json') + + await consumer._handle(msg) + + consumer.write_engine.write.assert_not_awaited() + consumer.logger.error.assert_called_once() + + +@pytest.mark.asyncio +async def test_start_subscribes_with_subject_and_queue(): + consumer = make_consumer(subject='opc.write', queue='gateway-writers') + mock_nc = AsyncMock() + mock_nc.subscribe = AsyncMock() + + with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: + mock_nats.connect = AsyncMock(return_value=mock_nc) + await consumer.start() + + mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222') + mock_nc.subscribe.assert_awaited_once_with( + 'opc.write', queue='gateway-writers', cb=consumer._handle + ) + + +@pytest.mark.asyncio +async def test_stop_before_start_does_not_raise(): + consumer = make_consumer() + + await consumer.stop() # no raise + + +@pytest.mark.asyncio +async def test_stop_after_start_drains_and_unsubscribes(): + consumer = make_consumer() + mock_nc = AsyncMock() + mock_sub = AsyncMock() + mock_nc.subscribe = AsyncMock(return_value=mock_sub) + + with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: + mock_nats.connect = AsyncMock(return_value=mock_nc) + await consumer.start() + await consumer.stop() + + mock_sub.unsubscribe.assert_awaited_once() + mock_nc.drain.assert_awaited_once() diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py new file mode 100644 index 0000000..c6bb5b3 --- /dev/null +++ b/tests/unit/engines/test_active_read.py @@ -0,0 +1,179 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +from asyncua import ua + +import ingestor.metrics as metrics +from ingestor.engines.active_read import ActiveReadEngine + + +def make_engine(): + eng = ActiveReadEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + ) + eng.emit_metric = AsyncMock() + return eng + + +def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None): + source_ts = source_ts or datetime(2026, 7, 13, 12, 0, 0) + server_ts = server_ts or source_ts + return ua.DataValue( + ua.Variant(value), + StatusCode=ua.StatusCode(getattr(ua.StatusCodes, status)), + SourceTimestamp=source_ts, + ServerTimestamp=server_ts, + ) + + +def make_client(dv=None): + node = MagicMock() + node.read_data_value = AsyncMock(return_value=dv or make_dv()) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + return client, node + + +NAMES = [{'tag_name': 'a', 'model_id': 'm1', 'collection': 'coll'}] + + +@pytest.mark.asyncio +async def test_read_now_persists_and_returns_fields_when_names_given(): + eng = make_engine() + client, node = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5', names=NAMES) + + client.get_node.assert_called_once_with('ns=2;i=5') + node.read_data_value.assert_awaited_once() + eng.sink.publish_reading.assert_awaited_once() + _, kwargs = eng.sink.publish_reading.call_args + assert kwargs['tag'] == 'ns=2;i=5' + assert kwargs['value'] == 42.0 + assert kwargs['names'] == NAMES + assert result['value'] == 42.0 + assert result['opc_status_code'] == 'Good' + assert result['persisted'] is True + assert result['source_timestamp'].startswith('2026-07-13 12:00:00') + assert result['server_timestamp'].startswith('2026-07-13 12:00:00') + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'success'}, + ) + latency_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_ACTIVE_READ_LATENCY + ) + assert latency_call.kwargs['method'] == 'observe' + assert latency_call.kwargs['value'] >= 0 + lag_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_READ_LAG + ) + assert lag_call.kwargs['method'] == 'set' + + +@pytest.mark.asyncio +async def test_read_now_without_names_does_not_persist(): + eng = make_engine() + client, _ = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5') + + eng.sink.publish_reading.assert_not_awaited() + assert result['persisted'] is False + + +@pytest.mark.asyncio +async def test_read_now_error_emits_metric_and_reraises(): + eng = make_engine() + node = MagicMock() + node.read_data_value = AsyncMock(side_effect=RuntimeError('boom')) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + eng.pool.get.return_value = client + + with pytest.raises(RuntimeError, match='boom'): + await eng.read_now('srv1', 'ns=2;i=5') + + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'error'}, + ) + eng.sink.publish_reading.assert_not_awaited() + + +def make_slot(cycle_multiple=None, enabled=True): + active_read = {'enabled': enabled} + if cycle_multiple is not None: + active_read['cycle_multiple'] = cycle_multiple + return { + 'srv_name': { + 'server_id': 'srv1', + 'url': 'opc.tcp://x', + 'tags': { + 'ns=2;i=5': {'frequency': 1000, 'names': NAMES, 'active_read': active_read}, + }, + } + } + + +@pytest.mark.asyncio +async def test_scheduled_tick_respects_cycle_multiple(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(cycle_multiple=3) + + await eng.scheduled_tick(slot) + await eng.scheduled_tick(slot) + eng.read_now.assert_not_awaited() + await eng.scheduled_tick(slot) + eng.read_now.assert_awaited_once_with( + 'srv1', 'ns=2;i=5', names=NAMES, server_cfg={'server_id': 'srv1', 'url': 'opc.tcp://x'} + ) + + +@pytest.mark.asyncio +async def test_scheduled_tick_skips_disabled_tags(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(enabled=False) + + await eng.scheduled_tick(slot) + + eng.read_now.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_scheduled_tick_continues_past_tag_error(): + eng = make_engine() + bad_slot = make_slot(cycle_multiple=1) + good_names = [{'tag_name': 'b', 'model_id': 'm2', 'collection': 'coll'}] + bad_slot['srv_name']['tags']['ns=2;i=6'] = { + 'frequency': 1000, + 'names': good_names, + 'active_read': {'enabled': True, 'cycle_multiple': 1}, + } + + calls = [] + + async def fake_read_now(server_id, addr, names=None, server_cfg=None): + calls.append(addr) + if addr == 'ns=2;i=5': + raise RuntimeError('boom') + + eng.read_now = AsyncMock(side_effect=fake_read_now) + + await eng.scheduled_tick(bad_slot) + + assert set(calls) == {'ns=2;i=5', 'ns=2;i=6'} + eng.logger.error.assert_called_once() diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py new file mode 100644 index 0000000..ad54c30 --- /dev/null +++ b/tests/unit/engines/test_write.py @@ -0,0 +1,259 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from asyncua import ua + +import ingestor.metrics as metrics +from ingestor.engines.write import WriteEngine + + +def make_engine(**kwargs): + eng = WriteEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + **kwargs, + ) + eng.emit_metric = AsyncMock() + return eng + + +def make_client(node=None): + node = node or make_node() + client = MagicMock() + client.get_node = MagicMock(return_value=node) + return client + + +def make_node(): + node = MagicMock() + node.write_value = AsyncMock() + return node + + +def base_cmd(**overrides): + cmd = { + 'server_id': 'srv1', + 'addr': 'ns=2;i=5', + 'value': 42.0, + 'data_type': 'float', + 'tag_type': 'prediction', + 'model_id': 'm1', + 'model_name': 'model-1', + 'timestamp': '2026-07-13T12:00:00Z', + 'schema': 'sientia_data', + 'table_name': 'predictions_m1', + } + cmd.update(overrides) + return cmd + + +@pytest.mark.asyncio +async def test_write_success_writes_datavalue_and_returns_ok(): + eng = make_engine() + node = make_node() + client = make_client(node) + eng.pool.get.return_value = client + + result = await eng.write(base_cmd()) + + client.get_node.assert_called_once_with('ns=2;i=5') + node.write_value.assert_awaited_once() + dv = node.write_value.await_args.args[0] + assert isinstance(dv, ua.DataValue) + assert dv.Value.Value == 42.0 + assert dv.Value.VariantType == ua.VariantType.Float + assert result.ok is True + assert result.status == 'ok' + assert result.opc_status_code == 'Good' + eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result) + + +@pytest.mark.asyncio +async def test_write_success_emits_metrics(): + eng = make_engine() + eng.pool.get.return_value = make_client() + + await eng.write(base_cmd()) + + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'success'}, + ) + status_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_BY_MODEL + ) + assert status_call.kwargs['value'] == 1 + assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'} + latency_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_LATENCY + ) + assert latency_call.kwargs['method'] == 'observe' + assert latency_call.kwargs['value'] >= 0 + + +@pytest.mark.asyncio +async def test_write_defaults_carry_statuscode_and_timestamp(): + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.StatusCode is not None + assert dv.StatusCode.is_good() + assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z') + + +@pytest.mark.asyncio +async def test_write_engine_flags_off_omit_statuscode_and_timestamp(): + eng = make_engine(write_timestamp=False, write_quality=False) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.StatusCode is None + assert dv.SourceTimestamp is None + + +@pytest.mark.asyncio +async def test_per_message_flag_override_disables_timestamp(): + # Engine default is True, but the command explicitly turns it off. + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd(write_timestamp=False)) + + dv = node.write_value.await_args.args[0] + assert dv.SourceTimestamp is None + assert dv.StatusCode is not None # write_quality untouched, stays on + + +@pytest.mark.asyncio +async def test_write_bad_type_returns_error_without_raising(): + eng = make_engine() + + result = await eng.write(base_cmd(value='abc', data_type='int')) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ERROR_BY_TYPE, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'conversion'}, + ) + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'error'}, + ) + + +@pytest.mark.asyncio +async def test_write_none_data_type_returns_error_without_raising(): + eng = make_engine() + + result = await eng.write(base_cmd(data_type=None)) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_missing_value_key_returns_error_without_raising(): + eng = make_engine() + cmd = base_cmd() + del cmd['value'] + + result = await eng.write(cmd) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_missing_data_type_key_returns_error_without_raising(): + eng = make_engine() + cmd = base_cmd() + del cmd['data_type'] + + result = await eng.write(cmd) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_value_raises_maps_to_error_result_not_raised(): + eng = make_engine() + node = make_node() + node.write_value.side_effect = RuntimeError('boom') + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.ok is False + assert result.status == 'error' + assert result.opc_status_code == 'RuntimeError' + eng.sink.update_write_status.assert_awaited_once() + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ERROR_BY_TYPE, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'RuntimeError'}, + ) + + +@pytest.mark.asyncio +async def test_write_session_bad_status_code_error_maps_to_session_bad(): + from asyncua.ua.uaerrors import BadSessionClosed + + eng = make_engine() + node = make_node() + node.write_value.side_effect = BadSessionClosed('closed') + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.ok is False + assert result.status == 'session_bad' + + +@pytest.mark.asyncio +async def test_rate_limit_sleeps_between_back_to_back_writes(): + eng = make_engine(min_pause_s=0.05) + eng.pool.get.return_value = make_client() + + with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock: + await eng.write(base_cmd()) + sleep_mock.assert_not_awaited() + await eng.write(base_cmd()) + sleep_mock.assert_awaited_once() + remaining = sleep_mock.await_args.args[0] + assert 0 < remaining <= 0.05 + + +@pytest.mark.asyncio +async def test_no_rate_limit_when_min_pause_is_zero(): + eng = make_engine(min_pause_s=0.0) + eng.pool.get.return_value = make_client() + + with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock: + await eng.write(base_cmd()) + await eng.write(base_cmd()) + sleep_mock.assert_not_awaited() diff --git a/tests/unit/io/test_data_sink_status.py b/tests/unit/io/test_data_sink_status.py new file mode 100644 index 0000000..a626036 --- /dev/null +++ b/tests/unit/io/test_data_sink_status.py @@ -0,0 +1,89 @@ +from unittest.mock import MagicMock + +import pytest + +from ingestor.engines.write import WriteResult +from ingestor.io.data_sink import DataSink + + +def make_sink(): + postgres = MagicMock() + connection = MagicMock() + postgres.engine.begin.return_value.__enter__.return_value = connection + sink = DataSink(data_manager=MagicMock(), postgres=postgres) + return sink, connection + + +@pytest.mark.asyncio +async def test_update_write_status_targets_model_row(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + + await sink.update_write_status(cmd, result) + + assert connection.execute.call_count == 1 + stmt, params = connection.execute.call_args[0] + sql = str(stmt) + assert 'update' in sql.lower() and 'write_status' in sql.lower() + assert 'sientia_data.pred_m1' in sql + # values must be bound params, never interpolated into the SQL text itself + assert params == {'status': 'ok', 'timestamp': '2026-07-13T00:00:00Z', 'model_id': '1'} + assert 'ok' not in sql + assert '2026-07-13T00:00:00Z' not in sql + + +@pytest.mark.asyncio +async def test_update_write_status_rejects_bad_schema(): + sink, connection = make_sink() + cmd = { + 'schema': 'a; DROP TABLE x', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_update_write_status_rejects_trailing_newline_identifier(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data\n', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_update_write_status_rejects_bad_table_name(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data', + 'table_name': 'pred_m1; DROP TABLE x', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() diff --git a/tests/unit/proto/__init__.py b/tests/unit/proto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/proto/test_proto.py b/tests/unit/proto/test_proto.py new file mode 100644 index 0000000..de8a319 --- /dev/null +++ b/tests/unit/proto/test_proto.py @@ -0,0 +1,36 @@ +from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc + + +def test_read_tag_request_roundtrip(): + req = opc_gateway_pb2.ReadTagRequest(server_id='s1', addr='ns=2;s=Tag1', data_type='Float') + + data = req.SerializeToString() + parsed = opc_gateway_pb2.ReadTagRequest() + parsed.ParseFromString(data) + + assert parsed.server_id == 's1' + assert parsed.addr == 'ns=2;s=Tag1' + assert parsed.data_type == 'Float' + + +def test_read_tag_response_roundtrip(): + resp = opc_gateway_pb2.ReadTagResponse( + value='42.0', + opc_status_code='Good', + source_timestamp='2026-07-14T00:00:00Z', + server_timestamp='2026-07-14T00:00:01Z', + persisted=True, + ) + + data = resp.SerializeToString() + parsed = opc_gateway_pb2.ReadTagResponse() + parsed.ParseFromString(data) + + assert parsed.value == '42.0' + assert parsed.opc_status_code == 'Good' + assert parsed.persisted is True + + +def test_grpc_stub_module_loads(): + assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayStub') + assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayServicer') diff --git a/tests/unit/test_converters.py b/tests/unit/test_converters.py index 4df64e7..c91be57 100644 --- a/tests/unit/test_converters.py +++ b/tests/unit/test_converters.py @@ -38,3 +38,10 @@ def test_float_and_int(): def test_invalid_raises_conversion_error(raw, dtype): with pytest.raises(ConversionError): to_typed_value(raw, dtype) + + +@pytest.mark.parametrize('dtype', [None, 123]) +def test_non_string_data_type_raises_conversion_error(dtype): + # contract: never a bare AttributeError/TypeError, always ConversionError + with pytest.raises(ConversionError): + to_typed_value('5', dtype)