Skip to content
Snippets Groups Projects
Commit 57ca9227 authored by Johan Venter's avatar Johan Venter
Browse files

Revert "Merge branch 'SAR-149-transaction-ids' into 'master'" 

This reverts merge request !7
parent 52d118c0
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
CHANGELOG.rst
+ 0
− 6
View file @ 57ca9227
......@@ -5,12 +5,6 @@ Change Log
All notable changes to this project will be documented in this file.
This project adheres to `Semantic Versioning <http://semver.org/>`_.
v0.4.0
******
Add a `transaction` context handler to propagate transaction IDs in logs for tracing transactions
Usage like `with transaction('My Command', parameters) as transaction_id:`.
v0.3.0
******
......
Pipfile
+ 2
− 4
View file @ 57ca9227
[[source]]
url = 'https://pypi.python.org/simple'
url = "https://nexus.engageska-portugal.pt/repository/pypi-proxy/simple"
verify_ssl = true
name = 'pypi'
name = "nexus-proxy"
[[source]]
url = "https://nexus.engageska-portugal.pt/repository/pypi/simple"
......@@ -16,7 +16,6 @@ pylint = "*"
pylint-junit = "*"
pytest = "*"
pytest-cov = "*"
pytest-mock = "*"
pytest-pylint = "*"
pytest-json-report = "*"
python-dotenv = ">=0.5.1"
......@@ -25,7 +24,6 @@ sphinx_rtd_theme = "*"
sphinx-autobuild = "*"
sphinxcontrib-websupport = "*"
recommonmark = "*"
ska-skuid = {version=">=1.2.0", index="nexus-hosted"}
[dev-packages]
......
README.md
+ 0
− 47
View file @ 57ca9227
......@@ -112,53 +112,6 @@ By default, calls to `configure_logging` do not disable existing non-root logger
multiple calls to the function, although that will generally not be required. This behaviour can be
overridden using the `"disable_existing_loggers"` key.
### Logging Transaction IDs
A transaction context handler is available to inject ids fetched from the the skuid service into logs. The transaction id will be logged on entry and exit of the context handler. In the event of an exception, the transaction id will be logged with the exception stack trace. The ID generated depends on whether or not the SKUID_URL environment variable is set.
```python
from ska.logging import transaction
def command(self, parameter_json):
parameters = json.reads(parameter_json)
with transaction('My Command', parameters) as transaction_id:
# ...
parameters['transaction_id'] = transaction_id
device.further_command(json.dumps(parameters))
# ...
```
The context handler logs to the root logger by default. Logs can be sent to a custom logger by passing a logger object as a keyword argument. Use `configure_logging` method for ska formatted logs.
**Example ska formatted logs for successful transaction**
Log message formats:
- On Entry:
- Transaction [id]: Enter [name] with parameters [arguments]
- On Exit:
- Transaction [id]: Exit [name]
- On exception:
- Transaction [id]: Exception [name]
--Stacktrace--
```txt
1|2020-09-22T11:43:22.760Z|INFO|Dummy-1|__enter__|transactions.py#124|tango-device:log/test/upstream|Transaction [txn-local-20200922-232544376]: Enter[CallWithContext] with parameters[{"ArgKey2": "ArgVal2"}]
1|2020-09-22T11:43:22.782Z|INFO|Dummy-1|__exit__|transactions.py#133|tango-device:log/test/upstream|Transaction[txn-local-20200922-232544376]: Exit[CallWithContext]
```
**Example ska formatted logs for failed transaction**
```txt
1|2020-09-22T11:45:07.122Z|ERROR|Dummy-1|__exit__|transactions.py#131|tango-device:log/test/upstream|Transaction[txn-local-20200922-362753747]: Exception[CallRaisesException]
Traceback (most recent call last):
File "LogTestUpstream.py", line 47, in CallRaisesException
raise RuntimeError("An exception has occured")
RuntimeError: An exception has occured
1|2020-09-22T11:45:07.123Z|INFO|Dummy-1|__exit__|transactions.py#133|tango-device:log/test/upstream|Transaction[txn-local-20200922-362753747]: Exit[CallRaisesException]
```
Requirements
------------
......
setup.py
+ 1
− 1
View file @ 57ca9227
......@@ -45,6 +45,6 @@ setup(
"recommonmark",
],
use_katversion=True,
tests_require=["pytest", "pytest-cov", "pytest-json-report", "pytest-mock", "pycodestyle"],
tests_require=["pytest", "pytest-cov", "pytest-json-report", "pycodestyle"],
extras_require={"dev": ["prospector[with_pyroma]", "black", "isort"]},
)
src/ska/logging/__init__.py
+ 0
− 4
View file @ 57ca9227
......@@ -5,7 +5,6 @@
__all__ = (
"configure_logging",
"get_default_formatter",
"transaction",
)
__author__ = "Anton Joubert"
__email__ = "ajoubert+ska@ska.ac.za"
......@@ -15,9 +14,6 @@ from .configuration import (
configure_logging,
get_default_formatter,
)
from .transactions import (
Transaction as transaction,
)
# BEGIN VERSION CHECK
......
src/ska/logging/transactions.py deleted 100644 → 0
+ 0
− 209
View file @ 52d118c0
# -*- coding: utf-8 -*-
"""This module provides the transaction logging mechanism."""
import json
import logging
import os
from typing import Mapping, Optional, Text
from ska.skuid.client import SkuidClient
class Transaction:
"""Transaction context handler.
Provides:
- Transaction ID:
- Re-use existing transaction ID, if available
- If no transaction ID, or empty or None, then generate a new ID
- context handler returns the transaction ID used
- Log messages on entry, exit, and exception
.. Examples::
def command(self, parameter_json):
parameters = json.reads(parameter_json)
with ska.logging.transaction('My Command', parameters) as transaction_id:
# ...
parameters['transaction_id'] = transaction_id
device.further_command(json.dumps(pars))
# ...
def command(self, parameter_json):
parameters = json.reads(parameter_json)
with ska.logging.transaction('My Command', parameters, transaction_id="123") as transaction_id:
# ...
parameters['transaction_id'] = transaction_id
device.further_command(json.dumps(pars))
# ...
def command(self, parameter_json):
parameters = json.reads(parameter_json)
parameters["txn_id_key"] = 123
with ska.logging.transaction('My Command', parameters, transaction_id_key="txn_id_key") as transaction_id:
# ...
parameters['transaction_id'] = transaction_id
device.further_command(json.dumps(pars))
# ...
Log message formats:
On Entry:
Transaction [id]: Enter [name] with parameters [arguments]
On Exit:
Transaction [id]: Exit [name]
On exception:
Transaction [id]: Exception [name]
Stacktrace
"""
def __init__(
self,
name: str,
params: dict = {},
transaction_id: str = "",
transaction_id_key: str = "transaction_id",
logger: Optional[logging.Logger] = None, # pylint: disable=E1101
):
"""Create the transaction context handler.
A new transaction ID is generated if none is passed in via `transaction_id` or
in `params`.
If there is a transaction ID in `params` and `transaction_id` is also passed in
then the passed in `transaction_id` will take precedence.
However, if both transaction IDs provided in `params` and `transaction_id` are deemed
invalid (not a string or an empty string), then a new transaction ID will be generated.
By default the key `transaction_id` will be used to get a transaction ID out of
`params`. If a different key is required then `transaction_id_key` can be
specified.
Parameters
----------
name : str
A description for the context. This is usually the Tango device command.
params : dict, optional
The parameters will be logged and will be used to retrieve the transaction
ID if `transaction_id` is not passed in, by default {}
transaction_id : str, optional
The transaction ID to be used for the context, by default ""
transaction_id_key : str, optional
The key to use to get the transaction ID from params,
by default "transaction_id"
logger : logging.Logger, optional
The logger to use for logging, by default None.
If no logger is specified a new one named `ska.transaction` will be used.
Raises
------
TransactionParamsError
If the `params` passed is not valid.
"""
if not isinstance(params, Mapping):
raise TransactionParamsError("params must be dict-like (Mapping)")
if logger:
self.logger = logger
else:
self.logger = logging.getLogger("ska.transaction") # pylint: disable=E1101
self._name = name
self._params = params
self._transaction_id_key = transaction_id_key
self._transaction_id = self._get_id_from_params_or_generate_new_id(transaction_id)
if transaction_id and params.get(self._transaction_id_key):
self.logger.info(
f"Received 2 transaction IDs {transaction_id} and"
f" {params.get(transaction_id_key)}, using {self._transaction_id}"
)
def __enter__(self):
params_json = json.dumps(self._params)
self.logger.info(
f"Transaction [{self._transaction_id}]: Enter[{self._name}] "
f"with parameters[{params_json}]"
)
return self._transaction_id
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type:
self.logger.exception(f"Transaction[{self._transaction_id}]: Exception[{self._name}]")
self.logger.info(f"Transaction[{self._transaction_id}]: Exit[{self._name}]")
if exc_type:
raise
def _get_id_from_params_or_generate_new_id(self, transaction_id):
"""At first use the transaction_id passed or use the transaction_id_key to get the
transaction ID from the parameters or generate a new one if it's not there.
Parameters
----------
transaction_id : [String]
[The transaction ID]
Returns
-------
[String]
[transaction ID]
"""
_transaction_id = (
transaction_id if transaction_id else self._params.get(self._transaction_id_key)
)
if not self._is_valid_id(_transaction_id):
_transaction_id = self._generate_new_id()
self.logger.info(f"Generated transaction ID {_transaction_id}")
return _transaction_id
def _is_valid_id(self, transaction_id):
"""Check if the ID is valid
Parameters
----------
transaction_id : [String]
[The transaction ID]
Returns
-------
[bool]
[Whether the ID is valid or not]
"""
if isinstance(transaction_id, Text) and transaction_id.strip():
return True
return False
def _generate_new_id(self):
"""Use TransactionIdGenerator to generate a new transaction ID
Returns
-------
[String]
[The newly generated transaction ID]
"""
id_generator = TransactionIdGenerator()
return id_generator.next() # pylint: disable=E1102
class TransactionIdGenerator:
"""
TransactionIdGenerator retrieves a transaction id from skuid.
Skuid may fetch the id from the service if the SKUID_URL is set or alternatively generate one.
"""
def __init__(self):
if os.environ.get("SKUID_URL"):
client = SkuidClient(os.environ["SKUID_URL"])
self._get_id = client.fetch_transaction_id
else:
self._get_id = SkuidClient.get_local_transaction_id
def next(self):
return self._get_id()
class TransactionParamsError(TypeError):
"""Invalid data type for transaction parameters."""
tests/conftest.py deleted 100644 → 0
+ 0
− 138
View file @ 52d118c0
"""
A module defining pytest fixtures for testing ska.logging.
"""
import logging
import pytest
from ska.logging import configure_logging
@pytest.fixture
def reset_logging():
"""Cleanup logging module's state after each test (at least try to)."""
yield
manager = logging.root.manager
manager.disabled = logging.NOTSET
for logger in list(manager.loggerDict.values()) + [logging.root]:
if isinstance(logger, logging.Logger):
logger.setLevel(logging.NOTSET)
logger.propagate = True
logger.disabled = False
logger.filters.clear()
handlers = logger.handlers.copy()
for handler in handlers:
# Copied from `logging.shutdown`.
try:
handler.acquire()
handler.flush()
handler.close()
except (OSError, ValueError):
pass
finally:
handler.release()
logger.removeHandler(handler)
class AppendHandler(logging.Handler):
"""Handler that keeps a history of the records and formatted log messages."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.logs = []
self.records = []
def emit(self, record):
self.records.append(record)
self.logs.append(self.format(record))
"""Override configuration for recording loggers"""
RECORDER_OVERRIDES = {
"handlers": {"recorder": {"()": AppendHandler, "formatter": "default"}},
"root": {"handlers": ["console", "recorder"]},
}
@pytest.fixture
def default_logger():
"""Return user logger instance with default configuration."""
configure_logging()
yield logging.getLogger("ska.test.app")
@pytest.fixture
def recording_logger():
"""Return user logger, including a recording handler.
The additional handler has the name "recorder". It uses the default formatter,
and stores all formatted output strings as a list in its `logs` attribute.
It also keeps a list of the raw log records in its `records` attribute.
Note: we use this instead of pytest's `caplog` fixture because we cannot change
the formatter that it uses.
"""
configure_logging(overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
@pytest.fixture
def recording_tags_logger():
"""Return user logger like :func:`recording_logger`, but including tags filter."""
class MyFilter(logging.Filter):
def filter(self, record):
record.tags = "key1:value1,key2:value2"
return True
configure_logging(tags_filter=MyFilter, overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
def get_first_record_and_log_message(logger):
recorder = get_named_handler(logger, "recorder")
record = recorder.records[0]
log_message = recorder.logs[0]
return record, log_message
def get_second_record_and_log_message(logger):
recorder = get_named_handler(logger, "recorder")
record = recorder.records[1]
log_message = recorder.logs[1]
return record, log_message
def get_last_record_and_log_message(logger):
recorder = get_named_handler(logger, "recorder")
record = recorder.records[-1]
log_message = recorder.logs[-1]
return record, log_message
def get_all_record_logs(logger):
recorder = get_named_handler(logger, "recorder")
return recorder.logs
def get_named_handler(logger, name):
"""Search up through logger hierarchy to find a handler with the specified name."""
while logger:
for handler in logger.handlers:
if handler.name == name:
return handler
logger = logger.parent
@pytest.fixture
def id_generator_stub(mocker):
"""Replace the standard transactions.TransactionIdGenerator with a simple stub implementation."""
class TransactionIdGeneratorStub:
last_id = "NOT SET"
call_count = 0
def next(self):
TransactionIdGeneratorStub.last_id = "XYZ-789"
TransactionIdGeneratorStub.call_count += 1
return TransactionIdGeneratorStub.last_id
mocker.patch('ska.logging.transactions.TransactionIdGenerator', TransactionIdGeneratorStub)
yield TransactionIdGeneratorStub
tests/test_configuration.py
+ 109
− 13
View file @ 57ca9227
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Tests for the logging configuration module."""
"""Tests for the logging module."""
import logging
import logging.handlers
import time
import pytest
import ska.logging.configuration as configuration
from ska.logging import configure_logging, get_default_formatter
from tests.conftest import get_named_handler, get_first_record_and_log_message
@pytest.fixture
def reset_logging():
"""Cleanup logging module's state after each test (at least try to)."""
yield
manager = logging.root.manager
manager.disabled = logging.NOTSET
for logger in list(manager.loggerDict.values()) + [logging.root]:
if isinstance(logger, logging.Logger):
logger.setLevel(logging.NOTSET)
logger.propagate = True
logger.disabled = False
logger.filters.clear()
handlers = logger.handlers.copy()
for handler in handlers:
# Copied from `logging.shutdown`.
try:
handler.acquire()
handler.flush()
handler.close()
except (OSError, ValueError):
pass
finally:
handler.release()
logger.removeHandler(handler)
def get_named_handler(logger, name):
"""Search up through logger hierarchy to find a handler with the specified name."""
while logger:
for handler in logger.handlers:
if handler.name == name:
return handler
logger = logger.parent
class AppendHandler(logging.Handler):
"""Handler that keeps a history of the records and formatted log messages."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.logs = []
self.records = []
def emit(self, record):
self.records.append(record)
self.logs.append(self.format(record))
"""Override configuration for recording loggers"""
RECORDER_OVERRIDES = {
"handlers": {"recorder": {"()": AppendHandler, "formatter": "default"}},
"root": {"handlers": ["console", "recorder"]},
}
@pytest.fixture
def default_logger():
"""Return user logger instance with default configuration."""
configure_logging()
yield logging.getLogger("ska.test.app")
@pytest.fixture
def recording_logger():
"""Return user logger, including a recording handler.
The additional handler has the name "recorder". It uses the default formatter,
and stores all formatted output strings as a list in its `logs` attribute.
It also keeps a list of the raw log records in its `records` attribute.
Note: we use this instead of pytest's `caplog` fixture because we cannot change
the formatter that it uses.
"""
configure_logging(overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
@pytest.fixture
def recording_tags_logger():
"""Return user logger like :func:`recording_logger`, but including tags filter."""
class MyFilter(logging.Filter):
def filter(self, record):
record.tags = "key1:value1,key2:value2"
return True
configure_logging(tags_filter=MyFilter, overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
@pytest.mark.usefixtures("reset_logging")
......@@ -43,7 +132,9 @@ class TestConfigureLogging:
def test_default_uses_utc_time(self, recording_logger):
recording_logger.info("UTC message")
record, log_message = get_first_record_and_log_message(recording_logger)
recorder = get_named_handler(recording_logger, "recorder")
record = recorder.records[0]
log_message = recorder.logs[0]
expected_time = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
assert "UTC message" in log_message
assert expected_time in log_message # just testing UTC, so ignore the milliseconds part
......@@ -60,7 +151,9 @@ class TestConfigureLogging:
def test_tags_filter_emits_tags_value(self, recording_tags_logger):
recording_tags_logger.info("Tags message")
record, log_message = get_first_record_and_log_message(recording_tags_logger)
recorder = get_named_handler(recording_tags_logger, "recorder")
record = recorder.records[0]
log_message = recorder.logs[0]
assert record.tags == "key1:value1,key2:value2"
assert log_message.endswith("|key1:value1,key2:value2|Tags message")
......@@ -79,27 +172,30 @@ class TestConfigureLogging:
class TestGetDefaultFormatter:
"""Tests for :func:`~ska.logging.configuration.get_default_formatter`."""
def get_logged_test_message(self, logger):
def get_recorded_message(self, logger):
logger.info("test message")
return get_first_record_and_log_message(logger)
recorder = get_named_handler(logger, "recorder")
record = recorder.records[0]
message = recorder.logs[0]
return record, message
def test_default_no_tags(self, recording_logger):
formatter = get_default_formatter()
record, log_message = self.get_logged_test_message(recording_logger)
record, message = self.get_recorded_message(recording_logger)
actual_log_message = formatter.format(record)
assert actual_log_message == log_message
assert actual_log_message == message
def test_get_tags_disabled(self, recording_logger):
formatter = get_default_formatter(tags=False)
record, log_message = self.get_logged_test_message(recording_logger)
record, message = self.get_recorded_message(recording_logger)
actual_log_message = formatter.format(record)
assert actual_log_message == log_message
assert actual_log_message == message
def test_get_tags_enabled(self, recording_tags_logger):
formatter = get_default_formatter(tags=True)
record, log_message = self.get_logged_test_message(recording_tags_logger)
record, message = self.get_recorded_message(recording_tags_logger)
actual_log_message = formatter.format(record)
assert actual_log_message == log_message
assert actual_log_message == message
class TestOverride:
......
tests/test_transactions.py deleted 100644 → 0
+ 0
− 184
View file @ 52d118c0
# -*- coding: utf-8 -*-
"""Tests for the logging transactions module."""
import json
import os
import pytest
import concurrent.futures
from unittest.mock import patch, MagicMock
from ska.logging import transaction
from ska.logging.transactions import TransactionIdGenerator, TransactionParamsError
from tests.conftest import (
get_first_record_and_log_message,
get_last_record_and_log_message,
get_all_record_logs,
get_second_record_and_log_message,
)
class TestTransactionIdGeneration:
"""Tests for :class:`~ska.logging.transaction` related to ID generation."""
def test_error_if_params_type_is_not_mapping(self):
parameters = []
with pytest.raises(TransactionParamsError):
with transaction("name", parameters):
pass
def test_preference_order(self, id_generator_stub):
parameters = {
"other": "config",
"transaction_id": "xyz123",
"other_transaction_id_key": "def789",
}
with transaction("name", parameters, transaction_id="abc1234") as transaction_id:
assert transaction_id == "abc1234"
parameters = {
"other": "config",
"transaction_id": "xyz123",
"other_transaction_id_key": "def789",
}
with transaction(
"name", parameters, transaction_id_key="other_transaction_id_key"
) as transaction_id:
assert transaction_id == "def789"
parameters = {
"other": "config",
"transaction_id": "xyz123",
"other_transaction_id_key": "def789",
}
with transaction(
"name",
parameters,
transaction_id="abc1234",
transaction_id_key="other_transaction_id_key",
) as transaction_id:
assert transaction_id == "abc1234"
def test_new_id_generated_if_invalid_ids_passed_in_params(self, id_generator_stub):
parameters = {
"other": "config",
"transaction_id": 54321,
}
with transaction("name", parameters, transaction_id=12345) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_id_is_empty(self, id_generator_stub):
parameters = {"transaction_id": "", "other": "config"}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_id_is_none(self, id_generator_stub):
parameters = {"transaction_id": None, "other": "config"}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_id_is_only_white_space(self, id_generator_stub):
parameters = {"transaction_id": "\t\n \r\n", "other": "config"}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_id_is_not_string_type(self, id_generator_stub):
parameters = {"transaction_id": 1234.5, "other": "config"}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_id_is_not_present(self, id_generator_stub):
parameters = {"other": "config"}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_new_id_generated_if_params_empty(self, id_generator_stub):
parameters = {}
with transaction("name", parameters) as transaction_id:
assert transaction_id == id_generator_stub.last_id
def test_id_provider_only_used_once_for_one_new_id(self, id_generator_stub):
parameters = {}
with transaction("name", parameters):
assert id_generator_stub.call_count == 1
def test_id_provider_not_used_for_existing_valid_id(self, id_generator_stub):
parameters = {"transaction_id": "abc1234"}
with transaction("name", parameters):
assert id_generator_stub.call_count == 0
class TestTransactionLogging:
"""Tests for :class:`~ska.logging.transaction` related to logging."""
def test_name_and_id_and_params_in_context_handler(self, id_generator_stub, recording_logger):
parameters = {"other": ["config", 1, 2, 3.0]}
with transaction("name", parameters) as transaction_id:
pass
_, first_log_message = get_first_record_and_log_message(recording_logger)
_, second_log_message = get_second_record_and_log_message(recording_logger)
_, last_log_message = get_last_record_and_log_message(recording_logger)
# __enter__ log message
assert "Generated transaction ID" in first_log_message
assert "Enter" in second_log_message
assert "name" in second_log_message
assert "other" in second_log_message
assert transaction_id in second_log_message
assert json.dumps(parameters) in second_log_message
# __exit__ log message
assert "Exit" in last_log_message
assert "name" in last_log_message
assert transaction_id in last_log_message
def test_exception_logs_transaction_id_and_command(self, id_generator_stub, recording_logger):
parameters = {"other": ["config", 1, 2, 3.0]}
with pytest.raises(RuntimeError):
with transaction("name", parameters) as transaction_id:
raise RuntimeError("Something went wrong")
record_logs = get_all_record_logs(recording_logger)
for log_msg in record_logs:
if "RuntimeError" in log_msg and transaction_id in log_msg and "name" in log_msg:
return
assert 0, f"RuntimeError and transaction tag not found in exception logs: {record_logs}"
def test_specified_logger(self):
logger = MagicMock()
parameters = {}
with transaction("name", parameters, logger=logger) as transaction_id:
logger.info("A message")
for i, message in enumerate(["Generated", "Enter", "A message", "Exit"]):
assert logger.info.call_args_list[i].starts_with(message)
assert logger.info.call_count == 4, f"Log calls incorrect {logger.info.call_args_list}"
class TestTransactionIdGenerator:
"""Tests for :class:`~ska.logging.transactions.TransactionIdGenerator`."""
def test_local_id_generator_increments_on_next(self, monkeypatch):
monkeypatch.delenv("SKUID_URL", raising=False)
generator = TransactionIdGenerator()
assert generator.next()
assert generator.next() != generator.next()
def test_remote_id_generator_increments_on_next(self, monkeypatch):
monkeypatch.setenv("SKUID_URL", "endpoint/to/skuid-client")
with patch("ska.skuid.client.requests.get") as mocked_req:
response = MagicMock()
response.json.side_effect = [
json.dumps({"transaction_id": 1}),
json.dumps({"transaction_id": 2}),
json.dumps({"transaction_id": 3}),
]
mocked_req.return_value = response
generator = TransactionIdGenerator()
assert generator.next() == 1
assert generator.next() == 2
assert generator.next() == 3
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment