codekingpro/portable-devtools

1import os
2import urllib.request as urllib_request
3from shutil import rmtree
4from tempfile import NamedTemporaryFile, mkdtemp, mkstemp
5from urllib.error import URLError
6from urllib.parse import urlparse
7
8import pytest
9
10import numpy.lib._datasource as datasource
11from numpy.testing import assert_, assert_equal, assert_raises
12
13
14def urlopen_stub(url, data=None):
15    '''Stub to replace urlopen for testing.'''
16    if url == valid_httpurl():
17        tmpfile = NamedTemporaryFile(prefix='urltmp_')
18        return tmpfile
19    else:
20        raise URLError('Name or service not known')
21
22
23# setup and teardown
24old_urlopen = None
25
26
27def setup_module():
28    global old_urlopen
29
30    old_urlopen = urllib_request.urlopen
31    urllib_request.urlopen = urlopen_stub
32
33
34def teardown_module():
35    urllib_request.urlopen = old_urlopen
36
37
38# A valid website for more robust testing
39http_path = 'http://www.google.com/'
40http_file = 'index.html'
41
42http_fakepath = 'http://fake.abc.web/site/'
43http_fakefile = 'fake.txt'
44
45malicious_files = ['/etc/shadow', '../../shadow',
46                   '..\\system.dat', 'c:\\windows\\system.dat']
47
48magic_line = b'three is the magic number'
49
50
51# Utility functions used by many tests
52def valid_textfile(filedir):
53    # Generate and return a valid temporary file.
54    fd, path = mkstemp(suffix='.txt', prefix='dstmp_', dir=filedir, text=True)
55    os.close(fd)
56    return path
57
58
59def invalid_textfile(filedir):
60    # Generate and return an invalid filename.
61    fd, path = mkstemp(suffix='.txt', prefix='dstmp_', dir=filedir)
62    os.close(fd)
63    os.remove(path)
64    return path
65
66
67def valid_httpurl():
68    return http_path + http_file
69
70
71def invalid_httpurl():
72    return http_fakepath + http_fakefile
73
74
75def valid_baseurl():
76    return http_path
77
78
79def invalid_baseurl():
80    return http_fakepath
81
82
83def valid_httpfile():
84    return http_file
85
86
87def invalid_httpfile():
88    return http_fakefile
89
90
91class TestDataSourceOpen:
92    def test_ValidHTTP(self, tmp_path):
93        ds = datasource.DataSource(tmp_path)
94        fh = ds.open(valid_httpurl())
95        assert_(fh)
96        fh.close()
97
98    def test_InvalidHTTP(self, tmp_path):
99        ds = datasource.DataSource(tmp_path)
100        url = invalid_httpurl()
101        assert_raises(OSError, ds.open, url)
102        try:
103            ds.open(url)
104        except OSError as e:
105            # Regression test for bug fixed in r4342.
106            assert_(e.errno is None)
107
108    def test_InvalidHTTPCacheURLError(self, tmp_path):
109        ds = datasource.DataSource(tmp_path)
110        assert_raises(URLError, ds._cache, invalid_httpurl())
111
112    def test_ValidFile(self, tmp_path):
113        ds = datasource.DataSource(tmp_path)
114        local_file = valid_textfile(tmp_path)
115        fh = ds.open(local_file)
116        assert_(fh)
117        fh.close()
118
119    def test_InvalidFile(self, tmp_path):
120        ds = datasource.DataSource(tmp_path)
121        invalid_file = invalid_textfile(tmp_path)
122        assert_raises(OSError, ds.open, invalid_file)
123
124    def test_ValidGzipFile(self, tmp_path):
125        try:
126            import gzip
127        except ImportError:
128            # We don't have the gzip capabilities to test.
129            pytest.skip()
130        # Test datasource's internal file_opener for Gzip files.
131        ds = datasource.DataSource(tmp_path)
132        filepath = os.path.join(tmp_path, 'foobar.txt.gz')
133        fp = gzip.open(filepath, 'w')
134        fp.write(magic_line)
135        fp.close()
136        fp = ds.open(filepath)
137        result = fp.readline()
138        fp.close()
139        assert_equal(magic_line, result)
140
141    def test_ValidBz2File(self, tmp_path):
142        try:
143            import bz2
144        except ImportError:
145            # We don't have the bz2 capabilities to test.
146            pytest.skip()
147        # Test datasource's internal file_opener for BZip2 files.
148        ds = datasource.DataSource(tmp_path)
149        filepath = os.path.join(tmp_path, 'foobar.txt.bz2')
150        fp = bz2.BZ2File(filepath, 'w')
151        fp.write(magic_line)
152        fp.close()
153        fp = ds.open(filepath)
154        result = fp.readline()
155        fp.close()
156        assert_equal(magic_line, result)
157
158
159class TestDataSourceExists:
160    def test_ValidHTTP(self, tmp_path):
161        ds = datasource.DataSource(tmp_path)
162        assert_(ds.exists(valid_httpurl()))
163
164    def test_InvalidHTTP(self, tmp_path):
165        ds = datasource.DataSource(tmp_path)
166        assert_equal(ds.exists(invalid_httpurl()), False)
167
168    def test_ValidFile(self, tmp_path):
169        # Test valid file in destpath
170        ds = datasource.DataSource(tmp_path)
171        tmpfile = valid_textfile(tmp_path)
172        assert_(ds.exists(tmpfile))
173        # Test valid local file not in destpath
174        localdir = mkdtemp()
175        tmpfile = valid_textfile(localdir)
176        assert_(ds.exists(tmpfile))
177        rmtree(localdir)
178
179    def test_InvalidFile(self, tmp_path):
180        ds = datasource.DataSource(tmp_path)
181        tmpfile = invalid_textfile(tmp_path)
182        assert_equal(ds.exists(tmpfile), False)
183
184
185class TestDataSourceAbspath:
186    def test_ValidHTTP(self, tmp_path):
187        ds = datasource.DataSource(tmp_path)
188        _, netloc, upath, _, _, _ = urlparse(valid_httpurl())
189        local_path = os.path.join(tmp_path, netloc,
190                                  upath.strip(os.sep).strip('/'))
191        assert_equal(local_path, ds.abspath(valid_httpurl()))
192
193    def test_ValidFile(self, tmp_path):
194        ds = datasource.DataSource(tmp_path)
195        tmpfile = valid_textfile(tmp_path)
196        tmpfilename = os.path.split(tmpfile)[-1]
197        # Test with filename only
198        assert_equal(tmpfile, ds.abspath(tmpfilename))
199        # Test filename with complete path
200        assert_equal(tmpfile, ds.abspath(tmpfile))
201
202    def test_InvalidHTTP(self, tmp_path):
203        ds = datasource.DataSource(tmp_path)
204        _, netloc, upath, _, _, _ = urlparse(invalid_httpurl())
205        invalidhttp = os.path.join(tmp_path, netloc,
206                                   upath.strip(os.sep).strip('/'))
207        assert_(invalidhttp != ds.abspath(valid_httpurl()))
208
209    def test_InvalidFile(self, tmp_path):
210        ds = datasource.DataSource(tmp_path)
211        invalidfile = valid_textfile(tmp_path)
212        tmpfile = valid_textfile(tmp_path)
213        tmpfilename = os.path.split(tmpfile)[-1]
214        # Test with filename only
215        assert_(invalidfile != ds.abspath(tmpfilename))
216        # Test filename with complete path
217        assert_(invalidfile != ds.abspath(tmpfile))
218
219    def test_sandboxing(self, tmp_path):
220        ds = datasource.DataSource(tmp_path)
221        tmpfile = valid_textfile(tmp_path)
222        tmpfilename = os.path.split(tmpfile)[-1]
223
224        path = lambda x: os.path.abspath(ds.abspath(x))
225
226        assert_(path(valid_httpurl()).startswith(str(tmp_path)))
227        assert_(path(invalid_httpurl()).startswith(str(tmp_path)))
228        assert_(path(tmpfile).startswith(str(tmp_path)))
229        assert_(path(tmpfilename).startswith(str(tmp_path)))
230        for fn in malicious_files:
231            assert_(path(http_path + fn).startswith(str(tmp_path)))
232            assert_(path(fn).startswith(str(tmp_path)))
233
234    def test_windows_os_sep(self, tmp_path):
235        orig_os_sep = os.sep
236        try:
237            os.sep = '\\'
238            self.test_ValidHTTP(tmp_path)
239            self.test_ValidFile(tmp_path)
240            self.test_InvalidHTTP(tmp_path)
241            self.test_InvalidFile(tmp_path)
242            self.test_sandboxing(tmp_path)
243        finally:
244            os.sep = orig_os_sep
245
246
247class TestRepositoryAbspath:
248    def test_ValidHTTP(self, tmp_path):
249        repos = datasource.Repository(valid_baseurl(), tmp_path)
250        _, netloc, upath, _, _, _ = urlparse(valid_httpurl())
251        local_path = os.path.join(repos._destpath, netloc,
252                                  upath.strip(os.sep).strip('/'))
253        filepath = repos.abspath(valid_httpfile())
254        assert_equal(local_path, filepath)
255
256    def test_sandboxing(self, tmp_path):
257        repos = datasource.Repository(valid_baseurl(), tmp_path)
258        path = lambda x: os.path.abspath(repos.abspath(x))
259        assert_(path(valid_httpfile()).startswith(str(tmp_path)))
260        for fn in malicious_files:
261            assert_(path(http_path + fn).startswith(str(tmp_path)))
262            assert_(path(fn).startswith(str(tmp_path)))
263
264    def test_windows_os_sep(self, tmp_path):
265        orig_os_sep = os.sep
266        try:
267            os.sep = '\\'
268            self.test_ValidHTTP(tmp_path)
269            self.test_sandboxing(tmp_path)
270        finally:
271            os.sep = orig_os_sep
272
273
274class TestRepositoryExists:
275    def test_ValidFile(self, tmp_path):
276        # Create local temp file
277        repos = datasource.Repository(valid_baseurl(), tmp_path)
278        tmpfile = valid_textfile(tmp_path)
279        assert_(repos.exists(tmpfile))
280
281    def test_InvalidFile(self, tmp_path):
282        repos = datasource.Repository(valid_baseurl(), tmp_path)
283        tmpfile = invalid_textfile(tmp_path)
284        assert_equal(repos.exists(tmpfile), False)
285
286    def test_RemoveHTTPFile(self, tmp_path):
287        repos = datasource.Repository(valid_baseurl(), tmp_path)
288        assert_(repos.exists(valid_httpurl()))
289
290    def test_CachedHTTPFile(self, tmp_path):
291        localfile = valid_httpurl()
292        # Create a locally cached temp file with an URL based
293        # directory structure.  This is similar to what Repository.open
294        # would do.
295        repos = datasource.Repository(valid_baseurl(), tmp_path)
296        _, netloc, _, _, _, _ = urlparse(localfile)
297        local_path = os.path.join(repos._destpath, netloc)
298        os.mkdir(local_path, 0o0700)
299        tmpfile = valid_textfile(local_path)
300        assert_(repos.exists(tmpfile))
301
302
303class TestOpenFunc:
304    def test_DataSourceOpen(self, tmp_path):
305        local_file = valid_textfile(tmp_path)
306        # Test case where destpath is passed in
307        fp = datasource.open(local_file, destpath=tmp_path)
308        assert_(fp)
309        fp.close()
310        # Test case where default destpath is used
311        fp = datasource.open(local_file)
312        assert_(fp)
313        fp.close()
314
315def test_del_attr_handling():
316    # DataSource __del__ can be called
317    # even if __init__ fails when the
318    # Exception object is caught by the
319    # caller as happens in refguide_check
320    # is_deprecated() function
321
322    ds = datasource.DataSource()
323    # simulate failed __init__ by removing key attribute
324    # produced within __init__ and expected by __del__
325    del ds._istmpdest
326    # should not raise an AttributeError if __del__
327    # gracefully handles failed __init__:
328    ds.__del__()
329