Commit ae73de09 authored by M. Huang's avatar M. Huang
Browse files

conftest and verbosity control

parent cd0ab27b
......@@ -140,12 +140,12 @@ versiontag:
PYTEST = python3 -m pytest
TESTLOG = /tmp/fdi-tests.log
OPT = -r P -v -l --pdb -s --show-capture=all #--log-file=$(TESTLOG)
L = INFO #WARNING
OPT = -r P -v -l --pdb -s --show-capture=all --log-level=$(L)
T =
test: test1 test2
test: test1 test2 test5
testpns: test5 test4
testpns: test4
testhttp: test6 test7 test8
......@@ -156,19 +156,19 @@ test2:
$(PYTEST) tests/test_pal.py -k 'not _http' $(T) --cov=fdi/pal $(OPT)
test3:
$(PYTEST) $(OPT) -k 'server' $(T) tests/test_pns.py --cov=fdi/pns
$(PYTEST) $(OPT) -k 'server' $(T) tests/serv/test_pns.py --cov=fdi/pns
test4:
$(PYTEST) $(OPT) -k 'not server' $(T) tests/test_pns.py --cov=fdi/pns
$(PYTEST) $(OPT) -k 'not server' $(T) tests/serv/test_pns.py --cov=fdi/pns
test5:
$(PYTEST) $(OPT) $(T) tests/test_utils.py --cov=fdi/utils
test6:
$(PYTEST) $(OPT) $(T) tests/test_httppool.py
$(PYTEST) $(OPT) $(T) tests/serv/test_httppool.py
test7:
$(PYTEST) $(OPT) $(T) tests/test_httpclientpool.py
$(PYTEST) $(OPT) $(T) tests/serv/test_httpclientpool.py
test8:
$(PYTEST) $(OPT) $(T) tests/test_pal.py -k '_http'
......
......@@ -7,5 +7,5 @@ git add fdi/dataset/*.py fdi/dataset/resources
git add fdi/pns/*.py fdi/pns/resources
git add fdi/pal/*.py fdi/pal/resources
git add fdi/utils/*.py
git add tests/*.py
git add tests/*.py tests/serv/*.py
git add docs/sphinx/index.rst docs/sphinx/usage docs/sphinx/api docs/sphinx/conf.py docs/sphinx/Makefile docs/sphinx/_static docs/sphinx/_templates
fdi.httppool package
====================
.. automodule:: fdi.httppool
:members:
:undoc-members:
:show-inheritance:
from fdi.dataset.product import _Model_Spec as PPI
from .product import Product
from .numericparameter import NumericParameter
from .stringparameter import StringParameter
from .datatypes import Vector
from .dataset import CompositeDataset, TableDataset
from .arraydataset import ArrayDataset, Column
from ..pal.context import Context, MapContext
from .finetime import FineTime
......@@ -52,3 +57,42 @@ class SP(Product):
assert PPI['metadata']['version']['data_type'] == 'string'
super().__init__(zInfo=zInfo, **metasToBeInstalled, **kwds)
# super().installMetas(metasToBeInstalled)
def get_sample_product():
"""
"""
compo = CompositeDataset()
# two arraydsets
a1 = [768, 4.4, 5.4E3]
a2 = 'ev'
a3 = 'arraydset 1'
a4 = ArrayDataset(data=a1, unit=a2, description=a3)
a5, a6, a7 = [[1.09, 289], [3455, 564]
], 'count', 'background -- arraydset in compo'
a8 = ArrayDataset(data=a5, unit=a6, description=a7)
a10 = 'calibration_arraydset'
compo.set(a10, a8)
# a tabledataset
ELECTRON_VOLTS = 'eV'
SECONDS = 'sec'
t = [x * 1.0 for x in range(5)]
e = [2 * x + 100 for x in t]
x = TableDataset(description="Example table")
x["Time"] = Column(data=t, unit=SECONDS)
x["Energy"] = Column(data=e, unit=ELECTRON_VOLTS)
# set a tabledataset ans an arraydset, with a parameter in metadata
a13 = 'energy_table'
# metadata to the dataset
compo[a13] = x
a11 = 'm1'
a12 = StringParameter('EX')
compo.meta[a11] = a12
prodx = Product('complex prod')
prodx.meta['extra'] = NumericParameter(description='a different param in metadata',
value=Vector((1.1, 2.2, 3.3)), valid={(1, 22): 'normal', (30, 33): 'fast'}, unit='meter')
prodx[a3] = a4
prodx['results'] = compo
return prodx
......@@ -23,7 +23,7 @@ DEFAULT_POOL = 'pool_' + getpass.getuser()
def remoteRegister(p, poolurl):
logger.info('Register %s on the server', poolurl)
logger.debug('Register %s on the server', poolurl)
try:
res, msg = put_on_server('urn:::0', poolurl, 'pool')
except ConnectionError as e:
......@@ -39,7 +39,7 @@ def remoteUnregister(poolurl):
if not poolurl.lower().startswith('http'):
logger.warning('Ignored: %s not for a remote pool.' % poolurl)
return 1
logger.info('unregister %s on the server', poolurl)
logger.debug('unregister %s on the server', poolurl)
#url = api_baseurl + post_poolid
#x = requests.delete(url, auth=HTTPBasicAuth(auth_user, auth_pass))
#o = deserialize(x.text)
......
# -*- coding: utf-8 -*-
from fdi.pal.poolmanager import PoolManager
from fdi.pns.jsonio import getJsonObj, postJsonObj, putJsonObj, commonheaders
from fdi.utils.common import lls
import pytest
import importlib
import base64
import copy
from urllib.error import HTTPError
import os
import logging
logger = logging.getLogger(__name__)
@pytest.fixture(scope='package')
def clean_board():
importlib.invalidate_caches()
# importlib.reload(Classes)
from fdi.dataset.classes import Classes
global Class_Look_Up
# importlib.reload(Class_Look_Up)
from fdi.dataset.deserialize import Class_Look_Up
Classes.updateMapping()
return Classes
@pytest.fixture(scope="package")
def getConfig(clean_board):
from fdi.utils.getconfig import getConfig as getc
return getc
@pytest.fixture(scope="package")
def pc(getConfig):
""" get configuration.
"""
return getConfig()
def checkserver(aburl):
""" make sure the server is running when tests start
"""
# check if data already exists
try:
o = getJsonObj(aburl)
assert o is not None, 'Cannot connect to the server'
logger.info('%s initial server response %s' % (aburl, lls(o, 100)))
except HTTPError as e:
if e.code == 308:
logger.info('%s alive. initial server response 308' % (aburl))
else:
raise
# assert 'result' is not None, 'please start the server to refresh.'
# initialize test data.
@pytest.fixture(scope="package")
def setup(pc):
""" Prepares server absolute base url and common headers fr clients to use.
Based on ``PoolManager.PlacePaths[scheme]`` where ``scheme`` is `http` or `https` and auth info from `pnsconfig` from the configuration file and commandline.
e.g. ```'http://0.0.0.0:5000/v0.7/', ('foo', 'bar')```
return: url has no trailing '/'
"""
testname = 'SVOM'
# client side.
# pool url from a local client
cschm = 'http'
aburl = cschm + '://' + PoolManager.PlacePaths[cschm]
# aburl='http://' + pc['node']['host'] + ':' + \
# str(pc['node']['port']) + pc['baseurl']
checkserver(aburl)
up = bytes((pc['node']['username'] + ':' +
pc['node']['password']).encode('ascii'))
code = base64.b64encode(up).decode("ascii")
headers = copy.copy(commonheaders)
headers.update({'Authorization': 'Basic %s' % (code)})
del up, code
yield aburl, headers
del aburl, headers
@pytest.fixture(scope="package")
def userpass(pc):
auth_user = pc['auth_user']
auth_pass = pc['auth_pass']
return auth_user, auth_pass
@pytest.fixture
def local_pools_dir(pc):
""" this is a path in the local OS, where the server runs.
the path is used to directly access pool server's internals.
return: has no trailing '/'
"""
# http server pool
schm = 'server'
#basepath = pc['server_local_pools_dir']
basepath = PoolManager.PlacePaths[schm]
local_pools_dir = os.path.join(basepath, pc['api_version'])
return local_pools_dir
......@@ -6,63 +6,6 @@ from fdi.utils import getconfig
from fdi.utils.common import lls
import pytest
import copy
import os
import base64
import logging
from urllib.error import HTTPError
logger = logging.getLogger(__name__)
def checkserver(aburl):
""" make sure the server is running when tests start
"""
# check if data already exists
try:
o = getJsonObj(aburl)
assert o is not None, 'Cannot connect to the server'
logger.info('%s initial server response %s' % (aburl, lls(o, 100)))
except HTTPError as e:
if e.code == 308:
logger.info('%s alive. initial server response 308' % (aburl))
else:
raise
# assert 'result' is not None, 'please start the server to refresh.'
# initialize test data.
@pytest.fixture(scope="package")
def setup(pc):
""" Prepares server absolute base url and common headers fr clients to use.
Based on ``PoolManager.PlacePaths[scheme]`` where ``scheme`` is `http` or `https` and auth info from `pnsconfig` from the configuration file and commandline.
e.g. ```'http://0.0.0.0:5000/v0.7/', ('foo', 'bar')```
return: url has no trailing '/'
"""
testname = 'SVOM'
# client side.
# pool url from a local client
cschm = 'http'
aburl = cschm + '://' + PoolManager.PlacePaths[cschm]
# aburl='http://' + pc['node']['host'] + ':' + \
# str(pc['node']['port']) + pc['baseurl']
checkserver(aburl)
up = bytes((pc['node']['username'] + ':' +
pc['node']['password']).encode('ascii'))
code = base64.b64encode(up).decode("ascii")
headers = copy.copy(commonheaders)
headers.update({'Authorization': 'Basic %s' % (code)})
del up, code
yield aburl, headers
del aburl, headers
@pytest.fixture(scope="package")
def userpass(pc):
auth_user = pc['auth_user']
auth_pass = pc['auth_pass']
return auth_user, auth_pass
# -*- coding: utf-8 -*-
import getpass
# different user has different log file to allow multiple user
logfile = '/tmp/fdi_tests_' + getpass.getuser() + '.log'
logdict = {
"version": 1,
"formatters": {
"short": {
"format": "%S %(funcName)s() %(message)s"
},
"full": {
"format": "%(asctime)s %(name)s %(levelname)s %(args)s %(funcName)s():%(lineno)s - %(message)s",
'datefmt': '%Y%m%d %H:%M:%S'
}
},
"handlers": {
"console": {
"level": "INFO",
"class": "logging.StreamHandler",
"formatter": "full",
"stream": "ext://sys.stdout"
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"level": "DEBUG",
"formatter": "full",
"filename": logfile,
"maxBytes": 20000000,
"backupCount": 3
}
},
"loggers": {
"": {
"level": "INFO",
"handlers": ["console", "file"],
"propagate": False
}
},
"root": {
"level": "INFO",
"handlers": ["console", "file"],
},
'disable_existing_loggers': False
}
# -*- coding: utf-8 -*-
# This is to be able to test w/ or w/o installing the package
# https://docs.python-guide.org/writing/structure/
import os
import sys
sys.path.insert(0, os.path.abspath(
os.path.join(os.path.dirname(__file__), '..')))
import fdi
#from fdi import dataset
#from fdi import pal
#from fdi import pns
#from fdi import tests
# print(dir(dataset))
......@@ -28,11 +28,11 @@ def setuplogging():
return logging
logger = logging.getLogger(__name__)
logging = setuplogging()
logger = logging.getLogger()
logger.setLevel(logging.INFO)
setuplogging()
logger.debug('logging level %d' % (logger.getEffectiveLevel()))
......
......@@ -5,8 +5,7 @@
#################
from .test_utils import get_sample_product
import pytest
from fdi.dataset.testproducts import get_sample_product
from fdi.dataset.serializable import serialize
from fdi.dataset.deserialize import deserialize
from fdi.dataset.product import Product
......@@ -23,6 +22,7 @@ from requests.auth import HTTPBasicAuth
import requests
import random
import os
import pytest
from pprint import pprint
import time
......@@ -195,7 +195,7 @@ def populate_server(poolid, aburl, auth):
x.creator = i
data = serialize(x)
url = aburl + '/' + poolid + '/' + prodt + '/' + str(index)
print(len(data))
# print(len(data))
x = requests.post(url, auth=HTTPBasicAuth(*auth), data=data)
o = deserialize(x.text)
......@@ -386,7 +386,7 @@ def test_product_path(setup, userpass):
o = deserialize(x.text)
check_response(o)
c = o['result']
pprint(c)
# pprint(c)
assert 'metadata' in c
# test product paths
......@@ -444,7 +444,7 @@ def test_product_path(setup, userpass):
x = requests.get(url, auth=auth)
assert x.headers['Content-Type'] == 'text/plain'
c = x.text
print(c)
# print(c)
assert 'UNKNOWN' in c
......
......@@ -9,6 +9,7 @@ from fdi.dataset.arraydataset import Column, ArrayDataset
from fdi.dataset.datatypes import Vector, Quaternion
from fdi.dataset.deserialize import Class_Look_Up
from fdi.dataset.product import Product
from fdi.dataset.testproducts import get_sample_product
from fdi.pal.urn import Urn
from fdi.utils.checkjson import checkjson
from fdi.utils.loadfiles import loadcsv
......@@ -66,45 +67,6 @@ def checkgeneral(v):
assert false
def get_sample_product():
"""
"""
compo = CompositeDataset()
# two arraydsets
a1 = [768, 4.4, 5.4E3]
a2 = 'ev'
a3 = 'arraydset 1'
a4 = ArrayDataset(data=a1, unit=a2, description=a3)
a5, a6, a7 = [[1.09, 289], [3455, 564]
], 'count', 'background -- arraydset in compo'
a8 = ArrayDataset(data=a5, unit=a6, description=a7)
a10 = 'calibration_arraydset'
compo.set(a10, a8)
# a tabledataset
ELECTRON_VOLTS = 'eV'
SECONDS = 'sec'
t = [x * 1.0 for x in range(5)]
e = [2 * x + 100 for x in t]
x = TableDataset(description="Example table")
x["Time"] = Column(data=t, unit=SECONDS)
x["Energy"] = Column(data=e, unit=ELECTRON_VOLTS)
# set a tabledataset ans an arraydset, with a parameter in metadata
a13 = 'energy_table'
# metadata to the dataset
compo[a13] = x
a11 = 'm1'
a12 = StringParameter('EX')
compo.meta[a11] = a12
prodx = Product('complex prod')
prodx.meta['extra'] = NumericParameter(description='a different param in metadata',
value=Vector((1.1, 2.2, 3.3)), valid={(1, 22): 'normal', (30, 33): 'fast'}, unit='meter')
prodx[a3] = a4
prodx['results'] = compo
return prodx
def test_fetch():
# simple nested structure
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment