pyHeatpump/tests/models/test_variable_value.py

#!/usr/bin/env python3
import pytest
from starlette.authentication import UnauthenticatedUser
from starlette.testclient import TestClient
#from pyheatpump.config import app, config, default_config, CONFIG_FILES, get_config, set_config, config_route, ROUTES
from unittest.mock import patch, MagicMock
from pprint import pprint
import json
from tempfile import mkstemp
from configparser import ConfigParser
import os
import sys
from datetime import datetime

from pyheatpump.config import config
from pyheatpump.db import initialize, connect
from pyheatpump.variable_types import app, get_variable_types, set_variable_types, ROUTES
from pyheatpump.models import *


@pytest.fixture()
def set_test_db():
    _, tmpdb = mkstemp(suffix='.db', dir=os.getcwd(), )
    print(f'Will store database in {tmpdb}')
    config['heatpump']['database'] = tmpdb
    if not initialize(os.path.join(os.getcwd(), 'db/pyheatpump.sql')):
        sys.exit(-1)

    if not initialize(os.path.join(os.getcwd(), 'db/test_variables.sql')):
        sys.exit(-1)

    if not initialize(os.path.join(os.getcwd(), 'db/test_variable_values.sql')):
        sys.exit(-1)

    yield


    os.unlink(tmpdb)


@pytest.fixture
def var_types():
    return VariableType.getall()


def test_last_update(set_test_db, var_types):
    for _, var_type in var_types.items():
        for _, variable in var_type.get_variables().items():
            try:
                if not variable.last_update:
                    continue
                time = datetime.fromtimestamp(variable.last_update)

                val = VariableValue.get(variable.type,
                    variable.address,
                    time)
                assert val is not None
                assert variable.last_update == val.time
            except StopIteration:
                assert False


def test_var_value_getval(set_test_db, var_types):
    var_type = var_types['Analog']
    for _, variable in var_type.get_variables().items():
        try:
            if not variable.last_update:
                continue
            time = datetime.fromtimestamp(variable.last_update)

            value = VariableValue.get(variable.type,
                variable.address,
                time).get_value()
            assert type(value) == float
        except StopIteration:
            assert False


    var_type = var_types['Integer']
    for _, variable in var_type.get_variables().items():
        try:
            if not variable.last_update:
                continue
            time = datetime.fromtimestamp(variable.last_update)

            value = VariableValue.get(variable.type,
                variable.address,
                time).get_value()
            assert type(value) == int
        except StopIteration:
            assert False


    var_type = var_types['Digital']
    for _, variable in var_type.get_variables().items():
        try:
            if not variable.last_update:
                continue
            time = datetime.fromtimestamp(variable.last_update)

            value = VariableValue.get(variable.type,
                variable.address,
                time).get_value()
            assert type(value) == bool
        except StopIteration:
            assert False