123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990 |
- #!/usr/bin/env python3
- import pytest
- from starlette.authentication import UnauthenticatedUser
- from starlette.exceptions import HTTPException
- from starlette.testclient import TestClient
- 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, timedelta
-
- from pyheatpump.conf import config
- from pyheatpump.db import initialize, connect
- from pyheatpump.variables import app, get_variables, set_variables, ROUTES
- from pyheatpump.variable_values import app, get_variable_value, set_variable_value, ROUTES
- from pyheatpump.models.variable_value import VariableValue
-
- @pytest.fixture(scope='module')
- 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)
-
-
- def test_get_type_address(set_test_db):
- c = TestClient(app)
- r = c.get('/A/10')
- assert r.status_code == 200
-
- d_resp = r.content.decode()
- assert int(d_resp) == 42
-
- try:
- r = c.get('/X/10')
- except HTTPException as e:
- assert e.status_code == 404
-
- r = c.get('/A/10/{}'.format(
- (datetime.now() - timedelta(days=1)).isoformat()
- ))
- d_resp = r.content.decode()
- assert int(d_resp) == 20
-
- r = c.get('/A/10/{}'.format(
- (datetime.now() - timedelta(days=2)).isoformat()
- ))
- d_resp = r.content.decode()
- assert int(d_resp) == 0
-
- r = c.get('/I/5010')
- d_resp = r.content.decode()
- assert int(d_resp) == 42
-
- r = c.get('/I/5011')
- d_resp = r.content.decode()
- assert int(d_resp) == 24
-
- r = c.get('/I/5010/{}'.format(
- (datetime.now() - timedelta(hours=1)).isoformat()
- ))
- d_resp = r.content.decode()
- assert int(d_resp) == 20
-
- r = c.get('/I/5010/{}'.format(
- (datetime.now() - timedelta(hours=2)).isoformat()
- ))
- d_resp = r.content.decode()
- assert int(d_resp) == 0
-
-
- @pytest.mark.skip
- def test_set_(set_test_db):
- c = TestClient(app)
- r = c.post('/')
- assert r.status_code == 200
|