|
7 | 7 | from ..core import HSAClass
|
8 | 8 |
|
9 | 9 |
|
10 |
| - |
11 | 10 | PACS_ENDINGS = ["571.xml", "571.jpg", "214.fits.gz", "008.fits.gz",
|
12 | 11 | "674.fits.gz", "350.fits.gz", "README.pdf"]
|
13 | 12 | SPIRE_ENDINGS = ["898.xml", "898.jpg", "141.fits.gz", "045.fits.gz", "952.fits.gz",
|
|
20 | 19 | @pytest.mark.remote_data
|
21 | 20 | class TestHSARemote:
|
22 | 21 |
|
23 |
| - def test_download_data_observation_pacs(self, tmp_path): |
24 |
| - obs_id = "1342191813" |
25 |
| - parameters = {'retrieval_type': "OBSERVATION", |
26 |
| - 'observation_id': obs_id, |
27 |
| - 'instrument_name': "PACS", |
28 |
| - 'product_level': 'LEVEL3', |
29 |
| - 'cache': False, |
30 |
| - 'download_dir': tmp_path} |
31 |
| - expected_res = Path(tmp_path, obs_id + ".tar") |
32 |
| - hsa = HSAClass() |
33 |
| - res = hsa.download_data(**parameters) |
34 |
| - assert Path(res) == expected_res |
35 |
| - assert Path(res).is_file() |
36 |
| - with tarfile.open(res) as tar: |
37 |
| - names = tar.getnames() |
38 |
| - assert len(names) == len(PACS_ENDINGS) |
39 |
| - for name, ending in zip(names, PACS_ENDINGS): |
40 |
| - assert name.endswith(ending) |
41 |
| - |
42 |
| - def test_download_data_observation_pacs_filename(self, tmp_path): |
43 |
| - obs_id = "1342191813" |
44 |
| - fname = "output_file" |
45 |
| - parameters = {'retrieval_type': "OBSERVATION", |
46 |
| - 'observation_id': obs_id, |
47 |
| - 'instrument_name': "PACS", |
48 |
| - 'product_level': 'LEVEL3', |
49 |
| - 'filename': fname, |
50 |
| - 'cache': False, |
51 |
| - 'download_dir': tmp_path} |
52 |
| - expected_res = Path(tmp_path, fname + ".tar") |
53 |
| - hsa = HSAClass() |
54 |
| - res = hsa.download_data(**parameters) |
55 |
| - assert Path(res) == expected_res |
56 |
| - assert Path(res).is_file() |
57 |
| - with tarfile.open(res) as tar: |
58 |
| - names = tar.getnames() |
59 |
| - assert len(names) == len(PACS_ENDINGS) |
60 |
| - for name, ending in zip(names, PACS_ENDINGS): |
61 |
| - assert name.endswith(ending) |
62 |
| - |
63 |
| - def test_download_data_observation_pacs_compressed(self, tmp_path): |
64 |
| - obs_id = "1342191813" |
65 |
| - parameters = {'retrieval_type': "OBSERVATION", |
66 |
| - 'observation_id': obs_id, |
67 |
| - 'instrument_name': "PACS", |
68 |
| - 'product_level': 'LEVEL3', |
69 |
| - 'compress': 'true', |
70 |
| - 'cache': False, |
71 |
| - 'download_dir': tmp_path} |
72 |
| - expected_res = Path(tmp_path, obs_id + ".tgz") |
73 |
| - hsa = HSAClass() |
74 |
| - res = hsa.download_data(**parameters) |
75 |
| - assert Path(res) == expected_res |
76 |
| - assert Path(res).is_file() |
77 |
| - with tarfile.open(res) as tar: |
78 |
| - names = tar.getnames() |
79 |
| - assert len(names) == len(PACS_ENDINGS) |
80 |
| - for name, ending in zip(names, PACS_ENDINGS): |
81 |
| - assert name.endswith(ending) |
82 |
| - |
83 |
| - def test_download_data_observation_spire(self, tmp_path): |
84 |
| - obs_id = "1342191188" |
85 |
| - parameters = {'retrieval_type': "OBSERVATION", |
86 |
| - 'observation_id': obs_id, |
87 |
| - 'instrument_name': "SPIRE", |
88 |
| - 'product_level': 'LEVEL2', |
89 |
| - 'cache': False, |
90 |
| - 'download_dir': tmp_path} |
91 |
| - expected_res = Path(tmp_path, obs_id + ".tar") |
92 |
| - hsa = HSAClass() |
93 |
| - res = hsa.download_data(**parameters) |
94 |
| - assert Path(res) == expected_res |
95 |
| - assert Path(res).is_file() |
96 |
| - with tarfile.open(res) as tar: |
97 |
| - names = tar.getnames() |
98 |
| - assert len(names) == len(SPIRE_ENDINGS) |
99 |
| - for name, ending in zip(names, SPIRE_ENDINGS): |
100 |
| - assert name.endswith(ending) |
101 |
| - |
102 |
| - def test_download_data_postcard_pacs(self, tmp_path): |
103 |
| - obs_id = "1342191813" |
104 |
| - parameters = {'retrieval_type': "POSTCARD", |
105 |
| - 'observation_id': obs_id, |
106 |
| - 'instrument_name': "PACS", |
107 |
| - 'cache': False, |
108 |
| - 'download_dir': tmp_path} |
109 |
| - expected_res = Path(tmp_path, obs_id + ".jpg") |
110 |
| - hsa = HSAClass() |
111 |
| - res = hsa.download_data(**parameters) |
112 |
| - assert Path(res) == expected_res |
113 |
| - assert Path(res).is_file() |
114 |
| - |
115 |
| - def test_download_data_postcard_pacs_filename(self, tmp_path): |
116 |
| - obs_id = "1342191813" |
117 |
| - fname = "output_file" |
118 |
| - parameters = {'retrieval_type': "POSTCARD", |
119 |
| - 'observation_id': obs_id, |
120 |
| - 'instrument_name': "PACS", |
121 |
| - 'filename': fname, |
122 |
| - 'cache': False, |
123 |
| - 'download_dir': tmp_path} |
124 |
| - expected_res = Path(tmp_path, fname + ".jpg") |
125 |
| - hsa = HSAClass() |
126 |
| - res = hsa.download_data(**parameters) |
127 |
| - assert Path(res) == expected_res |
128 |
| - assert Path(res).is_file() |
129 |
| - |
130 |
| - def test_get_observation(self, tmp_path): |
131 |
| - obs_id = "1342191813" |
132 |
| - parameters = {'observation_id': obs_id, |
| 22 | + @pytest.mark.parametrize( |
| 23 | + "method,kwargs,expected_filename,expected_endings", |
| 24 | + [ |
| 25 | + ("download_data", {}, "1342191813.tar", PACS_ENDINGS), |
| 26 | + ("download_data", {"filename": "output_file"}, "output_file.tar", PACS_ENDINGS), |
| 27 | + ("download_data", {"compress": "true"}, "1342191813.tgz", PACS_ENDINGS), |
| 28 | + ( |
| 29 | + "download_data", |
| 30 | + { |
| 31 | + "observation_id": "1342191188", |
| 32 | + "instrument_name": "SPIRE", |
| 33 | + "product_level": "LEVEL2", |
| 34 | + }, |
| 35 | + "1342191188.tar", |
| 36 | + SPIRE_ENDINGS, |
| 37 | + ), |
| 38 | + ("get_observation", {}, "1342191813.tar", PACS_ENDINGS), |
| 39 | + ], |
| 40 | + ) |
| 41 | + def test_download_data_observation( |
| 42 | + self, method, kwargs, expected_filename, expected_endings, tmp_path |
| 43 | + ): |
| 44 | + parameters = {"observation_id": "1342191813", |
133 | 45 | 'instrument_name': "PACS",
|
134 | 46 | 'product_level': 'LEVEL3',
|
135 | 47 | 'cache': False,
|
136 | 48 | 'download_dir': tmp_path}
|
137 |
| - expected_res = Path(tmp_path, obs_id + ".tar") |
138 |
| - hsa = HSAClass() |
139 |
| - res = hsa.get_observation(**parameters) |
140 |
| - assert Path(res) == expected_res |
| 49 | + parameters.update(kwargs) |
| 50 | + if method == "download_data": |
| 51 | + res = HSAClass().download_data(**parameters, retrieval_type="OBSERVATION") |
| 52 | + elif method == "get_observation": |
| 53 | + res = HSAClass().get_observation(**parameters) |
| 54 | + assert Path(res) == tmp_path / expected_filename |
141 | 55 | assert Path(res).is_file()
|
142 | 56 | with tarfile.open(res) as tar:
|
143 | 57 | names = tar.getnames()
|
144 |
| - assert len(names) == len(PACS_ENDINGS) |
145 |
| - for name, ending in zip(names, PACS_ENDINGS): |
| 58 | + assert len(names) == len(expected_endings) |
| 59 | + for name, ending in zip(names, expected_endings): |
146 | 60 | assert name.endswith(ending)
|
147 | 61 |
|
148 |
| - def test_get_postcard(self, tmp_path): |
149 |
| - obs_id = "1342191813" |
150 |
| - parameters = {'observation_id': obs_id, |
| 62 | + @pytest.mark.parametrize( |
| 63 | + "method,kwargs,expected_filename", |
| 64 | + [ |
| 65 | + ("download_data", {}, "1342191813.jpg"), |
| 66 | + ("download_data", {"filename": "output_file"}, "output_file.jpg"), |
| 67 | + ("get_postcard", {}, "1342191813.jpg"), |
| 68 | + ], |
| 69 | + ) |
| 70 | + def test_download_data_postcard(self, method, kwargs, expected_filename, tmp_path): |
| 71 | + parameters = {"observation_id": "1342191813", |
151 | 72 | 'instrument_name': "PACS",
|
152 | 73 | 'cache': False,
|
153 | 74 | 'download_dir': tmp_path}
|
154 |
| - expected_res = Path(tmp_path, obs_id + ".jpg") |
155 |
| - hsa = HSAClass() |
156 |
| - res = hsa.get_postcard(**parameters) |
157 |
| - assert Path(res) == expected_res |
| 75 | + if method == "download_data": |
| 76 | + res = HSAClass().download_data(**parameters, **kwargs, retrieval_type="POSTCARD") |
| 77 | + elif method == "get_postcard": |
| 78 | + res = HSAClass().get_postcard(**parameters, **kwargs) |
| 79 | + assert Path(res) == tmp_path / expected_filename |
158 | 80 | assert Path(res).is_file()
|
0 commit comments