#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2021 Satpy developers
#
# This file is part of satpy.
#
# satpy is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# satpy is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# satpy. If not, see <http://www.gnu.org/licenses/>.
"""Test for ancillary data downloading."""
from unittest import mock
import pytest
import yaml
from satpy.aux_download import DataDownloadMixin
from satpy.modifiers import ModifierBase
# NOTE:
# The following fixtures are not defined in this file, but are used and injected by Pytest:
# - tmpdir
pooch = pytest.importorskip("pooch")
README_URL = "https://raw.githubusercontent.com/pytroll/satpy/main/README.rst"
[docs]
class UnfriendlyModifier(ModifierBase, DataDownloadMixin):
"""Fake modifier that raises an exception in __init__."""
def __init__(self, name, prerequisites=None, optional_prerequisites=None, **kwargs):
"""Raise an exception if we weren't provided any prerequisites."""
if not prerequisites or len(prerequisites) != 1:
raise ValueError("Unexpected number of prereqs")
super().__init__(name, prerequisites, optional_prerequisites, **kwargs)
self.register_data_files({"url": kwargs["url"],
"filename": kwargs["filename"],
"known_hash": kwargs["known_hash"]})
[docs]
def _setup_custom_composite_config(base_dir):
from satpy.composites import StaticImageCompositor
from satpy.modifiers.atmosphere import ReflectanceCorrector
composite_config = base_dir.mkdir("composites").join("visir.yaml")
with open(composite_config, "w") as comp_file:
yaml.dump({
"sensor_name": "visir",
"modifiers": {
"test_modifier": {
"modifier": ReflectanceCorrector,
"url": README_URL,
"known_hash": None,
},
"unfriendly_modifier": {
"modifier": UnfriendlyModifier,
"url": README_URL,
"filename": "unfriendly.rst",
"known_hash": None,
}
},
"composites": {
"test_static": {
"compositor": StaticImageCompositor,
"url": README_URL,
"known_hash": None,
},
},
}, comp_file)
[docs]
def _setup_custom_reader_config(base_dir):
reader_config = base_dir.mkdir("readers").join("fake.yaml")
with open(reader_config, "wt") as comp_file:
# abstract base classes can't be converted so we do raw string
comp_file.write("""
reader:
name: "fake"
reader: !!python/name:satpy.readers.yaml_reader.FileYAMLReader
data_files:
- url: {}
known_hash: null
- url: {}
filename: "README2.rst"
known_hash: null
file_types: {{}}
""".format(README_URL, README_URL))
[docs]
def _setup_custom_writer_config(base_dir):
writer_config = base_dir.mkdir("writers").join("fake.yaml")
with open(writer_config, "wt") as comp_file:
# abstract base classes can't be converted so we do raw string
comp_file.write("""
writer:
name: "fake"
writer: !!python/name:satpy.writers.Writer
data_files:
- url: {}
known_hash: null
- url: {}
filename: "README2.rst"
known_hash: null
""".format(README_URL, README_URL))
[docs]
def _assert_reader_files_downloaded(readers, found_files):
r_cond1 = "readers/README.rst" in found_files
r_cond2 = "readers/README2.rst" in found_files
if readers is not None and not readers:
r_cond1 = not r_cond1
r_cond2 = not r_cond2
assert r_cond1
assert r_cond2
[docs]
def _assert_writer_files_downloaded(writers, found_files):
w_cond1 = "writers/README.rst" in found_files
w_cond2 = "writers/README2.rst" in found_files
if writers is not None and not writers:
w_cond1 = not w_cond1
w_cond2 = not w_cond2
assert w_cond1
assert w_cond2
[docs]
def _assert_comp_files_downloaded(comp_sensors, found_files):
comp_cond = "composites/README.rst" in found_files
if comp_sensors is not None and not comp_sensors:
comp_cond = not comp_cond
assert comp_cond
[docs]
def _assert_mod_files_downloaded(comp_sensors, found_files):
mod_cond = "modifiers/README.rst" in found_files
unfriendly_cond = "modifiers/unfriendly.rst" in found_files
if comp_sensors is not None and not comp_sensors:
mod_cond = not mod_cond
assert mod_cond
assert not unfriendly_cond
[docs]
class TestDataDownload:
"""Test basic data downloading functionality."""
[docs]
@pytest.fixture(autouse=True)
def _setup_custom_configs(self, tmpdir):
_setup_custom_composite_config(tmpdir)
_setup_custom_reader_config(tmpdir)
_setup_custom_writer_config(tmpdir)
self.tmpdir = tmpdir
[docs]
@pytest.mark.parametrize("comp_sensors", [tuple(), None, ("visir",)])
@pytest.mark.parametrize("writers", [[], None, ["fake"]])
@pytest.mark.parametrize("readers", [[], None, ["fake"]])
def test_find_registerable(self, readers, writers, comp_sensors):
"""Test that find_registerable finds some things."""
import satpy
from satpy.aux_download import find_registerable_files
with satpy.config.set(config_path=[self.tmpdir]), \
mock.patch("satpy.aux_download._FILE_REGISTRY", {}):
found_files = find_registerable_files(
readers=readers, writers=writers,
composite_sensors=comp_sensors,
)
_assert_reader_files_downloaded(readers, found_files)
_assert_writer_files_downloaded(writers, found_files)
_assert_comp_files_downloaded(comp_sensors, found_files)
_assert_mod_files_downloaded(comp_sensors, found_files)
[docs]
def test_limited_find_registerable(self):
"""Test that find_registerable doesn't find anything when limited."""
import satpy
from satpy.aux_download import find_registerable_files
file_registry = {}
with satpy.config.set(config_path=[self.tmpdir]), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry):
found_files = find_registerable_files(
readers=[], writers=[], composite_sensors=[],
)
assert not found_files
[docs]
def test_retrieve(self):
"""Test retrieving a single file."""
import satpy
from satpy.aux_download import find_registerable_files, retrieve
file_registry = {}
with satpy.config.set(config_path=[self.tmpdir], data_dir=str(self.tmpdir)), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry):
comp_file = "composites/README.rst"
found_files = find_registerable_files()
assert comp_file in found_files
assert not self.tmpdir.join(comp_file).exists()
retrieve(comp_file)
assert self.tmpdir.join(comp_file).exists()
[docs]
def test_offline_retrieve(self):
"""Test retrieving a single file when offline."""
import satpy
from satpy.aux_download import find_registerable_files, retrieve
file_registry = {}
with satpy.config.set(config_path=[self.tmpdir], data_dir=str(self.tmpdir), download_aux=True), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry):
comp_file = "composites/README.rst"
found_files = find_registerable_files()
assert comp_file in found_files
# the file doesn't exist, we can't download it
assert not self.tmpdir.join(comp_file).exists()
with satpy.config.set(download_aux=False):
pytest.raises(RuntimeError, retrieve, comp_file)
# allow downloading and get it
retrieve(comp_file)
assert self.tmpdir.join(comp_file).exists()
# turn off downloading and make sure we get local file
with satpy.config.set(download_aux=False):
local_file = retrieve(comp_file)
assert local_file
[docs]
def test_offline_retrieve_all(self):
"""Test registering and retrieving all files fails when offline."""
import satpy
from satpy.aux_download import retrieve_all
with satpy.config.set(config_path=[self.tmpdir], data_dir=str(self.tmpdir), download_aux=False):
pytest.raises(RuntimeError, retrieve_all)
[docs]
def test_retrieve_all(self):
"""Test registering and retrieving all files."""
import satpy
from satpy.aux_download import retrieve_all
file_registry = {}
file_urls = {}
with satpy.config.set(config_path=[self.tmpdir], data_dir=str(self.tmpdir)), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry), \
mock.patch("satpy.aux_download._FILE_URLS", file_urls), \
mock.patch("satpy.aux_download.find_registerable_files"):
comp_file = "composites/README.rst"
file_registry[comp_file] = None
file_urls[comp_file] = README_URL
assert not self.tmpdir.join(comp_file).exists()
retrieve_all()
assert self.tmpdir.join(comp_file).exists()
[docs]
def test_no_downloads_in_tests(self):
"""Test that tests aren't allowed to download stuff."""
import satpy
from satpy.aux_download import register_file, retrieve
file_registry = {}
with satpy.config.set(config_path=[self.tmpdir], data_dir=str(self.tmpdir),
download_aux=True), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry):
cache_key = "myfile.rst"
register_file(README_URL, cache_key)
assert not self.tmpdir.join(cache_key).exists()
pytest.raises(RuntimeError, retrieve, cache_key)
# touch the file so it gets created
open(self.tmpdir.join(cache_key), "w").close()
# offline downloading should still be allowed
with satpy.config.set(download_aux=False):
retrieve(cache_key)
[docs]
def test_download_script(self):
"""Test basic functionality of the download script."""
import satpy
from satpy.aux_download import retrieve_all_cmd
file_registry = {}
file_urls = {}
with satpy.config.set(config_path=[self.tmpdir]), \
mock.patch("satpy.aux_download._FILE_REGISTRY", file_registry), \
mock.patch("satpy.aux_download._FILE_URLS", file_urls), \
mock.patch("satpy.aux_download.find_registerable_files"):
comp_file = "composites/README.rst"
file_registry[comp_file] = None
file_urls[comp_file] = README_URL
assert not self.tmpdir.join(comp_file).exists()
retrieve_all_cmd(argv=["--data-dir", str(self.tmpdir)])
assert self.tmpdir.join(comp_file).exists()