forked from emissions-api/sentinel5dl
-
Notifications
You must be signed in to change notification settings - Fork 0
/
__main__.py
111 lines (87 loc) · 3.66 KB
/
__main__.py
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
import os
import sentinel5dl
import sentinel5dl.__main__ as executable
import tempfile
import unittest
import logging
import sys
testpath = os.path.dirname(os.path.abspath(__file__))
class TestSentinel5dl(unittest.TestCase):
def _mock_http_request(self, path, filename=None):
'''Mock HTTP requests to the ESA API
'''
# download
if filename is not None:
self._count_download += 1
with open(filename, 'wb') as f:
f.write(b'123')
return
# search request
if path.startswith('/api/stub/products?'):
self._count_search_request += 1
with open(os.path.join(testpath, 'products.json'), 'rb') as f:
return f.read()
# checksum request
if path.endswith('/Checksum/Value/$value'):
self._count_checksum_request += 1
# MD5 checksum for string `123`
return b'202CB962AC59075B964B07152D234B70'
def setUp(self):
'''Patch cURL based operation in sentinel5dl so that we do not really
make any HTTP requests and reset the request counters.
'''
setattr(sentinel5dl, '__http_request', self._mock_http_request)
self._count_search_request = 0
self._count_checksum_request = 0
self._count_download = 0
logging.getLogger(sentinel5dl.__name__).setLevel(logging.WARNING)
def test(self):
'''Test search and download.
'''
result = sentinel5dl.search(
polygon='POLYGON((7 49,13 49,13 52,7 52,7 49))',
begin_ts='2019-09-01T00:00:00.000Z',
end_ts='2019-09-17T23:59:59.999Z',
product='L2__CO____')
# The result returned by the mock contains four products but claims a
# total of eight products, making sentinel5dl request resources twice.
self.assertEqual(self._count_search_request, 2)
self.assertEqual(result['totalresults'], 8)
self.assertEqual(result['totalresults'], len(result['products']))
products = result['products']
with tempfile.TemporaryDirectory() as tmpdir:
# prepare a file which is half-downloaded
file_one = os.path.join(tmpdir, products[0]['identifier'] + '.nc')
with open(file_one, 'wb') as f:
f.write(b'12')
sentinel5dl.download(products, tmpdir)
# test files
for product in products:
filename = os.path.join(tmpdir, product['identifier'] + '.nc')
with open(filename, 'rb') as f:
self.assertEqual(f.read(), b'123')
# We should have downloaded four files and have an additional four
# files storing md5 checksums
self.assertEqual(len(os.listdir(tmpdir)), 8)
# We should have four checksum requests. One for each file
self.assertEqual(self._count_checksum_request, 4)
# We should have downloaded four unique files
self.assertEqual(self._count_download, 4)
class TestExecutable(unittest.TestCase):
def _mock_search(self, *args, **kwargs):
return {'products': []}
def _mock_download(self, products):
self.assertEqual(products, [])
def setUp(self):
# Mock library calls
setattr(executable, 'search', self._mock_search)
setattr(executable, 'download', self._mock_download)
logging.getLogger(sentinel5dl.__name__).setLevel(logging.WARNING)
# override sys.argv. Otherwise argparse is trying to parse it.
sys.argv = sys.argv[0:1]
def test(self):
'''Test the executable.
'''
executable.main()
if __name__ == '__main__':
unittest.main()