File: test_odata.py

package info (click to toggle)
sentinelsat 0.14-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 5,288 kB
  • sloc: python: 2,466; makefile: 10
file content (149 lines) | stat: -rw-r--r-- 5,696 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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
105
106
107
108
109
110
111
112
113
114
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
"""
Tests for functionality related to the OData API of SciHub (https://scihub.copernicus.eu/apihub/odata/v1/...)
"""
from datetime import datetime

import pytest
import requests_mock

from sentinelsat import SentinelAPIError, SentinelAPI
from sentinelsat.sentinel import _parse_odata_timestamp


@pytest.mark.fast
def test_convert_timestamp():
    assert _parse_odata_timestamp("/Date(1445588544652)/") == datetime(
        2015, 10, 23, 8, 22, 24, 652000
    )


@pytest.mark.vcr
@pytest.mark.scihub
def test_get_product_odata_short(api, smallest_online_products, read_yaml):
    responses = {}
    for prod in smallest_online_products:
        id = prod["id"]
        responses[id] = api.get_product_odata(id)
    expected = read_yaml("odata_response_short.yml", responses)
    assert sorted(responses) == sorted(expected)


def scrub_string(string, replacement=""):
    """Scrub a string from a VCR response body string
    """

    def before_record_response(response):
        response["body"]["string"] = response["body"]["string"].replace(string, replacement)
        return response

    return before_record_response


@pytest.mark.scihub
def test_get_product_odata_short_with_missing_online_key(api, vcr):
    uuid = "8df46c9e-a20c-43db-a19a-4240c2ed3b8b"
    expected_short = {
        "id": "8df46c9e-a20c-43db-a19a-4240c2ed3b8b",
        "size": 143549851,
        "md5": "D5E4DF5C38C6E97BF7E7BD540AB21C05",
        "url": "https://scihub.copernicus.eu/apihub/odata/v1/Products('8df46c9e-a20c-43db-a19a-4240c2ed3b8b')/$value",
        "date": datetime(2015, 11, 21, 10, 3, 56, 675000),
        "footprint": "POLYGON((-63.852531 -5.880887,-67.495872 -5.075419,-67.066071 -3.084356,-63.430576 -3.880541,"
        "-63.852531 -5.880887))",
        "title": "S1A_EW_GRDM_1SDV_20151121T100356_20151121T100429_008701_00C622_A0EC",
        "Online": True,
        "Creation Date": datetime(2015, 11, 21, 13, 22, 1, 652000),
        "Ingestion Date": datetime(2015, 11, 21, 13, 22, 4, 992000),
    }

    # scrub 'Online' key from response
    with vcr.use_cassette(
        "test_get_product_odata_short_with_missing_online_key",
        before_record_response=scrub_string(b'"Online":false,', b""),
    ):
        response = api.get_product_odata(uuid)
        assert response == expected_short


@pytest.mark.vcr
@pytest.mark.scihub
def test_get_product_odata_full(api, smallest_online_products, read_yaml):
    responses = {}
    for prod in smallest_online_products:
        id = prod["id"]
        responses[id] = api.get_product_odata(id, full=True)
    expected = read_yaml("odata_response_full.yml", responses)
    assert sorted(responses) == sorted(expected)


@pytest.mark.vcr
@pytest.mark.scihub
def test_get_product_info_bad_key(api):
    with pytest.raises(SentinelAPIError) as excinfo:
        api.get_product_odata("invalid-xyz")
    assert excinfo.value.msg == "InvalidKeyException : Invalid key (invalid-xyz) to access Products"


@pytest.mark.mock_api
def test_get_product_odata_scihub_down(read_fixture_file):
    api = SentinelAPI("mock_user", "mock_password")

    request_url = "https://scihub.copernicus.eu/apihub/odata/v1/Products('8df46c9e-a20c-43db-a19a-4240c2ed3b8b')?$format=json"

    with requests_mock.mock() as rqst:
        rqst.get(request_url, text="Mock SciHub is Down", status_code=503)
        with pytest.raises(SentinelAPIError) as excinfo:
            api.get_product_odata("8df46c9e-a20c-43db-a19a-4240c2ed3b8b")
        assert excinfo.value.msg == "Mock SciHub is Down"

        rqst.get(
            request_url,
            text='{"error":{"code":null,"message":{"lang":"en","value":'
            "\"No Products found with key '8df46c9e-a20c-43db-a19a-4240c2ed3b8b' \"}}}",
            status_code=500,
        )
        with pytest.raises(SentinelAPIError) as excinfo:
            api.get_product_odata("8df46c9e-a20c-43db-a19a-4240c2ed3b8b")
        assert (
            excinfo.value.msg
            == "No Products found with key '8df46c9e-a20c-43db-a19a-4240c2ed3b8b' "
        )

        rqst.get(request_url, text="Mock SciHub is Down", status_code=200)
        with pytest.raises(SentinelAPIError) as excinfo:
            api.get_product_odata("8df46c9e-a20c-43db-a19a-4240c2ed3b8b")
        assert excinfo.value.msg == "Mock SciHub is Down"

        # Test with a real "server under maintenance" response
        rqst.get(request_url, text=read_fixture_file("server_maintenance.html"), status_code=502)
        with pytest.raises(SentinelAPIError) as excinfo:
            api.get_product_odata("8df46c9e-a20c-43db-a19a-4240c2ed3b8b")
        assert "The Sentinels Scientific Data Hub will be back soon!" in excinfo.value.msg


@pytest.mark.mock_api
def test_is_online():
    api = SentinelAPI("mock_user", "mock_password")

    uuid = "98ca202b-2155-4181-be88-4358b2cbaaa0"
    invalid_uuid = "98ca202b-2155-4181-be88-xxxxxxxxxxxx"

    request_url = "https://scihub.copernicus.eu/apihub/odata/v1/Products('{}')/Online/$value"

    with requests_mock.mock() as rqst:
        rqst.get(request_url.format(uuid), text="true", status_code=200)
        assert api.is_online(uuid) == True

    with requests_mock.mock() as rqst:
        rqst.get(request_url.format(uuid), text="false", status_code=200)
        assert api.is_online(uuid) == False

    with requests_mock.mock() as rqst:
        rqst.get(
            request_url.format(invalid_uuid),
            text='{{"error":{{"code":null,"message":{{"lang":"en","value":'
            "Invalid key ({}) to access Products}}}}}}".format(invalid_uuid),
            status_code=200,
        )
        with pytest.raises(SentinelAPIError) as excinfo:
            api.is_online(invalid_uuid)