test_reader.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-

import pytest
import numpy as np

from saqc.core.reader import checkConfig
from saqc.core.config import Fields as F
from test.common import initData, initMetaDict, initMetaString, TESTFLAGGER, TESTNODATA


@pytest.fixture
def data():
    return initData(3)


def test_configPreparation(data):
    var1, var2, var3, *_ = data.columns
    date = data.index[len(data.index) // 2]

    # NOTE:
    # time slicing support is currently disabled
    tests = [
        # {F.VARNAME: var1, F.START: date, F.TESTS: "flagAll()", F.PLOT: True},
        {F.VARNAME: var2, F.TESTS: "flagAll()", F.PLOT: False},
        # {F.VARNAME: var3, F.END: date, F.TESTS: "flagAll()"},
        {F.VARNAME: var3, F.TESTS: "flagAll()",},
    ]

    defaults = {
        F.START: data.index.min(),
        F.END: data.index.max(),
        F.PLOT: False,
        F.LINENUMBER: 2,
    }

    for i, test in enumerate(tests):
        _, meta_frame = initMetaDict([test], data)
        result = dict(zip(meta_frame.columns, meta_frame.iloc[0]))
        expected = {**defaults, **test}
        assert result == expected


def test_variableWildcards(data):

    tests = [
        ("*", ".*"),
        (".*", ".*"),
        ("var(1|2)", "var(1|2)"),
        ("(.*3)", "(.*3)")
    ]

    for config_wc, expected_wc in tests:
        _, config = initMetaDict(
            [{F.VARNAME: config_wc, F.TESTS: "flagAll()"}],
            data
        )
        expected = data.columns[data.columns.str.match(expected_wc)]
        assert np.all(config[F.VARNAME] == expected)


def test_inlineComments(data):
    """
    adresses issue #3
    """
    config = f"""
    {F.VARNAME}|{F.TESTS}|{F.PLOT}
    pre2|flagAll() # test|False # test
    """
    _, meta_frame = initMetaString(config, data)
    assert meta_frame.loc[0, F.PLOT] == False
    assert meta_frame.loc[0, F.TESTS] == "flagAll()"


def test_configReaderLineNumbers(data):
    config = f"""
    {F.VARNAME}|{F.TESTS}
    #temp1|flagAll()
    pre1|flagAll()
    pre2|flagAll()
    SM|flagAll()
    #SM|flagAll()
    SM1|flagAll()
    """
    meta_fname, meta_frame = initMetaString(config, data)
    result = meta_frame[F.LINENUMBER].tolist()
    expected = [3, 4, 5, 7]
    assert result == expected


@pytest.mark.parametrize("flagger", TESTFLAGGER)
@pytest.mark.parametrize("nodata", TESTNODATA)
def test_configChecks(data, flagger, nodata, caplog):

    flagger = flagger.initFlags(data)
    flags = flagger.getFlags()
    var1, var2, var3, *_ = data.columns

    tests = [
        ({F.VARNAME: var1, F.TESTS: "range(mn=0)"}, TypeError),
        ({F.VARNAME: var3, F.TESTS: "flagNothing()"}, NameError),
        ({F.VARNAME: "", F.TESTS: "range(min=3)"}, SyntaxError),
        ({F.VARNAME: var1, F.TESTS: ""}, SyntaxError),
        ({F.VARNAME: ""}, SyntaxError),
        ({F.TESTS: "range(min=3)"}, SyntaxError),
    ]

    for config_dict, expected in tests:
        _, config_df = initMetaDict([config_dict], data)
        with pytest.raises(expected):
            checkConfig(config_df, data, flagger, nodata)