openeo-python-client/tests/rest/test_imagecollectionclient.py at a4abdc1c4d74375be11830985beb19d89274faf0 · Open-EO/openeo-python-client · GitHub

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
import pathlib

import pytest

import openeo
from openeo.graphbuilder import GraphBuilder
from openeo.imagecollection import CollectionMetadata
from openeo.rest.imagecollectionclient import ImageCollectionClient

API_URL = "https://oeo.net"


@pytest.fixture
def session040(requests_mock):
    requests_mock.get(API_URL + "/", json={"api_version": "0.4.0"})
    session = openeo.connect(API_URL)
    # Reset graph builder
    GraphBuilder.id_counter = {}
    return session


def test_metadata_from_api(session040, requests_mock):
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={"foo": "bar"})
    metadata = session040.collection_metadata("SENTINEL2")
    assert metadata.get("foo") == "bar"


def test_metadata_load_collection(session040, requests_mock):
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={
        "properties": {
            "cube:dimensions": {
                "bands": {"type": "bands", "values": ["B2", "B3"]}
            },
            "eo:bands": [
                {"name": "B2", "common_name": "blue"},
                {"name": "B3", "common_name": "green"},
            ]
        }
    })
    im = ImageCollectionClient.load_collection('SENTINEL2', session=session040)
    assert im.metadata.bands == [
        CollectionMetadata.Band("B2", "blue", None),
        CollectionMetadata.Band("B3", "green", None)
    ]


def test_empty_mask():
    from shapely import geometry
    polygon = geometry.Polygon([[1.0, 1.0], [2.0, 1.0], [2.0, 1.0], [1.0, 1.0]])

    client = ImageCollectionClient(node_id=None, builder=GraphBuilder(), session=None)

    with pytest.raises(ValueError, match=r"Mask .+ has an area of 0.0"):
        client.mask(polygon)


def test_download(session040, requests_mock, tmpdir):
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={"foo": "bar"})
    requests_mock.post(API_URL + '/result', text="tiffdata")
    path = tmpdir.join("tmp.tiff")
    session040.load_collection("SENTINEL2").download(str(path), format="GTIFF")
    assert path.read() == "tiffdata"


def test_download_pathlib(session040, requests_mock, tmpdir):
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={"foo": "bar"})
    requests_mock.post(API_URL + '/result', text="tiffdata")
    path = tmpdir.join("tmp.tiff")
    session040.load_collection("SENTINEL2").download(pathlib.Path(str(path)), format="GTIFF")
    assert path.read() == "tiffdata"


def test_download_with_bearer_token(session040, requests_mock, tmpdir):
    """https://github.com/Open-EO/openeo-python-client/issues/95"""
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={"foo": "bar"})
    requests_mock.get(API_URL + '/credentials/basic', json={"access_token": "w3lc0m3"})
    session040.authenticate_basic("test", "test123")

    def result_callback(request, context):
        assert request.headers["Authorization"] == "Bearer w3lc0m3"
        return "tiffdata"

    requests_mock.post(API_URL + '/result', text=result_callback)
    path = tmpdir.join("tmp.tiff")
    session040.load_collection("SENTINEL2").download(str(path), format="GTIFF")
    assert path.read() == "tiffdata"


def test_dynamic_cube_method(session040, requests_mock):
    processes = [
        {
            "id": "make_larger",
            "description": "multiply a raster cube with a factor",
            "parameters": [
                {"name": "data", "schema": {"type": "object", "subtype": "raster-cube"}},
                {"name": "factor", "schema": {"type": "float"}},
            ]}
    ]
    requests_mock.get(API_URL + '/processes', json={"processes": processes})
    requests_mock.get(API_URL + "/collections/SENTINEL2", json={"foo": "bar"})

    cube = session040.load_collection("SENTINEL2")
    cube = cube.dynamic.make_larger(factor=42)
    assert set(cube.graph.keys()) == {"loadcollection1", "makelarger1"}
    assert cube.graph["makelarger1"] == {
        "process_id": "make_larger",
        "arguments": {
            "data": {"from_node": "loadcollection1"},
            "factor": 42,
        },
        "result": False
    }