Commit 5cf90cef authored by julian.gethmann's avatar julian.gethmann

Create tests that are not depending on network using pytest and mocking

parent 8113ce50
......@@ -7,6 +7,16 @@ Changelog
------------------
* Delete not implemented argument `show_url` of function `dump_cassandra_data`
0.7.2 (2018-)
------------------
* remove not implemented argument `show_url` of `dump_cassandra_data()`
* write tests that are independent of access to the IBPT network
0.7.2 (2018-)
------------------
* remove not implemented argument `show_url` of `dump_cassandra_data()`
* write tests that are independent of access to the IBPT network
0.7.1 (2017-10-31)
------------------
* Use `PyScaffold` structure
......
......@@ -19,6 +19,11 @@ from sys import version_info
__all__ = ["Cassandra", "CassandraHelper", "Pvs"]
try:
from urllib.error import URLError
except ImportError:
from urllib2 import URLError
class Pvs(object):
"""Provide shortcuts for some PV names.
......@@ -279,7 +284,7 @@ class Cassandra(object):
Raises:
URLError: in case the connection times out (e.g. if one has not
connected to the ANKA-LAN)
connected to the IBPT-LAN)
Examples:
>>> from datetime import datetime
......@@ -347,9 +352,10 @@ class Cassandra(object):
count=1)
tmp.timeout = 1
tmp._download_cassandra_data()
ok = True
except TimeoutError:
except (TimeoutError, URLError):
ok = False
else:
ok = True
return ok
def gen_url(self):
......@@ -389,31 +395,48 @@ class Cassandra(object):
_timeout = 5
if CassandraHelper.PY3:
from urllib import request, error
from urllib import request
for err_count in range(self.RETRIES):
try:
with request.urlopen(url, timeout=_timeout) as response:
json_data = response.read().decode("utf8")
break
except error.URLError:
json_data = json.loads(response.read().decode("utf8"))
<<<<<<< Updated upstream
except URLError:
pass
else:
raise error.URLError(
"Request had a timeout. Maybe you're not inside the IBPT-CN-LAN or provided a wrong PV name"
=======
unset = False
except URLError:
err_count += 1
unset = True
if unset:
>>>>>>> Stashed changes
raise URLError(
"Request had a timeout. Maybe you're not inside the ANKA-LAN"
)
elif CassandraHelper.PY2:
import urllib2 as request
for err_count in range(self.RETRIES):
try:
fobj = request.urlopen(url, timeout=_timeout)
<<<<<<< Updated upstream
json_data = fobj.read()
fobj.close()
break
except request.URLError:
except URLError:
pass
else:
raise error.URLError(
raise URLError(
"Request had a timeout. Maybe you're not inside the IBPT-CN-LAN or provided a wrong PV name"
=======
unset = False
except URLError:
err_count += 1
unset = True
if unset:
raise URLError(
"Request had a timeout. Maybe you're not inside the ANKA-LAN"
>>>>>>> Stashed changes
)
else:
raise NotImplementedError(
......@@ -422,7 +445,10 @@ class Cassandra(object):
return json.loads(json_data)
def dump_cassandra_data(self):
<<<<<<< Updated upstream
# type: (bool) -> str
=======
>>>>>>> Stashed changes
"""Dump the JSON file to a file named like the PV and time that is returned
Dump a JSON file fetched from the Cassandra `host` and return its name.
......@@ -431,7 +457,7 @@ class Cassandra(object):
absolute path and filename of the JSON file.
.. versionchanged:: 0.7.2
Deleted not implemented argument `show_url`
Remove argument `show_url` since it was not implemented either.
"""
json_data = self._download_cassandra_data()
with open(self.json_file, "w") as fobj:
......
......@@ -6,16 +6,20 @@
:Contact: phd@gethmann.org
:Date: 2016-04-12
:Version: 0.1
Import CSS exported data and some helper functions to deal with these kind of data.
"""
from datetime import datetime
import numpy as np
from .cassandra import CassandraHelper
try:
import pandas as pd
except ImportError:
pandas = False
from .cassandra import CassandraHelper
from datetime import datetime
def _header(filename):
""" Return ...
......@@ -46,8 +50,11 @@ def _header(filename):
def _comma(obj):
# type: (str) -> float
""" Convert object/string with `,` as decimal marker to float
If the value is a string, but not convertible to a float, `np.nan` is returned.
Examples:
>>> _comma("3,14")
3.14
......
......@@ -6,14 +6,22 @@
.. versionadded:: 0.5.5
.. versionchanged:: 0.7.0
"""
from datetime import datetime
from typing import Iterable, List, Optional, Tuple # flake8: noqa
import pandas as pd
from .cassandra import Cassandra
from .cassandra import Pvs
from .cassandra import Cassandra, Pvs
def pvs2pd(start, end, pv_names, count=None, upsample=None, save_local=False):
# type: (datetime.datetime, datetime.datetime, Iterable[str],
def pvs2pd(
start, # type: datetime.datetime
end, # type: datetime.datetime
pv_names, # type: Iterable[str]
count=None, # type: Optional[int]
upsample=None, # type: Optional[str]
save_local=False, # type: Optional[bool]
): # type (...) -> pd.DataFrame
# Optional[int], Optional[str], Optional[bool]) -> pd.DataFrame
"""Return a `pd.DataFrame` with data for all `pv_names` and `time` as index
......@@ -63,11 +71,12 @@ def pvs2pd(start, end, pv_names, count=None, upsample=None, save_local=False):
pv = Pvs.pv[pv_name] if ":" not in pv_name else pv_name
if save_local:
cas = Cassandra(start, end, pv=pv, count=None, directory=".")
cas.get_json_local()
data = cas.get_json_local(
) # type: Tuple[List[datetime.datetime], List[float]]
collected_data = collected_data.join(
pd.DataFrame({
"time": cas[0],
pv_name: cas[1]
"time": data[0],
pv_name: data[1]
}).set_index("time"),
how="outer")
else:
......@@ -82,7 +91,7 @@ def pvs2pd(start, end, pv_names, count=None, upsample=None, save_local=False):
collected_data = collected_data[collected_data.first_valid_index():]
# if start in collected_data.index:
# collected_data = collected_data[start:]
# if end in collected_data.index:
# if end in collectecasd_data.index:
# collected_data = collected_data[:end]
if upsample:
collected_data = collected_data[start:end]
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Dummy conftest.py for cassandra.
If you don't know what this is for, just leave it empty.
Read more about conftest.py under:
https://pytest.org/latest/plugins.html
"""
from __future__ import print_function, absolute_import, division
from __future__ import absolute_import, division, print_function
import datetime
import os
import pathlib
from collections import namedtuple
from typing import NamedTuple # flake8: noqa
import cassandra
import pytest
# mock function to return JSON object
def request_openurl(url, timeout):
return (pathlib.Path(__file__).parent /
"data/A:SR:BBB:01:X:SRAM:PEAKTUNE2_11:45:00_12:13:00_1000.json"
).open("rb")
@pytest.fixture()
def setup_() -> NamedTuple["testcase", ("start", datetime.datetime), (
"end", datetime.datetime), ("pv", str)]:
tc = namedtuple("testcase", ["start", "end", "pv"])
start = datetime.datetime(2017, 11, 2, 10, 10, 10)
end = datetime.datetime(2017, 11, 2, 10, 10, 30)
pv = cassandra.cassandra.Pvs.pv["energy"]
return tc(start, end, pv)
@pytest.fixture()
def setup_dl() -> NamedTuple["testcase", ("start", datetime.datetime), (
"end", datetime.datetime), ("pv", str)]:
tc = namedtuple("testcase", ["start", "end", "pv"])
start = datetime.datetime(2016, 2, 3, 11, 45, 0)
end = datetime.datetime(2016, 2, 3, 12, 13, 0)
pv = "A:SR:BBB:01:X:SRAM:PEAKTUNE2"
return tc(start, end, pv)
@pytest.fixture(scope="function")
def cleandir(tmpdir_factory):
# newpath = tempfile.mkdtemp()
os.chdir(str(tmpdir_factory.getbasetemp()))
return tmpdir_factory.getbasetemp()
@pytest.fixture(scope="function")
def cas(cleandir, setup_):
start, end, pv = setup_
# start = datetime.datetime(2017, 11, 2, 10, 10, 10)
# end = datetime.datetime(2017, 11, 2, 10, 10, 30)
# pv = cassandra.cassandra.Pvs.pv["energy"]
cas_request = cassandra.cassandra.Cassandra(
start, end, pv, count=None, directory=str(cleandir))
yield cas_request
try:
os.unlink(cas_request.json_file)
except FileNotFoundError:
pass
This source diff could not be displayed because it is too large. You can view the blob instead.
[{"time": 1470414515020430694, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8119735610669162]}, {"time": 1470414516012211909, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8117395882077679]}, {"time": 1470414517003964911, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8113155124005615]}, {"time": 1470414517995720989, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8117395882077679]}, {"time": 1470414519029224112, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8113740056153487]}, {"time": 1470414520020976751, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8121636640149743]}, {"time": 1470414521012501370, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8113740056153487]}, {"time": 1470414522004172942, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8121051708001872]}, {"time": 1470414522995713209, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8113008890968648]}, {"time": 1470414524032108812, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8121490407112775]}, {"time": 1470414524968108447, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8119150678521292]}, {"time": 1470414526259853197, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8118127047262518]}, {"time": 1470414527252712083, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.811476368741226]}, {"time": 1470414528249702781, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8117395882077679]}, {"time": 1470414529277671847, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.811417875526439]}, {"time": 1470414530569890081, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8118711979410388]}, {"time": 1470414531562054724, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.8114617454375292]}]
[{"time": 1470414515564201945, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.724116752456715]}, {"time": 1470414516866162996, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7256375760411792]}, {"time": 1470414517868070829, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7250818905007019]}, {"time": 1470414518894333788, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7256960692559663]}, {"time": 1470414519828084092, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7242776087973795]}, {"time": 1470414521129962037, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7245408282639214]}, {"time": 1470414522132202735, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7251550070191858]}, {"time": 1470414524156689221, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7250818905007019]}, {"time": 1470414525372210847, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.724306855404773]}, {"time": 1470414526674425224, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7251550070191858]}, {"time": 1470414527676421120, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7251257604117922]}, {"time": 1470414528702690893, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7251988769302761]}, {"time": 1470414529636880805, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7251403837154891]}, {"time": 1470414530939050455, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7242044922788956]}, {"time": 1470414531940954451, "severity": {"level": "OK", "hasValue": true}, "status": "NO_ALARM", "quality": "Original", "metaData": {"type": "numeric", "precision": 5, "units": "", "displayLow": 0.0, "displayHigh": 0.0, "warnLow": "NaN", "warnHigh": "NaN", "alarmLow": "NaN", "alarmHigh": "NaN"}, "type": "double", "value": [0.7252135002339729]}]
......@@ -3,29 +3,122 @@
# :title: Unittests for calc_phasespaces.py
# :lastupdate: 2016-09-15
# :lastupdate: 2017-11-01
import unittest
import doctest
import datetime
import json
import os
import pathlib
from urllib import error, request
from urllib.error import URLError
from cassandra.cassandra import *
import cassandra
import pytest
class CassandraTestCase(unittest.TestCase):
# from cassandra.cassandra import *
from .conftest import request_openurl
def setUp(self):
pass
def tearDown(self):
# os.unlink(self.tmpfile)
pass
@pytest.mark.usefixtures("cleandir", "setup_")
class TestCassandraClass(object):
def test_init(self, cas, setup_):
start, end, pv = setup_
if __name__ == "__main__":
# unittest.main()
testSuite = unittest.TestSuite()
testSuite.addTest(unittest.makeSuite(CassandraTestCase))
testSuite.addTest(doctest.DocTestSuite(cassandra.cassandra))
unittest.TextTestRunner(verbosity=2).run(testSuite)
cas2 = cassandra.cassandra.Cassandra(start, end, pv)
assert cas2.pv == pv
assert cas2.count is None
assert cas2.start_time == start
assert cas2.end_time == end
assert pathlib.Path(cas2.directory).absolute() == pathlib.Path.cwd()
assert cas2.json_filename == "A:SR:BeamInfo:01:Energy_10:10:10_10:10:30_None.json"
def test_gen_url(self, cas, setup_):
url = cas.gen_url()
start, end, pv = setup_
assert cas.pv == pv
assert (
url ==
"http://ankasr-archiver.anka.kit.edu:9812/archive-access/api/1.0/archive/1/samples/A:SR:BeamInfo:01:Energy?start=1509613810000000000&end=1509613830000000000"
)
cas2 = cassandra.cassandra.Cassandra(
start, end, pv, directory="/tmp", count=1000)
url = cas2.gen_url()
assert (
url ==
"http://ankasr-archiver.anka.kit.edu:9812/archive-access/api/1.0/archive/1/samples/A:SR:BeamInfo:01:Energy?start=1509613810000000000&end=1509613830000000000&count=1000"
)
def test_check_connection(
self,
cas,
monkeypatch,
):
def off(url, timeout):
raise URLError(
"Request had a timeout. Maybe you're not inside the IBPT-CN-LAN or provided a wrong PV name"
)
monkeypatch.setattr(request, "urlopen", off)
assert not cas.check_connection()
def off(url, timeout):
raise TimeoutError(
"Request had a timeout. Maybe you're not inside the IBPT-CN-LAN or provided a wrong PV name"
)
monkeypatch.setattr(request, "urlopen", off)
assert not cas.check_connection()
monkeypatch.setattr(request, "urlopen", request_openurl)
assert cas.check_connection()
def test_download_cassandra_data_online(self, cas, monkeypatch):
with open(
os.path.dirname(os.path.abspath(__file__)) +
"/data/A:SR:BBB:01:X:SRAM:PEAKTUNE2_11:45:00_12:13:00_1000.json",
"r") as fobj:
json_obj = json.loads(fobj.read())
monkeypatch.setattr(request, "urlopen", request_openurl)
assert cas._download_cassandra_data() == json_obj
# TODO: mock failing several times, but being ok in the end.
def test_download_cassandra_data_offline(self, cas, monkeypatch):
def offline(url, timeout):
raise error.URLError("Name or service not known")
monkeypatch.setattr(request, "urlopen", offline)
with pytest.raises(error.URLError, match=r".*[Tt]imeout.*"):
cas._download_cassandra_data()
def test_dump_cassandra_data(self, cas, cleandir, monkeypatch):
assert not pathlib.Path(cleandir / cas.json_filename).is_file()
monkeypatch.setattr(request, "urlopen", request_openurl)
assert str(cleandir / cas.json_filename) == cas.dump_cassandra_data()
assert pathlib.Path(cleandir / cas.json_filename).is_file()
with open(str(cleandir / cas.json_filename), "r") as got, open(
os.path.dirname(os.path.abspath(__file__)) +
"/data/A:SR:BBB:01:X:SRAM:PEAKTUNE2_11:45:00_12:13:00_1000.json",
"r") as expected:
assert json.load(expected) == json.load(got)
def test_with(self, setup_dl, monkeypatch):
monkeypatch.setattr(request, "urlopen", request_openurl)
start, end, pv = setup_dl
with Cassandra(*setup_dl) as cas2:
assert abs(start - cas2[0][0]) < datetime.timedelta(seconds=2)
assert abs(end - cas2[0][-1]) < datetime.timedelta(seconds=2)
assert isinstance(cas2[0][0], datetime.datetime)
assert isinstance(cas2[1][0], float)
if __name__ == "__main__":
pass
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
......@@ -3,15 +3,15 @@ import doctest
import os
import unittest
from datetime import datetime
from tempfile import mktemp
import cassandra
import numpy as np
import pandas as pd
import cassandra
class CssreadTestSuite(unittest.TestCase):
tmpfile = "/tmp/uiae"
class CssreadTestCase(unittest.TestCase):
tmpfile = mktemp()
header = """# Created by CSS Data Browser Version 3.2.17.20140409
#
# Start Time : 2016/02/15 11:17:56.237000000
......@@ -78,7 +78,9 @@ class CssreadTestSuite(unittest.TestCase):
2016/02/15 11:30:00.000000000 0,0422 0,0000 8,5046 8,2804 8,3764 2,8566 2,8826 0,0027
2016/02/15 11:35:00.000000000 0,0420 0,0000 8,5042 8,2800 8,3759 2,8568 2,8827 0,0027
2016/02/15 11:40:00.000000000 0,0420 0,0000 8,5040 8,2799 8,3759 2,8566 2,8826 0,0027"""
header_names = ["Time", "A:SR:BeamInfo:01:Current Value",
header_names = [
"Time",
"A:SR:BeamInfo:01:Current Value",
"A:SR:BeamInfo:01:Energy Value",
"A:ID-S1:Wig:01:JB:Temp:27 Value",
"A:ID-S1:Wig:01:JB:Temp:28 Value",
......@@ -99,12 +101,25 @@ class CssreadTestSuite(unittest.TestCase):
self.test__header()
self.test_load_css_data()
def test__comma(self):
self.assertEqual(1.3, cassandra.css_read._comma("1.3"))
self.assertEqual(1.3, cassandra.css_read._comma("1,3"))
with self.assertRaises(AttributeError):
cassandra.css_read._comma(1.3)
# pytest, because of the nans
np.testing.assert_equal(np.nan, cassandra.css_read._comma("1NOPE3"))
def test__header(self):
res = cassandra.css_read._header(self.tmpfile)
self.assertIs(len(res), 2)
self.assertIs(res[1], len(self.header.split("\n")))
self.assertListEqual(res[0], self.header_names)
with self.assertRaises(ValueError):
cassandra.css_read._header("/dev/null")
with self.assertRaises(PermissionError):
cassandra.css_read._header("/etc/shadow")
def test_load_css_data(self):
res_np = cassandra.css_read.load_css_data(self.tmpfile)
......@@ -112,22 +127,32 @@ class CssreadTestSuite(unittest.TestCase):
self.assertIsInstance(res_np, np.ndarray)
self.assertIsInstance(res_pd, pd.DataFrame)
self.assertEqual(res_np["Time"][0], datetime(2016, 2, 15, 11, 15))
self.assertEqual(
res_pd.index[0].to_pydatetime(), datetime(2016, 2, 15, 11, 15))
self.assertEqual(res_pd.index[0].to_pydatetime(),
datetime(2016, 2, 15, 11, 15))
self.assertIsInstance(res_pd.index[0], pd.tslib.Timestamp)
res_pd_nodrop = cassandra.css_read.load_css_data(
self.tmpfile, "pd", dropna=False)
self.assertIsInstance(res_pd_nodrop, pd.DataFrame)
self.assertEqual(res_pd_nodrop.dropna(
).index[0].to_pydatetime(), datetime(2016, 2, 15, 11, 15))
self.assertEqual(
res_pd_nodrop.index[0].to_pydatetime(), datetime(2016, 2, 12, 16, 55))
self.assertEqual(res_pd_nodrop.dropna().index[0].to_pydatetime(),
datetime(2016, 2, 15, 11, 15))
self.assertEqual(res_pd_nodrop.index[0].to_pydatetime(),
datetime(2016, 2, 12, 16, 55))
self.assertIsInstance(res_pd_nodrop.index[0], pd.tslib.Timestamp)
with self.assertRaises(ValueError):
cassandra.css_read.load_css_data(
self.tmpfile,
"NONO",
)
if __name__ == '__main__':
def test_unittest():
testSuite = unittest.TestSuite()
testSuite.addTest(unittest.makeSuite(CssreadTestCase))
testSuite.addTest(doctest.DocTestSuite(cassandra.css_read))
unittest.TextTestRunner(verbosity=2).run(testSuite)
if __name__ == '__main__':
test_unittest()
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# :title: Unittests for pd.py
# :lastupdate: 2017-09-15
import datetime
import doctest
import unittest
import pandas as pd
import cassandra
from cassandra.pd import pvs2pd
class CassandraPdTestCase(unittest.TestCase):
def setUp(self):
self.start = datetime.datetime(2017, 9, 15, 13, 0, 0)
self.end = datetime.datetime(2017, 9, 15, 17, 52, 0)
self.pvs = ["nu_x", "rf", "current"]
def tearDown(self):
# os.unlink(self.tmpfile)
pass
def test_pvs2pd(self):
res = pvs2pd(self.start, self.end, self.pvs)
# test if the result stays within dates
self.assertLessEqual(res.index[-1], self.end)
self.assertGreaterEqual(res.index[0], self.start)
# test if the result does crop to much
res = pvs2pd(datetime.datetime(2017, 9, 8, 8),
datetime.datetime(2017, 9, 8, 8, 1), ["fill", "energy"])
# fill energy
# time
# 2017-09-07 08:55:35.836193 6447.0 2.499602
# 2017-09-08 08:08:04.635904 6447.0 2.526181
# 2017-09-08 08:08:17.853150 6448.0 2.526181
self.assertGreaterEqual(len(res), 2)
self.assertEqual(res.iloc[0]["fill"], 6447)
self.assertAlmostEqual(res.iloc[0]["energy"], 2.499602, 5)
# test upsample
res = pvs2pd(datetime.datetime(2017, 9, 8, 8),
datetime.datetime(2017, 9, 8, 8, 1), ["fill", "energy"],
upsample="ms")
self.assertEqual(len(res), 60 * 1000 + 1)
self.assertEqual(res.index[4], pd.Timestamp(
'2017-09-08 08:00:00.004000', freq='L'))
with self.assertRaises(ValueError):
pvs2pd(datetime.datetime(2017, 9, 8, 8),
datetime.datetime(2017, 9, 8, 8, 1), ["fill", "energy"],
upsample="This is wrong")
if __name__ == "__main__":
# unittest.main()
testSuite = unittest.TestSuite()
testSuite.addTest(unittest.makeSuite(CassandraPdTestCase))
testSuite.addTest(doctest.DocTestSuite(cassandra.pd))
unittest.TextTestRunner(verbosity=2).run(testSuite)
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Unit tests for cassandra.pd
"""
# :title: Unittests for calc_phasespaces.py
# :lastupdate: 2016-09-15
import datetime
import doctest
import json
import os
import pathlib
import unittest
from collections import namedtuple
from typing import List, NamedTuple
from urllib import request
import cassandra
import pytest
from cassandra.pd import pvs2pd
class CassandraTestCase(unittest.TestCase):
class TestCassandraPD(object):
def setUp(self):
self.pv_list = ["nu_x", "nu_y"]