TEST ONLY: sqlalchemy<2.1 (No module named psycopg with sientia_do 1.12.2) - dev test, do not merge

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -23,7 +23,6 @@ GRPC_ENABLED="false"
 GRPC_PORT="50051"
 CERT_BUCKET="sientia-dataops"
 CERT_CACHE_DIR="/tmp/opc-certs"
-# Defaults of the write options; the server document and the command override them.
 WRITE_TIMESTAMP="true"
 WRITE_QUALITY="true"
 WRITE_MIN_PAUSE_S="0.0"

.github/workflows/deploy.yml

@@ -1,15 +1,10 @@
 name: Deploy DataOps Worker
 
 on:
+  pull_request:
+    types: [closed]
+    branches: [main, 'release/**', 'feature/**']
   workflow_dispatch:
-    inputs:
-      chart_ref:
-        description: >
-          sientia-helm-chart branch to bump GITHUB_BRANCH on AND deploy — the literal
-          deploy target. Validated via git ls-remote; a branch that does not exist in
-          sientia-helm-chart is a hard error, not a silent fallback to main.
-        required: true
-        type: string
 
 permissions:
   contents: write
@@ -19,10 +14,10 @@ permissions:
 jobs:
   deploy:
     name: Deploy
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
     uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
     with:
-      branch_name: ${{ github.ref_name }}
-      chart_ref: ${{ inputs.chart_ref }}
+      branch_name: ${{ github.event.pull_request.head.ref }}
       project_name: 'ingestor'
       chart_name: 'opc-ingestor'
       env_list_path: '[".env"]'

.github/workflows/quality-gate.yml

@@ -5,6 +5,7 @@ on:
     branches:
       - main
       - 'release/**'
+      - 'feature/**'
   pull_request:
     branches:
       - main

.github/workflows/require-chart-ref-label.yml

@@ -0,0 +1,16 @@
+name: Require chart-ref label
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
+
+permissions:
+  pull-requests: write
+
+jobs:
+  require-chart-ref-label:
+    uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

.github/workflows/version-bump.yml

@@ -1,17 +0,0 @@
-name: Version Bump
-
-permissions:
-  contents: write
-
-on:
-  pull_request:
-    types: [closed]
-    branches: [main, 'release/**', 'feature/**']
-
-jobs:
-  bump:
-    if: github.event.pull_request.merged == true
-    uses: Aignosi/github_workflow_templates/.github/workflows/reusable-version-bump.yml@main
-    with:
-      update_version_in: '["pyproject"]'
-    secrets: inherit

README.md

@@ -79,21 +79,9 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975):
 {"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float",
  "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data",
  "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00",
- "quality": "Good"}
+ "write_timestamp": false, "write_quality": false}
 ```
 
-- **Timestamp and quality** (SIENTIAPDE-2202): besides the value, the write carries the
-  prediction `timestamp` as `SourceTimestamp` and the `quality` (the prediction status:
-  `Good`, `Bad` or `Uncertain`; `Good` when absent) as `StatusCode`. Each one is on or
-  off by, in order: `write_timestamp`/`write_quality` in the command, the same boolean
-  fields on the server document (`connectors` or `opc-servers`, cached like the
-  connection), then the `WRITE_TIMESTAMP`/`WRITE_QUALITY` env vars (default `true`).
-  When a server refuses them (`BadWriteNotSupported`, e.g. Kepware), the value is
-  written again alone, the server is remembered until the gateway restarts (its next
-  writes go with the value alone), a warning is logged and
-  `opc_write_value_only_fallback_total` is incremented. Set the two fields to `false` on
-  such a server's document to skip the refused attempt.
-
 - **Server connection**: the command carries only `server_id`. The connection is
   resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): `url`/
   `server_uri`/`use_security` in the command when present (override), else the

ingestor/connection/servers.py

@@ -4,7 +4,6 @@ The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`);
 connection (url, server URI, security) belongs to the gateway, which owns every OPC
 session. It is resolved with the same rule the orchestrator uses to build the read slots
 (SIENTIAPDE-1974): the tag database `connectors` first, the legacy `opc-servers` otherwise.
-The same document may also set the write options of the server (SIENTIAPDE-2202).
 """
 
 from typing import Any
@@ -16,7 +15,6 @@ from ingestor.connection.pool import OpcConnectionPool
 
 CONNECTORS = 'connectors'
 OPC_SERVERS = 'opc-servers'
-WRITE_OPTIONS = ('write_timestamp', 'write_quality')
 
 
 def _id_forms(server_id: str) -> list[Any]:
@@ -49,15 +47,8 @@ def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None:
     }
 
 
-def _write_options(document: dict[str, Any]) -> dict[str, bool]:
-    """The write options the server document sets. Only booleans count: a hand-edited
-    string such as "false" must not turn an option on."""
-    return {key: document[key] for key in WRITE_OPTIONS if isinstance(document.get(key), bool)}
-
-
 class ServerConfigResolver:
-    """Resolve and cache the connection settings (and write options) of an OPC server by
-    `server_id`.
+    """Resolve and cache the connection settings of an OPC server by `server_id`.
 
     Order: settings carried by the command itself (an explicit override; the C2
     command of the Laborious carries none), the live pooled connection (what the read
@@ -100,40 +91,17 @@ class ServerConfigResolver:
         pooled = self.pool.config_for(server_id)
         if pooled is not None:
             return pooled
-        server = await self._server(server_id)
-        return dict(server['connection']) if server is not None else None
-
-    async def write_options(self, server_id: str) -> dict[str, bool]:
-        """`write_timestamp`/`write_quality` set on the server document (SIENTIAPDE-2202),
-        only the ones it sets: the gateway defaults apply to the others.
-
-        Read from MongoDB even when the read path has the connection open (`resolve`
-        skips MongoDB then), with the same cache. Never raises: when MongoDB cannot be
-        read, the gateway defaults apply and the write goes on.
-        """
-        try:
-            server = await self._server(str(server_id))
-        except Exception as e:
-            self.logger.warning(
-                f'Write options of OPC server {server_id} not read, using the gateway defaults: {e}'
-            )
-            return {}
-        return dict(server['write_options']) if server is not None else {}
-
-    async def _server(self, server_id: str) -> dict[str, Any] | None:
         now = self.clock.elapsed()
         cached = self._cache.get(server_id)
         if cached is not None:
-            server, cached_at = cached
-            if now - cached_at < (self.ttl_s if server is not None else self.missing_ttl_s):
-                return server
-        server = await self._lookup(server_id)
-        self._cache[server_id] = (server, now)
-        return server
+            cfg, cached_at = cached
+            if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s):
+                return dict(cfg) if cfg is not None else None
+        cfg = await self._lookup(server_id)
+        self._cache[server_id] = (cfg, now)
+        return dict(cfg) if cfg is not None else None
 
     async def _lookup(self, server_id: str) -> dict[str, Any] | None:
-        """The connection settings and write options of the first document that knows
-        the server, or None."""
         ids = {'$in': _id_forms(server_id)}
         connector = await self.mongo.find_one(
             CONNECTORS,
@@ -142,12 +110,11 @@ class ServerConfigResolver:
         )
         cfg = _connector_cfg(connector) if connector else None
         if cfg is not None:
-            return {'connection': cfg, 'write_options': _write_options(connector)}
+            return cfg
         server = await self.mongo.find_one(OPC_SERVERS, {'id': ids}, self.metadata)
         cfg = _opc_server_cfg(server) if server else None
         if cfg is None:
             self.logger.warning(
                 f'OPC server {server_id} not found in {CONNECTORS} nor {OPC_SERVERS}'
             )
-            return None
-        return {'connection': cfg, 'write_options': _write_options(server)}
+        return cfg

ingestor/engines/write.py

@@ -9,7 +9,7 @@ import asyncio
 from dataclasses import dataclass
 
 from asyncua import ua
-from asyncua.ua.uaerrors import BadWriteNotSupported, UaStatusCodeError
+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
@@ -57,23 +57,6 @@ _SESSION_BAD_NAMES = frozenset(
 )
 
 
-# `quality` of the C2 command (the Laborious sends its prediction_status) -> OPC quality.
-_QUALITY_CODES = {
-    'good': ua.StatusCodes.Good,
-    'uncertain': ua.StatusCodes.Uncertain,
-    'bad': ua.StatusCodes.Bad,
-}
-
-
-class _InvalidFieldError(Exception):
-    """A command field needed by an option that is on (timestamp, quality) is bad."""
-
-    def __init__(self, error_type: str, message: str):
-        super().__init__(message)
-        self.error_type = error_type
-        self.message = message
-
-
 @dataclass
 class WriteResult:
     ok: bool
@@ -102,14 +85,11 @@ class WriteEngine(SientiaMonitoring):
         # Resolves the server connection by server_id (SIENTIAPDE-1975 GW1). Without it,
         # only the connection carried by the command itself is used.
         self.servers = servers
-        # Defaults (env); the server document and the command override them (SIENTIAPDE-2202).
         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
-        # Servers that refused timestamp/quality: written with the value alone.
-        self._value_only_servers: set[str] = set()
 
         SientiaMonitoring.__init__(
             self,
@@ -130,6 +110,12 @@ class WriteEngine(SientiaMonitoring):
         except ConversionError as e:
             return await self._emit_error(cmd, 'conversion', str(e), 'error')
 
+        try:
+            dv = self._build_datavalue(cmd, typed)
+        except (KeyError, TypeError, ValueError) as e:
+            # write_timestamp on with a missing or non-ISO timestamp.
+            return await self._emit_error(cmd, 'bad_timestamp', repr(e), 'error')
+
         try:
             server_cfg = await self._server_cfg(cmd)
         except Exception as e:
@@ -139,18 +125,11 @@ class WriteEngine(SientiaMonitoring):
                 cmd, 'server_not_found', f'OPC server {cmd["server_id"]} is not configured', 'error'
             )
 
-        server_id = str(cmd['server_id'])
-        write_timestamp, write_quality = await self._write_flags(server_id, cmd)
-        try:
-            dv = self._build_datavalue(cmd, typed, write_timestamp, write_quality)
-        except _InvalidFieldError as e:
-            return await self._emit_error(cmd, e.error_type, e.message, 'error')
-
         start = self.clock.elapsed()
         try:
             client = await self.pool.get(cmd['server_id'], server_cfg)
             node = client.get_node(cmd['addr'])
-            await self._write_node(node, dv, server_id)
+            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)
@@ -184,76 +163,17 @@ class WriteEngine(SientiaMonitoring):
         # direct index: unknown dtypes already became ConversionError in to_typed_value
         return _VARIANT_TYPES[data_type.strip().lower()]
 
-    async def _write_flags(self, server_id: str, cmd: dict) -> tuple[bool, bool]:
-        """`write_timestamp` and `write_quality` of this write (SIENTIAPDE-2202): the
-        command, then the server document, then the gateway env. A server that already
-        refused them (BadWriteNotSupported) gets the value alone."""
-        if server_id in self._value_only_servers:
-            return False, False
-        options = await self.servers.write_options(server_id) if self.servers is not None else {}
-
-        def flag(key: str, default: bool) -> bool:
-            value = cmd.get(key)
-            return bool(options.get(key, default) if value is None else value)
-
-        return flag('write_timestamp', self.write_timestamp), flag(
-            'write_quality', self.write_quality
-        )
-
-    def _build_datavalue(
-        self, cmd: dict, typed_value, write_timestamp: bool, write_quality: bool
-    ) -> ua.DataValue:
-        """The value, with the prediction timestamp and quality when they are on.
-
-        Raises:
-            _InvalidFieldError: When an option is on and its command field is missing or bad.
-        """
+    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 = self._status_code(cmd) if write_quality else None
-        try:
-            source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None
-        except (KeyError, TypeError, ValueError) as e:
-            raise _InvalidFieldError('bad_timestamp', repr(e)) from e
+        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)
 
-    @staticmethod
-    def _status_code(cmd: dict) -> ua.StatusCode:
-        """The OPC quality of the command's `quality` (the prediction status). Commands
-        without it (before SIENTIAPDE-2202) are written as Good."""
-        quality = cmd.get('quality') or 'good'
-        code = _QUALITY_CODES.get(str(quality).strip().lower())
-        if code is None:
-            raise _InvalidFieldError('bad_quality', f'unknown quality {quality!r}')
-        return ua.StatusCode(ua.UInt32(code))
-
-    async def _write_node(self, node, dv: ua.DataValue, server_id: str) -> None:
-        """Write `dv`. When the server refuses the timestamp/quality it carries
-        (BadWriteNotSupported, e.g. Kepware), write the value alone and remember the
-        server, so its next writes go without them."""
-        try:
-            await node.write_value(dv)
-        except BadWriteNotSupported:
-            if dv.StatusCode is None and dv.SourceTimestamp is None:
-                raise
-            await node.write_value(ua.DataValue(dv.Value, StatusCode=None))
-            await self._remember_value_only(server_id)
-
-    async def _remember_value_only(self, server_id: str) -> None:
-        self._value_only_servers.add(server_id)
-        self.logger.warning(
-            f'OPC server {server_id} refused the timestamp/quality of a write '
-            f'(BadWriteNotSupported): its writes go with the value alone until the gateway '
-            f'restarts. Set write_timestamp/write_quality to false on its document to skip '
-            f'the refused attempt.'
-        )
-        await self.emit_metric(
-            metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK,
-            tags={'pod_id': self.pod_id, 'server_name': server_id},
-        )
-
     def _is_session_bad(self, exc: Exception) -> bool:
         return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES
 

ingestor/managers/data_manager.py

@@ -12,22 +12,6 @@ from sientia_do.temporal.constants import now
 import ingestor.metrics as metrics
 
 
-def redact_uri(uri: str) -> str:
-    """
-    Returns ``uri`` without the user and password, safe to log
-    (``mongodb://user:pass@host:27017/db`` -> ``mongodb://***@host:27017/db``).
-    """
-    prefix, sep, rest = uri.partition('://')
-    if not sep:
-        prefix, rest = '', uri
-    ends = [i for i in (rest.find('/'), rest.find('?')) if i != -1]
-    end = min(ends) if ends else len(rest)
-    authority, tail = rest[:end], rest[end:]
-    if '@' not in authority:
-        return uri
-    return f'{prefix}{sep}***@{authority.rpartition("@")[2]}{tail}'
-
-
 class DataManager(SientiaMonitoring):
     """
     Manages data persistence operations for the OPC Ingestor.
@@ -82,8 +66,7 @@ class DataManager(SientiaMonitoring):
         )
 
         logger.info(
-            'Trying to initializing DataManager with MongoDB servers: '
-            f'{redact_uri(mongo_connection_string)}'
+            f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
         )
 
         self.connection_string = mongo_connection_string
