1# Copyright (c) 2018 The Pooch Developers.
2# Distributed under the terms of the BSD 3-Clause License.
3# SPDX-License-Identifier: BSD-3-Clause
4#
5# This code is part of the Fatiando a Terra project (https://www.fatiando.org)
6#
7"""
8Test the utility functions.
9"""
10import os
11import shutil
12import time
13from pathlib import Path
14import tempfile
15from tempfile import TemporaryDirectory
16from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
17
18import pytest
19
20from ..utils import (
21    parse_url,
22    make_local_storage,
23    temporary_file,
24    unique_file_name,
25)
26
27
28def test_unique_name_long():
29    "The file name should never be longer than 255 characters"
30    url = f"https://www.something.com/data{'a' * 500}.txt"
31    assert len(url) > 255
32    fname = unique_file_name(url)
33    assert len(fname) == 255
34    assert fname[-10:] == "aaaaaa.txt"
35    assert fname.split("-")[1][:10] == "aaaaaaaaaa"
36
37
38@pytest.mark.parametrize(
39    "pool",
40    [ThreadPoolExecutor, ProcessPoolExecutor],
41    ids=["threads", "processes"],
42)
43def test_make_local_storage_parallel(pool, monkeypatch):
44    "Try to create the cache folder in parallel"
45    # Can cause multiple attempts at creating the folder which leads to an
46    # exception. Check that this doesn't happen.
47    # See https://github.com/fatiando/pooch/issues/170
48
49    # Monkey path makedirs to make it delay before creating the directory.
50    # Otherwise, the dispatch is too fast and the directory will exist before
51    # another process tries to create it.
52
53    # Need to keep a reference to the original function to avoid infinite
54    # recursions from the monkey patching.
55    makedirs = os.makedirs
56
57    def mockmakedirs(path, exist_ok=False):  # pylint: disable=unused-argument
58        "Delay before calling makedirs"
59        time.sleep(1.5)
60        makedirs(path, exist_ok=exist_ok)
61
62    monkeypatch.setattr(os, "makedirs", mockmakedirs)
63
64    data_cache = os.path.join(os.curdir, "test_parallel_cache")
65    assert not os.path.exists(data_cache)
66
67    try:
68        with pool() as executor:
69            futures = [
70                executor.submit(make_local_storage, data_cache) for i in range(4)
71            ]
72            for future in futures:
73                future.result()
74            assert os.path.exists(data_cache)
75    finally:
76        if os.path.exists(data_cache):
77            shutil.rmtree(data_cache)
78
79
80def test_local_storage_makedirs_permissionerror(monkeypatch):
81    "Should warn the user when can't create the local data dir"
82
83    def mockmakedirs(path, exist_ok=False):  # pylint: disable=unused-argument
84        "Raise an exception to mimic permission issues"
85        raise PermissionError("Fake error")
86
87    data_cache = os.path.join(os.curdir, "test_permission")
88    assert not os.path.exists(data_cache)
89
90    monkeypatch.setattr(os, "makedirs", mockmakedirs)
91
92    with pytest.raises(PermissionError) as error:
93        make_local_storage(
94            path=data_cache,
95            env="SOME_VARIABLE",
96        )
97        assert "Pooch could not create data cache" in str(error)
98        assert "'SOME_VARIABLE'" in str(error)
99
100
101def test_local_storage_newfile_permissionerror(monkeypatch):
102    "Should warn the user when can't write to the local data dir"
103    # This is a separate function because there should be a warning if the data
104    # dir already exists but we can't write to it.
105
106    def mocktempfile(**kwargs):  # pylint: disable=unused-argument
107        "Raise an exception to mimic permission issues"
108        raise PermissionError("Fake error")
109
110    with TemporaryDirectory() as data_cache:
111        os.makedirs(os.path.join(data_cache, "1.0"))
112        assert os.path.exists(data_cache)
113
114        monkeypatch.setattr(tempfile, "NamedTemporaryFile", mocktempfile)
115
116        with pytest.raises(PermissionError) as error:
117            make_local_storage(
118                path=data_cache,
119                env="SOME_VARIABLE",
120            )
121            assert "Pooch could not write to data cache" in str(error)
122            assert "'SOME_VARIABLE'" in str(error)
123
124
125@pytest.mark.parametrize(
126    "url,output",
127    [
128        (
129            "http://127.0.0.1:8080/test.nc",
130            {"protocol": "http", "netloc": "127.0.0.1:8080", "path": "/test.nc"},
131        ),
132        (
133            "ftp://127.0.0.1:8080/test.nc",
134            {"protocol": "ftp", "netloc": "127.0.0.1:8080", "path": "/test.nc"},
135        ),
136        (
137            "doi:10.6084/m9.figshare.923450.v1/dike.json",
138            {
139                "protocol": "doi",
140                "netloc": "10.6084/m9.figshare.923450.v1",
141                "path": "/dike.json",
142            },
143        ),
144    ],
145    ids=["http", "ftp", "doi"],
146)
147def test_parse_url(url, output):
148    "Parse URL into 3 components"
149    assert parse_url(url) == output
150
151
152def test_parse_url_invalid_doi():
153    "Should fail if we forget to not include // in the DOI link"
154    with pytest.raises(ValueError):
155        parse_url("doi://XXX/XXX/fname.txt")
156
157
158def test_temporary_file():
159    "Make sure the file is writable and cleaned up in the end"
160    with temporary_file() as tmp:
161        assert Path(tmp).exists()
162        with open(tmp, "w") as outfile:
163            outfile.write("Meh")
164        with open(tmp) as infile:
165            assert infile.read().strip() == "Meh"
166    assert not Path(tmp).exists()
167
168
169def test_temporary_file_path():
170    "Make sure the file is writable and cleaned up in the end when given a dir"
171    with TemporaryDirectory() as path:
172        with temporary_file(path) as tmp:
173            assert Path(tmp).exists()
174            assert path in tmp
175            with open(tmp, "w") as outfile:
176                outfile.write("Meh")
177            with open(tmp) as infile:
178                assert infile.read().strip() == "Meh"
179        assert not Path(tmp).exists()
180
181
182def test_temporary_file_exception():
183    "Make sure the file is writable and cleaned up when there is an exception"
184    try:
185        with temporary_file() as tmp:
186            assert Path(tmp).exists()
187            raise ValueError("Nooooooooo!")
188    except ValueError:
189        assert not Path(tmp).exists()
190