You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
293 lines
8.6 KiB
293 lines
8.6 KiB
""" tests for the sartorius_logger package """ |
|
|
|
import pytest |
|
|
|
|
|
@pytest.fixture(scope="session") |
|
def settings_fixture(): |
|
from sartorius_logger.parsers import ParsedDuration, Settings |
|
from pathlib import Path |
|
|
|
duration = ParsedDuration(5, "m", 5 * 60) |
|
interval = ParsedDuration(10, "s", 10) |
|
directory = Path("/something/made/up") |
|
|
|
yield Settings(duration, interval, directory, "usb", 31) |
|
|
|
|
|
@pytest.fixture() |
|
def logger_fixture(): |
|
from . import stubs |
|
|
|
yield stubs.DummyLogger() |
|
|
|
|
|
@pytest.fixture() |
|
def scale_fixture(): |
|
from . import stubs |
|
|
|
yield stubs.DummyScale() |
|
|
|
|
|
def test_get_scale_info(scale_fixture): |
|
from sartorius_logger import get_scale_info |
|
|
|
result = get_scale_info(scale_fixture) |
|
|
|
assert result == { |
|
"Scale Model": "Quintix-1", |
|
"Scale Serial Number": "0815", |
|
"Software Version of Scale": "beta1", |
|
"Software Version of Control Unit": "1.2.3", |
|
} |
|
|
|
|
|
def test_get_scale_info_with_timeout(scale_fixture): |
|
from sartorius_logger import get_scale_info |
|
from sartoriusb import CMD_INFO_SNR |
|
|
|
scale_fixture.responses[CMD_INFO_SNR] = "" |
|
result = get_scale_info(scale_fixture) |
|
|
|
assert result == { |
|
"Scale Model": "Quintix-1", |
|
"Scale Serial Number": "", |
|
"Software Version of Scale": "beta1", |
|
"Software Version of Control Unit": "1.2.3", |
|
} |
|
|
|
|
|
def test_measure_and_log(scale_fixture, logger_fixture): |
|
from sartorius_logger import _measure_and_log |
|
from datetime import datetime |
|
|
|
result = _measure_and_log(42, scale_fixture, logger_fixture) |
|
|
|
reported_time = result.pop("time") |
|
assert isinstance(reported_time, datetime) |
|
assert (datetime.now() - reported_time).total_seconds() < 10 |
|
assert result == { |
|
"nr": 42, |
|
"mode": "G", |
|
"value": 850.1, |
|
"unit": None, |
|
"stable": False, |
|
"message": None, |
|
} |
|
assert logger_fixture.handle == [ |
|
"\t".join(["42", str(reported_time), "G", "+850.1", "", "False", ""]) |
|
] |
|
|
|
|
|
def test_measure_and_log_with_none_value(logger_fixture): |
|
from sartorius_logger import _measure_and_log |
|
from datetime import datetime |
|
from . import stubs |
|
|
|
scale = stubs.DummyScale(measurements=[]) |
|
result = _measure_and_log(42, scale, logger_fixture) |
|
|
|
reported_time = result.pop("time") |
|
assert isinstance(reported_time, datetime) |
|
assert result == { |
|
"nr": 42, |
|
"mode": None, |
|
"value": None, |
|
"unit": None, |
|
"stable": None, |
|
"message": "Connection Timeout", |
|
} |
|
assert logger_fixture.handle.captured == [ |
|
"\t".join( |
|
["42", str(reported_time), "", "", "", "", "Connection Timeout"] |
|
) |
|
] |
|
|
|
|
|
def test_no_progress_bar_returns_iterator_unchanged(): |
|
from sartorius_logger import no_progress_bar |
|
|
|
iterator = [] |
|
|
|
assert no_progress_bar(iterator) is iterator |
|
|
|
|
|
def test_get_log_file_path(settings_fixture): |
|
from sartorius_logger import _get_log_file_path |
|
from datetime import datetime |
|
from pathlib import Path |
|
|
|
result = _get_log_file_path(settings_fixture) |
|
|
|
assert isinstance(result, Path) |
|
assert result.suffix == ".txt" |
|
dt = datetime.strptime(result.stem, "%Y-%m-%d %H-%M-%S") |
|
assert (datetime.now() - dt).total_seconds() < 10 |
|
|
|
|
|
def test_log_measurement_info(logger_fixture, settings_fixture): |
|
from sartorius_logger import _log_measurement_info |
|
|
|
result = _log_measurement_info(logger_fixture, settings_fixture) |
|
|
|
assert result == { |
|
"Measurements": 31, |
|
"Duration": "5m", |
|
"Interval": "10s", |
|
"Com-Port": "usb", |
|
} |
|
assert logger_fixture.handle == [ |
|
"[Measurement Settings]", |
|
"Measurements\t31", |
|
"Duration\t5m", |
|
"Interval\t10s", |
|
"Com-Port\tusb", |
|
"", |
|
] |
|
|
|
|
|
def test_log_scale_info(logger_fixture, scale_fixture): |
|
from sartorius_logger import _log_scale_info |
|
|
|
result = _log_scale_info(logger_fixture, scale_fixture) |
|
|
|
assert result == { |
|
"Scale Model": "Quintix-1", |
|
"Scale Serial Number": "0815", |
|
"Software Version of Scale": "beta1", |
|
"Software Version of Control Unit": "1.2.3", |
|
} |
|
assert logger_fixture.handle == [ |
|
"[Scale Info]", |
|
"Scale Model\tQuintix-1", |
|
"Scale Serial Number\t0815", |
|
"Software Version of Scale\tbeta1", |
|
"Software Version of Control Unit\t1.2.3", |
|
"", |
|
] |
|
|
|
|
|
def test_measure_series(mocker, settings_fixture, logger_fixture): |
|
from sartorius_logger import measure_series, Result |
|
from pandas import DataFrame |
|
from pathlib import Path |
|
|
|
from .stubs import DummyScale |
|
|
|
mocker.patch("sartorius_logger.sartoriusb.SartoriusUsb", DummyScale) |
|
mocker.patch("sartorius_logger.time.sleep") |
|
result = measure_series(settings_fixture, data_logger=logger_fixture) |
|
|
|
assert isinstance(result, Result) |
|
|
|
assert isinstance(result.log_file, Path) |
|
assert result.log_file.parent == Path("log") |
|
assert result.log_file.suffix == ".txt" |
|
|
|
assert isinstance(result.info, DataFrame) |
|
assert result.info.to_dict() == { |
|
1: { |
|
"Measurements": 31, |
|
"Duration": "5m", |
|
"Interval": "10s", |
|
"Com-Port": "usb", |
|
} |
|
} |
|
|
|
assert isinstance(result.scale, DataFrame) |
|
assert result.scale.to_dict() == { |
|
1: { |
|
"Scale Model": "Quintix-1", |
|
"Scale Serial Number": "0815", |
|
"Software Version of Scale": "beta1", |
|
"Software Version of Control Unit": "1.2.3", |
|
} |
|
} |
|
|
|
assert isinstance(result.data, DataFrame) |
|
df_dict = {k: list(v.values()) for k, v in result.data.to_dict().items()} |
|
|
|
assert df_dict["nr"] == list(range(1, 32)) |
|
assert df_dict["mode"] == ["G"] * 10 + [None, "G", "G"] + [None] * 18 |
|
# some hassle with nan values |
|
values = [v if "." in str(v) else None for v in df_dict["value"]] |
|
assert ( |
|
values |
|
== [850.1, 851.0, 852.2, 853.0, 854.4, 857.0, 857.0] |
|
+ [858.7, 859.0, -1000.0, None, 0.0, 0.0] |
|
+ [None] * 18 |
|
) |
|
assert ( |
|
df_dict["unit"] |
|
== [None, None, None, "mg", "mg", "mg", "mg"] |
|
+ ["mg", "mg", "mg", None, None, "mg"] |
|
+ [None] * 18 |
|
) |
|
assert ( |
|
df_dict["stable"] |
|
== [False, False, False, True, True, True, True] |
|
+ [True, True, True, None, False, True] |
|
+ [None] * 18 |
|
) |
|
assert ( |
|
df_dict["message"] |
|
== [None] * 10 + ["Low", None, None] + ["Connection Timeout"] * 18 |
|
) |
|
|
|
log = logger_fixture.handle.captured |
|
assert len(log) == 45 |
|
assert log[0] == "[Measurement Settings]" |
|
assert log[6] == "[Scale Info]" |
|
assert log[12] == "[Measured Data]" |
|
assert log[13] == "Nr\tTime\tMode\tValue\tUnit\tStable\tMessage" |
|
assert log[14].startswith("1\t") |
|
assert log[14].endswith("\tG\t+850.1\t\tFalse\t") |
|
assert log[20].startswith("7\t") |
|
assert log[20].endswith("\tG\t+857.0\tmg\tTrue\t") |
|
assert log[24].startswith("11\t") |
|
assert log[24].endswith("\t\t\t\t\tLow") |
|
assert log[-1].startswith("31\t") |
|
assert log[-1].endswith("\t\t\t\t\tConnection Timeout") |
|
|
|
|
|
def test_export_as_excel(mocker, settings_fixture, logger_fixture): |
|
from sartorius_logger import measure_series, export_as_excel |
|
from pathlib import Path |
|
from tempfile import TemporaryDirectory |
|
from .stubs import DummyScale |
|
|
|
mocker.patch("sartorius_logger.sartoriusb.SartoriusUsb", DummyScale) |
|
mocker.patch("sartorius_logger.time.sleep") |
|
result = measure_series(settings_fixture, data_logger=logger_fixture) |
|
|
|
with TemporaryDirectory() as temp_path: |
|
temp_log_file = Path(temp_path) / "measurement.txt" |
|
expected_xls_path = Path(temp_path) / "measurement.xlsx" |
|
result = result._replace(log_file=temp_log_file) |
|
export_as_excel(result) |
|
|
|
assert expected_xls_path.is_file() |
|
|
|
|
|
def test_cli(mocker, settings_fixture): |
|
mocker.patch( |
|
"sartorius_logger.parse_cli_arguments", return_value=settings_fixture |
|
) |
|
mocker.patch("sartorius_logger.measure_series", return_value="Results") |
|
mocker.patch("sartorius_logger.export_as_excel") |
|
|
|
from sartorius_logger import cli, export_as_excel, measure_series |
|
from sartorius_logger.datalogger import DataLogger |
|
from tqdm import tqdm |
|
from unittest.mock import call |
|
|
|
cli() |
|
|
|
assert measure_series.call_count == 1 |
|
print(measure_series.call_args) |
|
call_args, call_kargs = measure_series.call_args |
|
assert call_args == (settings_fixture,) |
|
assert call_kargs["progress_bar"] == tqdm |
|
assert isinstance(call_kargs["data_logger"], DataLogger) |
|
assert export_as_excel.call_count == 1 |
|
assert export_as_excel.call_args == call("Results")
|
|
|