Compare commits
5 Commits
test-SIENT
...
279484b2aa
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
279484b2aa | ||
|
|
ce67105c16 | ||
|
|
d83917a30f | ||
| 54c854bb0a | |||
|
|
06caa88ed8 |
@@ -12,6 +12,22 @@ from sientia_do.temporal.constants import now
|
|||||||
import ingestor.metrics as metrics
|
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):
|
class DataManager(SientiaMonitoring):
|
||||||
"""
|
"""
|
||||||
Manages data persistence operations for the OPC Ingestor.
|
Manages data persistence operations for the OPC Ingestor.
|
||||||
@@ -66,7 +82,8 @@ class DataManager(SientiaMonitoring):
|
|||||||
)
|
)
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
|
'Trying to initializing DataManager with MongoDB servers: '
|
||||||
|
f'{redact_uri(mongo_connection_string)}'
|
||||||
)
|
)
|
||||||
|
|
||||||
self.connection_string = mongo_connection_string
|
self.connection_string = mongo_connection_string
|
||||||
@@ -82,7 +99,9 @@ class DataManager(SientiaMonitoring):
|
|||||||
|
|
||||||
self.metadata = metadata
|
self.metadata = metadata
|
||||||
|
|
||||||
logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
|
logger.info(
|
||||||
|
f'DataManager initialized with MongoDB servers: {redact_uri(self.connection_string)}'
|
||||||
|
)
|
||||||
|
|
||||||
def shutdown(self):
|
def shutdown(self):
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -561,8 +561,9 @@ class IngestorManager(SientiaMonitoring):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
# list() snapshots the readers: the loop awaits, so a reader added or removed
|
# list() snapshots the readers: the loop awaits, so a reader added or removed
|
||||||
# meanwhile must not break the iteration.
|
# meanwhile must not break the iteration. SonarQube flags it as an unnecessary
|
||||||
for server, reader in list(self.opc_managers.items()): # NOSONAR
|
# list() (S7504); the issue is accepted in SonarQube for this reason.
|
||||||
|
for server, reader in list(self.opc_managers.items()):
|
||||||
if slot in reader.subscriptions:
|
if slot in reader.subscriptions:
|
||||||
await self._unsubscribe_reader(server, reader, slot)
|
await self._unsubscribe_reader(server, reader, slot)
|
||||||
|
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "ingestor"
|
name = "ingestor"
|
||||||
version = "2.1.0-rc1"
|
version = "2.2.0-rc1"
|
||||||
description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
|
description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
|
|||||||
@@ -1,9 +1,8 @@
|
|||||||
asyncua
|
asyncua
|
||||||
redis
|
redis
|
||||||
# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
|
# 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1
|
||||||
# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
|
# defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'").
|
||||||
# Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
|
sientia_do==1.12.5
|
||||||
sientia_do==1.12.2
|
|
||||||
prometheus_client
|
prometheus_client
|
||||||
pymongo
|
pymongo
|
||||||
nats-py>=2.6
|
nats-py>=2.6
|
||||||
|
|||||||
@@ -1,9 +1,10 @@
|
|||||||
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
from pytest import fixture, mark
|
from pytest import fixture, mark
|
||||||
from sientia_do.notifications.models import NotificationLevel
|
from sientia_do.notifications.models import NotificationLevel
|
||||||
|
|
||||||
from ingestor.managers.data_manager import DataManager
|
from ingestor.managers.data_manager import DataManager, redact_uri
|
||||||
|
|
||||||
metadata = {
|
metadata = {
|
||||||
'metadata': {
|
'metadata': {
|
||||||
@@ -54,6 +55,48 @@ def test___init___success(mongodb_repository):
|
|||||||
assert data_manager.connection_string == 'mongodb://localhost:27017'
|
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):
|
def test_shutdown(data_manager):
|
||||||
close_mock = MagicMock()
|
close_mock = MagicMock()
|
||||||
data_manager.mongo_repository.close = close_mock
|
data_manager.mongo_repository.close = close_mock
|
||||||
|
|||||||
Reference in New Issue
Block a user