Commit 490cd72c authored by julian.gethmann's avatar julian.gethmann

Make tests compatible with older Pythons

parent 5934aaa3
......@@ -25,9 +25,9 @@ __copyright__ = "Julian Gethmann"
__license__ = "mit"
__all__ = ["Cassandra", "CassandraHelper", "Pvs"]
try: # Py2
try: # Py3
from urllib.error import URLError
except ImportError: # Py3
except ImportError: # Py2
from urllib2 import URLError
......@@ -217,16 +217,14 @@ class CassandraHelper(object):
except ValueError:
pass
except TypeError:
raise TypeError(
"Time ({}) must either be a `datetime.datetime` or a string"
" as described in the docstring and not a {}.".format(
time_string, type(time_string).__name__))
raise TypeError("Time ({}) must either be a `datetime.datetime` or a string"
" as described in the docstring and not a {}.".format(
time_string, type(time_string).__name__))
if not ret:
raise ValueError(
"`time_string` must not be of the form `{}` and type {}! "
"See docstring for possible forms.".format(
time_string, type(time_string).__name__))
raise ValueError("`time_string` must not be of the form `{}` and type {}! "
"See docstring for possible forms.".format(
time_string, type(time_string).__name__))
else:
return ret
......@@ -270,8 +268,8 @@ class CassandraHelper(object):
values.append(entry["value"])
# if entry["status"] == "OK" or entry["severity"]["level"] == "OK"]
timestamps = [
datetime.datetime.fromtimestamp(entry["time"] / 1e9)
for entry in json_data if entry["severity"]["level"] == "OK"
datetime.datetime.fromtimestamp(entry["time"] / 1e9) for entry in json_data
if entry["severity"]["level"] == "OK"
]
dataset = namedtuple("dataset", ["timestamps", "values"])
......@@ -359,17 +357,14 @@ class Cassandra(object):
@staticmethod
def check_connection():
# type: () -> bool
"""Return if the connection is ok or not
Returns:
bool: True if the connection is ok, False if an error occurred
"""
try:
tmp = Cassandra(
"2017-07-07T08:08:08",
"2017-07-07T08:08:08",
Pvs.pv["energy"],
count=1)
tmp = Cassandra("2017-07-07T08:08:08", "2017-07-07T08:08:08", Pvs.pv["energy"], count=1)
tmp.timeout = 1
tmp._download_cassandra_data()
except (TimeoutError, URLError):
......@@ -379,6 +374,7 @@ class Cassandra(object):
return ok
def gen_url(self):
# type: (...) -> str
"""Return the URL to be fetched from Cassandra.
If count is `None` then the count parameter is not included in the URL
......@@ -425,9 +421,7 @@ class Cassandra(object):
except URLError:
pass
else:
raise URLError(
"Request had a timeout. Maybe you're not inside the ANKA-LAN"
)
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):
......@@ -443,13 +437,12 @@ class Cassandra(object):
"Request had a timeout. Maybe you're not inside the IBPT-CN-LAN or provided a wrong PV name"
)
else:
raise NotImplementedError(
"Not implemented for other versions than 2 or 3!")
raise NotImplementedError("Not implemented for other versions than 2 or 3!")
return json_data
def dump_cassandra_data(self):
# type: (bool) -> str
# type: (...) -> str
"""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.
......@@ -466,6 +459,7 @@ class Cassandra(object):
return path.abspath(self.json_file)
def get_json_local(self):
# type: (...) -> Tuple[List[datetime.datetime], List[Union[List[float], float]]]
""" Return timestamps and values for given Cassandra object and copy JSON file if neccessary."""
if not path.isfile(self.json_file) or path.getsize(self.json_file) <= 2:
self.dump_cassandra_data()
......
......@@ -8,9 +8,6 @@
import datetime
import json
import os
import pathlib
from urllib import error, request
from urllib.error import URLError
import cassandra
import pytest
......@@ -19,6 +16,18 @@ from cassandra.cassandra import Cassandra
# from cassandra.cassandra import *
from .conftest import request_openurl
try:
import pathlib
except ImportError:
pass
try: # Py3
from urllib import request
from urllib.error import URLError
except ImportError: # Py2
from urllib2 import URLError
import urllib2 as request
@pytest.mark.usefixtures("cleandir", "setup_")
class TestCassandraClass(object):
......@@ -32,7 +41,9 @@ class TestCassandraClass(object):
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 os.path.abspath(cas2.directory) == os.path.abspath(os.path.curdir)
# 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_):
......@@ -41,16 +52,15 @@ class TestCassandraClass(object):
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"
)
"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)
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"
)
"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,
......@@ -80,8 +90,7 @@ class TestCassandraClass(object):
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:
"/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)
......@@ -92,22 +101,23 @@ class TestCassandraClass(object):
def test_download_cassandra_data_offline(self, cas, monkeypatch):
def offline(url, timeout):
raise error.URLError("Name or service not known")
raise URLError("Name or service not known")
monkeypatch.setattr(request, "urlopen", offline)
with pytest.raises(error.URLError, match=r".*[Tt]imeout.*"):
with pytest.raises(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()
assert not os.path.isfile(os.path.join(cleandir, cas.json_filename))
# 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()
assert os.path.isfile(os.path.join(cleandir, cas.json_filename))
# 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:
"/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):
......@@ -119,6 +129,7 @@ class TestCassandraClass(object):
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
......@@ -9,16 +9,23 @@ import datetime
import doctest
import json
import os
import pathlib
import sys
import unittest
from collections import namedtuple
from typing import List, NamedTuple
from urllib import request
import cassandra
import pytest
from cassandra.pd import pvs2pd
try:
import pathlib
except ImportError:
pass
try:
from urllib import request
except ImportError:
import urllib2 as request
setup = NamedTuple("setup", [
("start", datetime.datetime),
("end", datetime.datetime),
......@@ -30,26 +37,27 @@ class TestCassandraPD(object):
@pytest.fixture()
def setup_pd(self) -> setup:
"""Return `start` and `end` times and a list of PVs"""
pv_list = ["nu_x", "nu_y"]
start = datetime.datetime(2016, 8, 5, 18, 28, 36)
end = datetime.datetime(2016, 8, 5, 18, 28,
36) + datetime.timedelta(seconds=15)
end = datetime.datetime(2016, 8, 5, 18, 28, 36) + datetime.timedelta(seconds=15)
# setup = namedtuple("setup", ["start", "end", "pv_list"])
return setup(start, end, pv_list)
@staticmethod
def urlopen_nus(url: str, *args, **kwargs):
"""Return a filehandler with the JSON file of a tune readback for X and Y tune"""
if "A:SR:BBB:01:Y:SRAM:PEAKTUNE2" in url:
plane = "Y"
elif "" in url:
plane = "X"
else:
raise ValueError("Mocking failed!")
return (
pathlib.Path(__file__).parent /
f"data/A:SR:BBB:01:{plane}:SRAM:PEAKTUNE2_18:28:36_18:28:51_None.json"
).open("rb")
return (pathlib.Path(__file__).parent /
"data/A:SR:BBB:01:{plane}:SRAM:PEAKTUNE2_18:28:36_18:28:51_None.json".format(
plane=plane)).open("rb")
@pytest.mark.skipif(sys.version_info < (3, 4), reason="requires python3.4")
def test_pvs2pd(self, monkeypatch, setup_pd, cleandir):
monkeypatch.setattr(request, "urlopen", self.urlopen_nus)
ret = pvs2pd(*setup_pd)
......@@ -65,7 +73,10 @@ class TestCassandraPD(object):
assert not pathlib.Path(cleandir / fn).is_file()
pvs2pd(*setup_pd, save_local=True)
with open(str(cleandir / fn), "r") as got, open(
os.path.dirname(os.path.abspath(__file__)) + f"/data/{fn}",
"{abspath}/data/{fn}".format(
abspath=os.path.dirname(os.path.abspath(__file__)),
fn=fn,
),
"r",
) as expected:
assert json.load(expected) == json.load(got)
......
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