@@ -99,9 +82,7 @@ class DataManager(SientiaMonitoring):
 
         self.metadata = metadata
 
-        logger.info(
-            f'DataManager initialized with MongoDB servers: {redact_uri(self.connection_string)}'
-        )
+        logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
 
     def shutdown(self):
         """

ingestor/metrics.py

@@ -156,12 +156,6 @@ OPC_WRITE_STATUS_BY_MODEL = Gauge(
     'Last write status per model (1=ok,0=error)',
     ['pod_id', 'model_id', 'status'],
 )
-OPC_WRITE_VALUE_ONLY_FALLBACK = Counter(
-    'opc_write_value_only_fallback_total',
-    'Servers that refused the timestamp/quality of a write (BadWriteNotSupported); '
-    'their writes go with the value alone',
-    ['pod_id', 'server_name'],
-)
 OPC_WRITE_STATUS_UPDATE_TOTAL = Counter(
     'opc_write_status_update_total',
     'write_status UPDATEs on the prediction rows',

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "2.0.0-rc4"
+version = "2.2.0-rc1"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"

requirements.txt

@@ -1,8 +1,12 @@
 asyncua
 redis
-# 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1
-# defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'").
-sientia_do==1.12.5
+# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
+# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
+# Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
+sientia_do==1.12.2
+# TEST ONLY (SIENTIAPDE-1975 dev test): SQLAlchemy 2.1 makes psycopg (v3) the default driver for
+# postgresql:// URLs; sientia_do 1.12.2 builds that URL and ships only psycopg2-binary.
+sqlalchemy<2.1
 prometheus_client
 pymongo
 nats-py>=2.6

tests/unit/connection/test_servers.py

@@ -149,67 +149,3 @@ async def test_mongo_errors_propagate_to_the_caller():
 
     with pytest.raises(RuntimeError, match='mongo down'):
         await resolver.resolve('1', {})
-
-
-# --- SIENTIAPDE-2202: write options on the server document ---
-
-
-@pytest.mark.asyncio
-async def test_write_options_come_from_the_connector():
-    resolver = make_resolver(
-        connector={**CONNECTOR, 'write_timestamp': False, 'write_quality': True},
-        opc_server={**OPC_SERVER, 'write_timestamp': True},
-    )
-
-    assert await resolver.write_options('1') == {'write_timestamp': False, 'write_quality': True}
-
-
-@pytest.mark.asyncio
-async def test_write_options_come_from_opc_servers_without_tag_database():
-    resolver = make_resolver(opc_server={**OPC_SERVER, 'write_quality': False})
-
-    assert await resolver.write_options('1') == {'write_quality': False}
-
-
-@pytest.mark.asyncio
-async def test_only_boolean_write_options_count():
-    resolver = make_resolver(
-        connector={**CONNECTOR, 'write_timestamp': 'false', 'write_quality': 0}
-    )
-
-    assert await resolver.write_options('1') == {}
-
-
-@pytest.mark.asyncio
-async def test_write_options_are_read_even_with_a_live_pooled_connection():
-    pooled = {'url': 'opc.tcp://read:1', 'server_uri': 'urn:read', 'use_security': True}
-    resolver = make_resolver(connector={**CONNECTOR, 'write_timestamp': False}, pooled=pooled)
-
-    assert await resolver.resolve('1', {}) == pooled
-    assert await resolver.write_options('1') == {'write_timestamp': False}
-
-
-@pytest.mark.asyncio
-async def test_write_options_share_the_lookup_cache():
-    resolver = make_resolver(connector={**CONNECTOR, 'write_quality': False})
-
-    await resolver.resolve('1', {})
-    await resolver.write_options('1')
-
-    assert resolver.mongo.find_one.await_count == 1
-
-
-@pytest.mark.asyncio
-async def test_write_options_of_an_unknown_server_are_empty():
-    resolver = make_resolver()
-
-    assert await resolver.write_options('9') == {}
-
-
-@pytest.mark.asyncio
-async def test_write_options_never_raise():
-    resolver = make_resolver()
-    resolver.mongo.find_one = AsyncMock(side_effect=RuntimeError('mongo down'))
-
-    assert await resolver.write_options('1') == {}
-    assert 'mongo down' in resolver.logger.warning.call_args.args[0]

tests/unit/engines/test_write.py

@@ -3,7 +3,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from asyncua import ua
-from asyncua.ua.uaerrors import BadTypeMismatch, BadWriteNotSupported
 
 import ingestor.metrics as metrics
 from ingestor.engines.write import WriteEngine
@@ -11,10 +10,9 @@ from ingestor.engines.write import WriteEngine
 SERVER_CFG = {'url': 'opc.tcp://kepserver:49320', 'server_uri': 'urn:kep', 'use_security': True}
 
 
-def make_resolver(cfg=SERVER_CFG, write_options=None):
+def make_resolver(cfg=SERVER_CFG):
     resolver = MagicMock()
     resolver.resolve = AsyncMock(return_value=cfg)
-    resolver.write_options = AsyncMock(return_value=write_options or {})
     return resolver
 
 
@@ -369,151 +367,3 @@ async def test_write_bad_timestamp_records_error_without_raising(timestamp):
     assert result.opc_status_code == 'bad_timestamp'
     eng.pool.get.assert_not_awaited()
     eng.sink.update_write_status.assert_awaited_once()
-
-
-# --- SIENTIAPDE-2202: timestamp/quality options, quality from the command, value-only retry ---
-
-
-@pytest.mark.asyncio
-async def test_server_options_override_the_gateway_defaults():
-    eng = make_engine(
-        write_timestamp=True,
-        write_quality=True,
-        servers=make_resolver(write_options={'write_timestamp': False}),
-    )
-    node = make_node()
-    eng.pool.get.return_value = make_client(node)
-
-    await eng.write(base_cmd())
-
-    eng.servers.write_options.assert_awaited_once_with('srv1')
-    dv = node.write_value.await_args.args[0]
-    assert dv.SourceTimestamp is None
-    assert dv.StatusCode is not None  # not set on the server: the gateway default applies
-
-
-@pytest.mark.asyncio
-async def test_command_options_override_the_server_options():
-    eng = make_engine(
-        servers=make_resolver(write_options={'write_timestamp': False, 'write_quality': False})
-    )
-    node = make_node()
-    eng.pool.get.return_value = make_client(node)
-
-    await eng.write(base_cmd(write_timestamp=True))
-
-    dv = node.write_value.await_args.args[0]
-    assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z')
-    assert dv.StatusCode is None
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    ('quality', 'expected'),
-    [
-        ('Good', ua.StatusCodes.Good),
-        ('Bad', ua.StatusCodes.Bad),
-        ('uncertain', ua.StatusCodes.Uncertain),
-        (None, ua.StatusCodes.Good),
-    ],
-)
-async def test_quality_comes_from_the_command(quality, expected):
-    eng = make_engine(write_quality=True)
-    node = make_node()
-    eng.pool.get.return_value = make_client(node)
-
-    result = await eng.write(base_cmd(quality=quality))
-
-    assert result.status == 'ok'
-    assert node.write_value.await_args.args[0].StatusCode.value == expected
-
-
-@pytest.mark.asyncio
-async def test_unknown_quality_records_error_without_writing():
-    eng = make_engine(write_quality=True)
-
-    result = await eng.write(base_cmd(quality='Excellent'))
-
-    assert result.status == 'error'
-    assert result.opc_status_code == 'bad_quality'
-    eng.pool.get.assert_not_awaited()
-    eng.sink.update_write_status.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_unknown_quality_is_ignored_when_quality_is_off():
-    eng = make_engine(write_quality=False)
-    node = make_node()
-    eng.pool.get.return_value = make_client(node)
-
-    result = await eng.write(base_cmd(quality='Excellent'))
-
-    assert result.status == 'ok'
-    assert node.write_value.await_args.args[0].StatusCode is None
-
-
-@pytest.mark.asyncio
-async def test_refused_timestamp_and_quality_retry_with_the_value_alone():
-    eng = make_engine(write_timestamp=True, write_quality=True)
-    node = make_node()
-    node.write_value.side_effect = [BadWriteNotSupported(), None]
-    eng.pool.get.return_value = make_client(node)
-
-    result = await eng.write(base_cmd())
-
-    assert result.status == 'ok'
-    first, retry = (c.args[0] for c in node.write_value.await_args_list)
-    assert first.SourceTimestamp is not None and first.StatusCode is not None
-    assert retry.SourceTimestamp is None and retry.StatusCode is None
-    assert retry.Value.Value == 42.0
-    assert 'srv1' in eng.logger.warning.call_args.args[0]
-    eng.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK,
-        tags={'pod_id': eng.pod_id, 'server_name': 'srv1'},
-    )
-    eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result)
-
-
-@pytest.mark.asyncio
-async def test_server_that_refused_gets_the_value_alone_next_time():
-    eng = make_engine(write_timestamp=True, write_quality=True)
-    node = make_node()
-    node.write_value.side_effect = [BadWriteNotSupported(), None, None]
-    eng.pool.get.return_value = make_client(node)
-
-    await eng.write(base_cmd())
-    result = await eng.write(base_cmd())
-
-    assert result.status == 'ok'
-    assert node.write_value.await_count == 3  # refused + retry, then one value-only write
-    last = node.write_value.await_args.args[0]
-    assert last.SourceTimestamp is None and last.StatusCode is None
-    eng.servers.write_options.assert_awaited_once()  # not looked up once remembered
-    assert eng.logger.warning.call_count == 1
-
-
-@pytest.mark.asyncio
-async def test_refused_value_alone_is_an_error():
-    eng = make_engine(write_timestamp=False, write_quality=False)
-    node = make_node()
-    node.write_value.side_effect = BadWriteNotSupported()
-    eng.pool.get.return_value = make_client(node)
-
-    result = await eng.write(base_cmd())
-
-    assert result.status == 'error'
-    assert result.opc_status_code == 'BadWriteNotSupported'
-    assert node.write_value.await_count == 1
-
-
-@pytest.mark.asyncio
-async def test_failed_value_only_retry_is_reported():
-    eng = make_engine(write_timestamp=True, write_quality=True)
-    node = make_node()
-    node.write_value.side_effect = [BadWriteNotSupported(), BadTypeMismatch()]
-    eng.pool.get.return_value = make_client(node)
-
-    result = await eng.write(base_cmd())
-
-    assert result.status == 'error'
-    assert result.opc_status_code == 'BadTypeMismatch'

tests/unit/managers/test_data_manager.py

@@ -1,10 +1,9 @@
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
-import pytest
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
-from ingestor.managers.data_manager import DataManager, redact_uri
+from ingestor.managers.data_manager import DataManager
 
 metadata = {
     'metadata': {
@@ -55,48 +54,6 @@ def test___init___success(mongodb_repository):
     assert data_manager.connection_string == 'mongodb://localhost:27017'
 
 
-@patch('ingestor.managers.data_manager.MongoDBRepository')
-def test___init___does_not_log_the_credentials(mongodb_repository):
-    logger_mock = MagicMock()
-    uri = 'mongodb://root:s3cr3t@mongodb.mongodb.svc.cluster.local:27017'
-
-    data_manager = DataManager(
-        metadata=metadata['metadata'],
-        mongo_connection_string=uri,
-        mongo_database='sientia',
-        logger=logger_mock,
-        notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
-    )
-
-    logged = ' '.join(str(c) for c in logger_mock.info.call_args_list)
-    assert 's3cr3t' not in logged
-    assert 'root' not in logged
-    logger_mock.info.assert_any_call(
-        'DataManager initialized with MongoDB servers: '
-        'mongodb://***@mongodb.mongodb.svc.cluster.local:27017'
-    )
-    assert data_manager.connection_string == uri
-
-
-@pytest.mark.parametrize(
-    ('uri', 'expected'),
-    [
-        ('mongodb://localhost:27017', 'mongodb://localhost:27017'),
-        (
-            'mongodb://u:p@h1:27017,h2:27017/db?authSource=admin',
-            'mongodb://***@h1:27017,h2:27017/db?authSource=admin',
-        ),
-        ('mongodb://u:p@h:27017?authSource=admin', 'mongodb://***@h:27017?authSource=admin'),
-        ('mongodb+srv://u@cluster.example.com', 'mongodb+srv://***@cluster.example.com'),
-        ('u:p@h:27017', '***@h:27017'),
-        ('mongodb://h:27017/db?x=a@b', 'mongodb://h:27017/db?x=a@b'),
-    ],
-)
-def test_redact_uri(uri, expected):
-    assert redact_uri(uri) == expected
-
-
 def test_shutdown(data_manager):
     close_mock = MagicMock()
     data_manager.mongo_repository.close = close_mock