""" 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") @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")