/
__main__.py
139 lines (110 loc) · 4.74 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import datetime
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=datetime.datetime.fromtimestamp(0),
end_ts=datetime.datetime.now(),
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 testNoArguments(self):
'''Test the executable.
'''
executable.main()
def testPolygon(self):
'''Test with an invalid polygon.
'''
sys.argv = [sys.argv[0], '--polygon', '3 1, 4 4, 2 4, 1 2, 3 1', '.']
executable.main()
def testInvalidPolygons(self):
'''Tests with invalid polygons.
'''
invalid_polygons = (
'3 1 4 4, 2 4 1', # coordinates must be pairs
'3 1, 4 4, 2 4, 1 2', # polygon must be closed
'a s, d f, a s, d f', # coordinates must be numbers
)
for invalid_polygon in invalid_polygons:
sys.argv = [sys.argv[0], '--polygon', invalid_polygon, '.']
error_msg = f'Expecting SystemExit error when providing the '\
'invalid polygon "{invalid_polygon}"'
with self.assertRaises(SystemExit, msg=error_msg) as e:
executable.main()
error_msg = f'Expecting return code != 0 error when providing '\
'the invalid polygon "{invalid_polygon}"'
self.assertNotEqual(e.exception.code, 0, msg=error_msg)
if __name__ == '__main__':
unittest.main()