Add test for flat MAST downloads; consolidate test setup in fixture
# Licensed under a 3-clause BSD style license - see LICENSE.rst
+from re import sub
import numpy as np
import os
import pytest
@@ -21,6 +22,19 @@ from ...exceptions import (InvalidQueryError, MaxResultsWarning, NoResultsWarnin
OBSID = '1647157'
+@pytest.fixture(scope="module")
+def msa_product_table():
+ # Pull products for a JWST NIRSpec MSA observation with 6 known
+ # duplicates of the MSA configuration file, propID=2736
+ products = mast.Observations.get_product_list("87602009")
+
+ # Filter out everything but the MSA config file
+ mask = np.char.find(products["dataURI"], "_msa.fits") != -1
+ products = products[mask]
+
+ return products
+
+
@pytest.mark.remote_data
class TestMast:
@@ -290,21 +304,29 @@ class TestMast:
assert os.path.isfile(result2['Local Path'][0])
assert len(result2) == 1
- def test_observations_download_products_no_duplicates(self, tmpdir, caplog):
+ def test_observations_download_products_flat(self, tmp_path, msa_product_table):
- # Pull products for a JWST NIRSpec MSA observation with 6 known
- # duplicates of the MSA configuration file, propID=2736
- products = mast.Observations.get_product_list("87602009")
+ # Get a product list with 6 duplicate JWST MSA config files
+ products = msa_product_table
- # Filter out everything but the MSA config file
- mask = np.char.find(products["dataURI"], "_msa.fits") != -1
- products = products[mask]
+ # Download with flat=True
+ manifest = mast.Observations.download_products(products, flat=True,
+ download_dir=tmp_path)
+
+ downloaded_path = manifest["Local Path"][0]
+ filename = os.path.basename(downloaded_path)
+ assert downloaded_path == (tmp_path / filename).as_posix()
+
+ def test_observations_download_products_no_duplicates(self, tmp_path, caplog, msa_product_table):
+
+ # Get a product list with 6 duplicate JWST MSA config files
+ products = msa_product_table
assert len(products) == 6
# Download the product
manifest = mast.Observations.download_products(products,
- download_dir=str(tmpdir))
+ download_dir=tmp_path)
# Check that it downloads the MSA config file only once
assert len(manifest) == 1
@@ -313,6 +335,11 @@ class TestMast:
with caplog.at_level("INFO", logger="astroquery"):
assert "products were duplicates" in caplog.text
+ def test_observations_get_cloud_uris_no_duplicates(self, msa_product_table):
+
+ # Get a product list with 6 duplicate JWST MSA config files
+ products = msa_product_table
+
# enable access to public AWS S3 bucket
mast.Observations.enable_cloud_dataset()
@@ -320,7 +347,7 @@ class TestMast:
uris = mast.Observations.get_cloud_uris(products)
assert len(uris) == 1
- def test_observations_download_file(self, tmpdir):
+ def test_observations_download_file(self):
# enabling cloud connection
mast.Observations.enable_cloud_dataset(provider='AWS')
