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

import pytest

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

10
from sdc.client import KISClient
11
12
13


_BASE_URL = "http://dockertest:8083/sdc/"
14
15
16
17
18
19
_QUERY_BASE = "gris_observations?filter="
_EXAMPLE_QUERY = "{'description.OBS_NAME':'gris_20140426_000'}"
_EXAMPLE_RANGE = "{'description.THETA':{'$gt':70.0,'$lt':80}}"
_EXAMPLE_DATES = ("{'$and':[{'description.INSTRUMENT':'gris'},"
                  "{'description.DATE_BEG':{'$gte':{'$date':'2014-04-26T00:00:00'},"
                  "'$lte':{'$date':'2014-04-27T00:00:00'}}}]}")
20
21

try:
22
    response = urllib.request.urlopen(f"{_BASE_URL}{_QUERY_BASE}{_EXAMPLE_QUERY}")
23
24
25
26
27
28
29
    HAS_DOCKERTEST = True
except(HTTPError, URLError):
    HAS_DOCKERTEST = False


@pytest.fixture
def client():
30
    return KISClient()
31
32


33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def test_docker(client):
    """Test example queries on dockertest."""
    if not HAS_DOCKERTEST:
        pytest.xfail("No dockertest running")

    response = urllib.request.urlopen(f"{_BASE_URL}{_QUERY_BASE}{_EXAMPLE_QUERY}")
    data = json.loads(response.read())
    assert '_embedded' in data.keys()
    assert 'description' in data['_embedded'][0]

    response = urllib.request.urlopen(f"{_BASE_URL}{_QUERY_BASE}{_EXAMPLE_DATES}")
    data = json.loads(response.read())
    assert 'description' in data['_embedded'][0]

    response = urllib.request.urlopen(f"{_BASE_URL}{_QUERY_BASE}{_EXAMPLE_RANGE}")
    data = json.loads(response.read())
    assert 'description' in data['_embedded'][0]

    res = client.search(a.Instrument("GRIS") & a.sdc.ObsName('gris_20140426_000'))
    assert len(res) == 1
    description = res[0]['_embedded'][0].get('description')
    assert description['INSTRUMENT'] == 'gris'
    assert description['TELESCOPE'] == 'GREGOR'
    assert description['BTYPE'] == 'phot.count'


59
def test_search(client):
60
    """Test conversion of (supported) Attrs to query string."""
61

62
    assert not client._can_handle_query(a.Time("2019/01/01", "2021/01/01"))
63
64
65
66
    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"))

67
    assert not client._can_handle_query(a.Instrument("UVES"), a.Time("2019/01/01", "2021/01/01"))
68
69
70
71
    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"))

72
    query = a.Instrument("BBI") & a.Time("2019/01/01", "2021/01/01")
73
    # TODO: Verify returned observation records.
74
    assert client._can_handle_query(query)
75
    if HAS_DOCKERTEST:
76
        res = client.search(query)
77
78
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
79
80
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
81
82
                           r".http://dockertest:8083/sdc/bbi_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'bbi'},"
83
84
                           r"{'description.DATE_BEG':{'.lte':{'.date':'2021-01-01T00:00:00.000'}}},"
                           r"{'description.DATE_END':{'.gte':{'.date':'2019-01-01T00:00:00.000'}}}"
85
                           rf".*Confirm that RESTHeart is running on {_BASE_URL} and connected"):
86
87
88
89
90
91
            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)
92
93
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
94
95
96
97
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/lars_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'lars'},"
98
99
                           r"{'description.HPLT_TAN_MIN':{'.lte':12}},"
                           r"{'description.HPLT_TAN_MAX':{'.gte':-10}}"):
100
101
            client.search(query)

102
    query = a.Instrument("GRIS") & (a.sdc.Theta(85*u.deg, 600*u.arcmin) | a.sdc.PolStates('iq'))
103
104
105
    assert client._can_handle_query(query)
    if HAS_DOCKERTEST:
        res = client.search(query)
106
107
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
108
    else:
Derek Homeier's avatar
Derek Homeier committed
109
        with pytest.raises(URLError, match=r"Unable to execute search ") as exc:
110
111
            client.search(query)
        # Will raise on first of multi-part OR queries.
112
113
        assert "http://dockertest:8083/sdc/gris_observations?filter=" in str(exc.value)
        assert "{'description.THETA':{'$gte':10,'$lte':85}}" in str(exc.value)
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
        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),
130
131
132
133
                                   a.sdc.DataProduct('cube'), a.sdc.ObsName('gris_20140426_000'),
                                   a.sdc.Date('2021/01/31'), a.sdc.Filter('G'),
                                   a.sdc.PolStates('IQUV'), a.sdc.Telescope('VTT'),
                                   a.sdc.Target('Sunspot_22'), a.sdc.AtmosR0(*([1, 20000]*u.mm)),
134
135
136
137
                                   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),
138
                                   a.sdc.SpectralResolution(6000, 200000),
139
                                   a.sdc.TemporalResolution(2*u.s, 30*u.s),
140
                                   a.sdc.NDimensions(1, 2), a.sdc.PolXel(2, 4),
141
142
143
                                   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):
144
    """Test an example of all supported query attributes with automatic field names."""
145
146
147
    assert client._can_handle_query(a.Instrument("GRIS"), query)
    if HAS_DOCKERTEST:
        res = client.search(a.Instrument("GRIS") & query)
148
149
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
150
151
152
153
154
155
    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)
156
157
158
159
160
161
162
163
164
165
166
167
168


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)
169
170
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
171
172
173
174
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/bbi_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'bbi'},"
175
176
                           r"{'description.DATE_BEG':{'.lte':{'.date':'2019-01-01T00:00:00.000'}}},"
                           r"{'description.DATE_END':{'.gte':{'.date':'2019-01-09T00:00:00.000'}"):
177
178
179
180
181
182
183
184
185
            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)
186
187
        if len(res) > 0 and len(res[0].get('_embedded')) > 0:
            assert 'description' in res[0]['_embedded'][0]
188
189
190
191
    else:
        with pytest.raises(URLError, match=r"Unable to execute search "
                           r".http://dockertest:8083/sdc/lars_observations.filter="
                           r"{'.and':.{'description.INSTRUMENT':'lars'},"
192
193
                           r"{'description.HPLT_TAN_MIN':{'.lte':-10}},"
                           r"{'description.HPLT_TAN_MAX':{'.gte':12\}}"):
194
            client.search(query)