test_client.py 6.95 KB
Newer Older
1
2
3
4
5
import urllib.request
from urllib.error import HTTPError, URLError

import pytest

6
import astropy.units as u
7
8
from sunpy.net import attrs as a

9
from sdc.client import KISClient
10
11
12
13
14
15
16
17
18
19
20
21
22


_BASE_URL = "http://dockertest:8083/sdc/"

try:
    res = urllib.request.urlopen(_BASE_URL)
    HAS_DOCKERTEST = True
except(HTTPError, URLError):
    HAS_DOCKERTEST = False


@pytest.fixture
def client():
23
    return KISClient()
24
25
26


def test_search(client):
27
    """Test conversion of (supported) Attrs to query string."""
28

29
    assert not client._can_handle_query(a.Time("2019/01/01", "2021/01/01"))
30
31
32
33
    with pytest.raises(AttributeError, match=r"Query not possible: "
                       r"No 'Instrument' found in Attributes"):
        client.search(a.Time("2019/01/01", "2021/01/01"))

34
    assert not client._can_handle_query(a.Instrument("UVES"), a.Time("2019/01/01", "2021/01/01"))
35
36
37
38
    with pytest.raises(AttributeError, match=r"Query not possible: "
                       r"Instrument UVES not in registered list"):
        client.search(a.Instrument("UVES") & a.Time("2019/01/01", "2021/01/01"))

39
    query = a.Instrument("BBI") & a.Time("2019/01/01", "2021/01/01")
40
    # TODO: Verify the actual return from the RESTHeart server.
41
    assert client._can_handle_query(query)
42
    if HAS_DOCKERTEST:
43
        res = client.search(query)
44
45
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
46
47
48
                           r".http://dockertest:8083/sdc/bbi_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'bbi'},"
                           r"{'description.DATE_BEG':{'.le':{'2021-01-01T00:00:00.000'}}},"
49
50
                           r"{'description.DATE_END':{'.ge':{'2019-01-01T00:00:00.000'}}}.}"
                           rf".*Confirm that RESTHeart is running on {_BASE_URL} and connected"):
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
            client.search(query)

    query = a.Instrument("LARS") & a.sdc.HelioProjLat(-10*u.arcsec, 0.2*u.arcmin)
    assert client._can_handle_query(query)
    if HAS_DOCKERTEST:
        res = client.search(query)
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/lars_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'lars'},"
                           r"{'description.HPLT_TAN_MIN':{'.le':{12\}}},"
                           r"{'description.HPLT_TAN_MAX':{'.ge':{-10\}}}"):
            client.search(query)

    query = a.Instrument("Zimpol") & (a.sdc.Theta(85*u.deg, 600*u.arcmin) | a.sdc.PolStates('iq'))
    assert client._can_handle_query(query)
    if HAS_DOCKERTEST:
        res = client.search(query)
    else:
Derek Homeier's avatar
Derek Homeier committed
70
        with pytest.raises(URLError, match=r"Unable to execute search ") as exc:
71
72
            client.search(query)
        # Will raise on first of multi-part OR queries.
Derek Homeier's avatar
Derek Homeier committed
73
        assert "http://dockertest:8083/sdc/zimpol_observations?filter=" in str(exc.value)
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
        assert "{'description.THETA':{'$ge':{10},'$le':{85}}}" in str(exc.value)
        assert "{'description.POL_STATES':" not in str(exc.value)


@pytest.mark.parametrize("query", ((a.Instrument("GRIS") & a.Level(3)),
                                   (a.Instrument("ChroTel") & a.Physobs("perspective.vortex")),
                                   (a.Level(0) & a.Instrument("Bob")),
                                   (a.Instrument("LARS") & a.sdc.Telescope("Leviathan"))))
def test_cant_handle_query(client, query):
    """Some examples of invalid queries with exceptions."""
    assert not client._can_handle_query(*query.attrs)
    with pytest.raises(AttributeError, match=r"Query not possible: "
                       rf"[ILPT][a-z]* {query.attrs[1].value} not in [rs]"):
        client.search(query)


@pytest.mark.parametrize("query", (a.Level(1), a.Wavelength(3200*u.AA, 1.6*u.micron),
                                   a.sdc.DataProduct('cube'), a.sdc.ObsName('gris_211031_12'),
                                   a.sdc.Date('2021/10/31'), a.sdc.Filter('I'),
                                   a.sdc.PolStates('IQUV'), a.sdc.Telescope('Gregor'),
                                   a.sdc.Target('Sunspot_22'), a.sdc.AtmosR0(*([10, 200]*u.cm)),
                                   a.sdc.Theta(20*u.arcmin, 89*u.deg), a.sdc.Mu(0.1, 1),
                                   a.sdc.ExposureTime(*([5, 60]*u.min)),
                                   a.sdc.HelioProjLon(5*u.arcsec), a.sdc.HelioProjLat(9*u.arcsec),
                                   a.sdc.SpatialResolution(0.1*u.arcsec, 0.8*u.arcsec),
                                   a.sdc.SpectralResolution(600, 20000),
                                   a.sdc.TemporalResolution(2*u.s, 30*u.s),
                                   a.sdc.NDimensions(2, 3), a.sdc.PolXel(2, 4),
                                   a.sdc.SpatialXel1(200, 3000), a.sdc.SpatialXel2(100, 4000),
                                   a.sdc.SpectralXel(320, 4096), a.sdc.TimeXel(60, 86400)))
def test_all_queries(client, query):
105
    """Test an example of all supported query attributes with automatic field names."""
106
107
108
109
110
111
112
113
114
    assert client._can_handle_query(a.Instrument("GRIS"), query)
    if HAS_DOCKERTEST:
        res = client.search(a.Instrument("GRIS") & query)
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/gris_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'gris'},"
                           rf"{{'description.*{query.type_name.upper()}"):
            client.search(a.Instrument("GRIS") & query)
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149


def test_full_range(client):
    """
    Test 'fullrange' option - 'FIELD_MIN,_MAX' shall completely include `[Attr.min, Attr.max]`.
    """

    t = a.Time("2019/01/01", "2019/01/09")
    t.fullrange = True
    query = a.Instrument("BBI") & t
    assert client._can_handle_query(query)
    if HAS_DOCKERTEST:
        res = client.search(query)
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/bbi_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'bbi'},"
                           r"{'description.DATE_BEG':{'.le':{'2019-01-01T00:00:00.000'}}},"
                           r"{'description.DATE_END':{'.ge':{'2019-01-09T00:00:00.000'}}}.}"):
            client.search(query)

    # Test with inverted `min`, `max` inputs.
    hplt = a.sdc.HelioProjLat(0.2*u.arcmin, -10*u.arcsec)
    hplt.fullrange = True
    query = a.Instrument("LARS") & hplt
    assert client._can_handle_query(query)
    if HAS_DOCKERTEST:
        res = client.search(query)
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/lars_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'lars'},"
                           r"{'description.HPLT_TAN_MIN':{'.le':{-10\}}},"
                           r"{'description.HPLT_TAN_MAX':{'.ge':{12\}}}"):
            client.search(query)