Repository: ska-sa/katdal

Issue:
Replace underscores with dashes when loading buckets from RDBs
The new Ceph installation at MeerKAT has done away with underscores in bucket names. This is to make the ceph s3 comply with the s3 protocol.

That means that some of our legacy data from before 2019 has had their buckets renamed, but the RDBs have not been updated to reflect that.

Eg. http://archive-gw-1.kat.ac.za:7480/1540580961_sdp_l1_flags is now http://archive-gw-1.kat.ac.za:7480/1540580961-sdp-l1-flags

Link to the example in the web-archive. https://archive.sarao.ac.za/search/1540580961/ This data is currently in ceph, so a good test.

We need to make a fix for katdal to replace underscores with dashes for all buckets accessed directly from the archive.

I am happy to test this out on our transfer nodes once you have an update for this.

Discussion:
@ludwigschwardt: Thanks @ctgschollar! Could you copy and paste this into an SPR1 JIRA ticket as well?@ctgschollar: Here you go
https://skaafrica.atlassian.net/browse/SPR1-2990@spassmoor: Hi @ludwigschwardt  This is quite urgent now as the Galactic plane legacy survey us about to be released/published and a large chunk of that data is pre-2019 @ludwigschwardt: Fixed by #370.

Solution patch:
diff --git a/katdal/averager.py b/katdal/averager.py
index fd91cd4..a88b4ea 100644
--- a/katdal/averager.py
+++ b/katdal/averager.py
@@ -46,8 +46,8 @@ def _average_visibilities(vis, weight, flag, timeav, chanav, flagav):
         vis_sum = np.empty(bl_step, vis.dtype)
         vis_weight_sum = np.empty(bl_step, vis.dtype)
         weight_sum = np.empty(bl_step, weight.dtype)
-        flag_any = np.empty(bl_step, dtype=bool)
-        flag_all = np.empty(bl_step, dtype=bool)
+        flag_any = np.empty(bl_step, dtype=np.bool_)
+        flag_all = np.empty(bl_step, dtype=np.bool_)
         for av_t in range(0, av_n_time):
             tstart = av_t * timeav
             for bstart in range(0, n_bl, bl_step):
diff --git a/katdal/chunkstore_s3.py b/katdal/chunkstore_s3.py
index e8922c8..8281ccd 100644
--- a/katdal/chunkstore_s3.py
+++ b/katdal/chunkstore_s3.py
@@ -74,6 +74,7 @@ _BUCKET_POLICY = {
     ]
 }
 
+_CHUNK_EXTENSION = '.npy'
 # These HTTP responses typically indicate temporary S3 server / proxy overload,
 # which will trigger retries terminating in a missing data response if unsuccessful.
 _DEFAULT_SERVER_GLITCHES = (500, 502, 503, 504)
@@ -205,6 +206,17 @@ def _bucket_url(url):
     return split_url._replace(path=bucket_name).geturl()
 
 
+def _normalise_bucket_name(url):
+    """Ensure that S3 bucket name in `url` contains dashes and not underscores."""
+    split_url = urllib.parse.urlsplit(url)
+    # Split path into first component (S3 bucket name) and the rest (key name or empty)
+    path_components = split_url.path.lstrip('/').split('/', 1)
+    path_components[0] = path_components[0].replace('_', '-')
+    path = '/' + '/'.join(path_components)
+    # Note to self: namedtuple._replace is not a private method, despite the underscore!
+    return split_url._replace(path=path).geturl()
+
+
 class AuthorisationFailed(StoreUnavailable):
     """Authorisation failed, e.g. due to invalid, malformed or expired token."""
 
@@ -568,9 +580,34 @@ class S3ChunkStore(ChunkStore):
         self.public_read = public_read
         self.expiry_days = int(expiry_days)
 
-    def _chunk_url(self, chunk_name, extension='.npy'):
-        """Assemble URL corresponding to chunk (or array) name."""
-        return urllib.parse.urljoin(self._url, to_str(urllib.parse.quote(chunk_name + extension)))
+    def make_url(self, relative_path):
+        """Assemble valid URL by combining base URL with `relative_path`.
+
+        Replace underscores in the S3 bucket name with dashes as well.
+
+        Parameters
+        ----------
+        relative_path : str
+            Path relative to base store URL / endpoint
+
+        Returns
+        -------
+        url : str
+            Complete URL
+
+        Notes
+        -----
+        Before 19 December 2018 the MeerKAT Ceph archive had underscores in
+        its S3 bucket names, even though it was in violation of the S3 spec.
+        Since October 2023 the archive runs a stricter version of Ceph and
+        those older buckets were renamed to use dashes / hyphens instead.
+        The metadata in the corresponding RDB files still refer to bucket
+        names with underscores though, and this method fixes those instances
+        while assembling their URLs.
+        """
+        relative_path = to_str(urllib.parse.quote(relative_path))
+        url = urllib.parse.urljoin(self._url, relative_path)
+        return _normalise_bucket_name(url)
 
     def request(
         self,
@@ -662,7 +699,7 @@ class S3ChunkStore(ChunkStore):
         """See the docstring of :meth:`ChunkStore.get_chunk`."""
         dtype = np.dtype(dtype)
         chunk_name, shape = self.chunk_metadata(array_name, slices, dtype=dtype)
-        url = self._chunk_url(chunk_name)
+        url = self.make_url(chunk_name + _CHUNK_EXTENSION)
         # Our hacky optimisation to speed up response reading doesn't
         # work with non-identity encodings.
         headers = {'Accept-Encoding': 'identity'}
@@ -682,7 +719,7 @@ class S3ChunkStore(ChunkStore):
     def create_array(self, array_name):
         """See the docstring of :meth:`ChunkStore.create_array`."""
         # Array name is formatted as bucket/array but we only need to create bucket
-        array_url = self._chunk_url(array_name, extension='')
+        array_url = self.make_url(array_name)
         bucket_url = _bucket_url(array_url)
         self._create_bucket(bucket_url)
 
@@ -710,7 +747,7 @@ class S3ChunkStore(ChunkStore):
     def put_chunk(self, array_name, slices, chunk):
         """See the docstring of :meth:`ChunkStore.put_chunk`."""
         chunk_name, _ = self.chunk_metadata(array_name, slices, chunk=chunk)
-        url = self._chunk_url(chunk_name)
+        url = self.make_url(chunk_name + _CHUNK_EXTENSION)
         npy_header, chunk = npy_header_and_body(chunk)
         # Compute the MD5 sum to protect the object against corruption in
         # transmission.
@@ -725,13 +762,13 @@ class S3ChunkStore(ChunkStore):
         """See the docstring of :meth:`ChunkStore.mark_complete`."""
         self.create_array(array_name)
         obj_name = self.join(array_name, 'complete')
-        url = urllib.parse.urljoin(self._url, obj_name)
+        url = self.make_url(obj_name)
         self.request('PUT', url, chunk_name=obj_name, data=b'')
 
     def is_complete(self, array_name):
         """See the docstring of :meth:`ChunkStore.is_complete`."""
         obj_name = self.join(array_name, 'complete')
-        url = urllib.parse.urljoin(self._url, obj_name)
+        url = self.make_url(obj_name)
         try:
             self.request('GET', url, chunk_name=obj_name)
         except ChunkNotFound:
diff --git a/katdal/dataset.py b/katdal/dataset.py
index 51a5845..8613a82 100644
--- a/katdal/dataset.py
+++ b/katdal/dataset.py
@@ -781,20 +781,26 @@ class DataSet:
                 self._time_keep &= scan_keep
             elif k == 'targets':
                 targets = v if is_iterable(v) else [v]
-                target_keep = np.zeros(len(self._time_keep), dtype=bool)
-                target_index_sensor = self.sensor.get('Observation/target_index')
+                target_indices = []
                 for t in targets:
                     try:
                         if isinstance(t, numbers.Integral):
-                            target_index = t
+                            target_indices.append(t)
                         elif isinstance(t, katpoint.Target) or isinstance(t, str) and ',' in t:
-                            target_index = self.catalogue.targets.index(t)
+                            target_indices.append(self.catalogue.targets.index(t))
                         else:
-                            target_index = self.catalogue.targets.index(self.catalogue[t])
+                            targets_with_name = self.catalogue._targets_with_name(t)
+                            if not targets_with_name:
+                                raise KeyError(f"Catalogue has no target named '{t}'")
+                            for t2 in targets_with_name:
+                                target_indices.append(self.catalogue.targets.index(t2))
                     except (KeyError, ValueError):
                         # Warn here, in case the user gets the target subtly wrong and wonders why it is not selected
                         logger.warning("Skipping unknown selected target '%s'", t)
                         continue
+                target_keep = np.zeros(len(self._time_keep), dtype=bool)
+                target_index_sensor = self.sensor.get('Observation/target_index')
+                for target_index in set(target_indices):
                     target_keep |= (target_index_sensor == target_index)
                 self._time_keep &= target_keep
             # Selections that affect frequency axis


Test patch:
diff --git a/katdal/test/test_chunkstore_s3.py b/katdal/test/test_chunkstore_s3.py
index 50c276d..03d87c9 100644
--- a/katdal/test/test_chunkstore_s3.py
+++ b/katdal/test/test_chunkstore_s3.py
@@ -55,6 +55,7 @@ from urllib3.util.retry import Retry
 
 from katdal.chunkstore import ChunkNotFound, StoreUnavailable
 from katdal.chunkstore_s3 import (
+    _CHUNK_EXTENSION,
     _DEFAULT_SERVER_GLITCHES,
     InvalidToken,
     S3ChunkStore,
@@ -64,6 +65,7 @@ from katdal.chunkstore_s3 import (
     decode_jwt,
     read_array,
     _read_object,
+    _normalise_bucket_name,
 )
 from katdal.datasources import TelstateDataSource, DataSourceNotFound
 from katdal.test.s3_utils import MissingProgram, S3Server, S3User
@@ -71,8 +73,11 @@ from katdal.test.test_chunkstore import ChunkStoreTestBase, generate_arrays
 from katdal.test.test_datasources import (assert_telstate_data_source_equal,
                                           make_fake_data_source)
 
-# Use a standard bucket for most tests to ensure valid bucket name (regex '^[0-9a-z.-]{3,63}$')
-BUCKET = 'katdal-unittest'
+# Use a standard bucket for most tests to ensure a valid bucket name
+# (regex '^[0-9a-z.-]{3,63}$'). While we are at it, go a step further
+# and check the underscore-to-dash conversion in `S3ChunkStore.make_url`
+# by including an underscore here.
+BUCKET = 'katdal_unittest'
 # Also authorise this prefix for tests that will make their own buckets
 PREFIX = '1234567890'
 # Pick quick but different timeouts and retries for unit tests:
@@ -166,6 +171,19 @@ class TestReadArray:
         self._truncate_and_fail_to_read(-1, 2)
 
 
+@pytest.mark.parametrize('url,expected',
+    [
+        ('https://archive/bucket/key/000', 'https://archive/bucket/key/000'),
+        ('https://archive/bucket/key/0_0', 'https://archive/bucket/key/0_0'),
+        ('https://archive/bu_ket/key/0_0', 'https://archive/bu-ket/key/0_0'),
+        ('https://a-chive/bu_ket/key/0-0', 'https://a-chive/bu-ket/key/0-0'),
+        ('https://archive/just-a_bucket', 'https://archive/just-a-bucket'),
+    ]
+)
+def test_normalise_bucket_name(url, expected):
+    assert _normalise_bucket_name(url) == expected
+
+
 def encode_jwt(payload):
     """Generate JWT token with ES256-encoded signature."""
     header = {'alg': 'ES256', 'typ': 'JWT'}
@@ -350,7 +368,7 @@ class TestS3ChunkStore(ChunkStoreTestBase):
             rdb_path = self.store.join(suggestion, cbid, rdb_filename)
         else:
             rdb_path = self.store.join(cbid, rdb_filename)
-        rdb_url = urllib.parse.urljoin(self.store_url, rdb_path)
+        rdb_url = self.store.make_url(rdb_path)
         self.store.create_array(cbid)
         self.store.request('PUT', rdb_url, data=rdb_data)
         # Check that data source can be constructed from URL (with auto chunk store)
@@ -369,7 +387,7 @@ class TestS3ChunkStore(ChunkStoreTestBase):
         with pytest.raises(StoreUnavailable):
             self.store.get_chunk(f'{BUCKET}-empty/x', slices, dtype)
         # Check that the standard bucket has not been verified yet
-        bucket_url = urllib.parse.urljoin(self.store._url, BUCKET)
+        bucket_url = self.store.make_url(BUCKET)
         assert bucket_url not in self.store._verified_buckets
         # Check that the standard bucket remains verified after initial check
         self.test_chunk_non_existent()
@@ -580,7 +598,7 @@ class TestS3ChunkStoreToken(TestS3ChunkStore):
             raise RuntimeError('Cannot use multiple target URLs with http proxy')
         # The token authorises the standard bucket and anything starting with PREFIX
         # (as well as suggestions prepended to the path)
-        token = encode_jwt({'prefix': [BUCKET, PREFIX, 'please']})
+        token = encode_jwt({'prefix': [BUCKET.replace('_', '-'), PREFIX, 'please']})
         kwargs.setdefault('token', token)
         return super().prepare_store_args(cls.proxy_url, credentials=None, **kwargs)
 
@@ -619,7 +637,7 @@ class TestS3ChunkStoreToken(TestS3ChunkStore):
         suggestion = 'please-respond-with-500-for-0.2-seconds'
         chunk, slices, array_name = self._put_chunk(suggestion)
         chunk_name, _ = self.store.chunk_metadata(array_name, slices, dtype=chunk.dtype)
-        url = self.store._chunk_url(chunk_name)
+        url = self.store.make_url(chunk_name + _CHUNK_EXTENSION)
         response = self.store.request('GET', url, ignored_errors=(500,), retries=0, stream=True)
         assert response.status_code == 500
 
@@ -753,10 +771,10 @@ class TestS3ChunkStoreToken(TestS3ChunkStore):
         data = b'x' * 1000
         cbid = PREFIX
         path = self.store.join(cbid, f'test_recovery_from_{condition}.bin')
-        url = urllib.parse.urljoin(self.store_url, path)
+        url = self.store.make_url(path)
         self.store.create_array(cbid)
         self.store.request('PUT', url, data=data)
         suggestion = f'please-{condition}-read-after-400-bytes-for-0.52-seconds'
-        url = urllib.parse.urljoin(self.store_url, self.store.join(suggestion, path))
+        url = self.store.make_url(self.store.join(suggestion, path))
         retrieved_data = self.store.request('GET', url, process=_read_object)
         assert retrieved_data == data
diff --git a/katdal/test/test_dataset.py b/katdal/test/test_dataset.py
index b7415a0..16cc775 100644
--- a/katdal/test/test_dataset.py
+++ b/katdal/test/test_dataset.py
@@ -16,7 +16,10 @@
 
 """Tests for :py:mod:`katdal.dataset`."""
 
+import logging
+
 import numpy as np
+import pytest
 from katpoint import Antenna, Target, Timestamp, rad2deg
 from numpy.testing import assert_array_almost_equal, assert_array_equal
 
@@ -27,36 +30,87 @@ from katdal.sensordata import SensorCache
 from katdal.spectral_window import SpectralWindow
 
 
+ANTENNAS = [
+    Antenna('m000, -30:42:39.8, 21:26:38.0, 1086.6, 13.5, -8.264 -207.29 8.5965'),
+    Antenna('m063, -30:42:39.8, 21:26:38.0, 1086.6, 13.5, -3419.5845 -1840.48 16.3825')
+]
+CORRPRODS = [
+    ('m000h', 'm000h'), ('m000v', 'm000v'),
+    ('m063h', 'm063h'), ('m063v', 'm063v'),
+    ('m000h', 'm063h'), ('m000v', 'm063v')
+]
+SUBARRAY = Subarray(ANTENNAS, CORRPRODS)
+SPW = SpectralWindow(
+    centre_freq=1284e6, channel_width=0, num_chans=16, sideband=1, bandwidth=856e6
+)
+
+
 class MinimalDataSet(DataSet):
-    """Minimal data set containing a single target track."""
-    def __init__(self, target, subarray, spectral_window, timestamps):
+    """Minimal data set containing a series of slews and tracks.
+
+    The timestamps are divided evenly into compound scans (one per target).
+    Each compound scan consists of a 1-dump slew followed by a track.
+
+    This has to be a derived class instead of a factory function or fixture
+    because :class:`DataSet` is abstract. (XXX Actually it only needs
+    timestamps to be implemented for these tests to work, so it is nearly
+    there.)
+
+    Parameters
+    ----------
+    targets : list of :class:`katpoint.Target`
+    timestamps : array of float
+    subarray : :class:`katdal.dataset.Subarray`
+    spectral_window : :class:`katdal.spectral_window.SpectralWindow`
+    """
+    def __init__(self, targets, timestamps, subarray=SUBARRAY, spectral_window=SPW):
         super().__init__(name='test', ref_ant='array')
         num_dumps = len(timestamps)
         num_chans = spectral_window.num_chans
         num_corrprods = len(subarray.corr_products)
         dump_period = timestamps[1] - timestamps[0]
 
+        num_compscans = len(targets)
+        num_dumps_per_compscan = num_dumps // num_compscans
+        assert num_dumps_per_compscan * num_compscans == num_dumps, \
+            "len(timestamps) should be an integer multiple of len(targets)"
+        compscan_starts = np.arange(0, num_dumps, num_dumps_per_compscan)
+        compscan_events = np.r_[compscan_starts, num_dumps]
+        # Each slew contains just 1 dump to make things simple
+        scan_events = sorted(np.r_[compscan_starts, compscan_starts + 1, num_dumps])
+        target_sensor = CategoricalData(targets, compscan_events)
+
         def constant_sensor(value):
             return CategoricalData([value], [0, num_dumps])
 
         self.subarrays = [subarray]
         self.spectral_windows = [spectral_window]
         sensors = {}
+        sensors['Observation/spw_index'] = constant_sensor(0)
+        sensors['Observation/subarray_index'] = constant_sensor(0)
         for ant in subarray.ants:
             sensors[f'Antennas/{ant.name}/antenna'] = constant_sensor(ant)
-            az, el = target.azel(timestamps, ant)
-            sensors[f'Antennas/{ant.name}/az'] = az
-            sensors[f'Antennas/{ant.name}/el'] = el
-        # Extract array reference position as 'array_ant' from last antenna
-        array_ant_fields = ['array'] + ant.description.split(',')[1:5]
-        array_ant = Antenna(','.join(array_ant_fields))
+            ant_az = []
+            ant_el = []
+            for segment, target in target_sensor.segments():
+                az, el = target.azel(timestamps[segment], ant)
+                ant_az.append(az)
+                ant_el.append(el)
+            sensors[f'Antennas/{ant.name}/az'] = np.concatenate(ant_az)
+            sensors[f'Antennas/{ant.name}/el'] = np.concatenate(ant_el)
+        array_ant = subarray.ants[0].array_reference_antenna()
         sensors['Antennas/array/antenna'] = constant_sensor(array_ant)
 
-        sensors['Observation/target'] = constant_sensor(target)
-        for name in ('spw', 'subarray', 'scan', 'compscan', 'target'):
-            sensors[f'Observation/{name}_index'] = constant_sensor(0)
-        sensors['Observation/scan_state'] = constant_sensor('track')
-        sensors['Observation/label'] = constant_sensor('track')
+        compscan_sensor = CategoricalData(range(num_compscans), compscan_events)
+        label_sensor = CategoricalData(['track'] * num_compscans, compscan_events)
+        sensors['Observation/target'] = target_sensor
+        sensors['Observation/compscan_index'] = compscan_sensor
+        sensors['Observation/target_index'] = compscan_sensor
+        sensors['Observation/label'] = label_sensor
+        scan_sensor = CategoricalData(range(2 * num_compscans), scan_events)
+        state_sensor = CategoricalData(['slew', 'track'] * num_compscans, scan_events)
+        sensors['Observation/scan_index'] = scan_sensor
+        sensors['Observation/scan_state'] = state_sensor
 
         self._timestamps = timestamps
         self._time_keep = np.full(num_dumps, True, dtype=bool)
@@ -68,7 +122,7 @@ class MinimalDataSet(DataSet):
         self.sensor = SensorCache(sensors, timestamps, dump_period,
                                   keep=self._time_keep,
                                   virtual=DEFAULT_VIRTUAL_SENSORS)
-        self.catalogue.add(target)
+        self.catalogue.add(targets)
         self.catalogue.antenna = array_ant
         self.select(spw=0, subarray=0)
 
@@ -114,19 +168,10 @@ def test_selection_to_list():
 class TestVirtualSensors:
     def setup_method(self):
         self.target = Target('PKS1934-638, radec, 19:39, -63:42')
-        self.antennas = [Antenna('m000, -30:42:39.8, 21:26:38.0, 1086.6, 13.5, '
-                                 '-8.264 -207.29 8.5965'),
-                         Antenna('m063, -30:42:39.8, 21:26:38.0, 1086.6, 13.5, '
-                                 '-3419.5845 -1840.48 16.3825')]
-        corrprods = [('m000h', 'm000h'), ('m000v', 'm000v'),
-                     ('m063h', 'm063h'), ('m063v', 'm063v'),
-                     ('m000h', 'm063h'), ('m000v', 'm063v')]
-        subarray = Subarray(self.antennas, corrprods)
-        spw = SpectralWindow(centre_freq=1284e6, channel_width=0, num_chans=16,
-                             sideband=1, bandwidth=856e6)
         # Pick a time when the source is up as that seems more realistic
         self.timestamps = 1234667890.0 + 1.0 * np.arange(10)
-        self.dataset = MinimalDataSet(self.target, subarray, spw, self.timestamps)
+        self.dataset = MinimalDataSet([self.target], self.timestamps)
+        self.antennas = self.dataset.subarrays[0].ants
         self.array_ant = self.dataset.sensor.get('Antennas/array/antenna')[0]
 
     def test_timestamps(self):
@@ -164,8 +209,97 @@ class TestVirtualSensors:
         assert_array_equal(self.dataset.v[:, 5], v)
         assert_array_equal(self.dataset.w[:, 5], w)
 
-    def test_selecting_antenna(self):
-        self.dataset.select(ants='~m000')
-        assert_array_equal(
-            self.dataset.corr_products,
-            [('m063h', 'm063h'), ('m063v', 'm063v')])
+
+@pytest.fixture
+def dataset():
+    """A basic dataset used to test the selection mechanism."""
+    targets = [
+        # It would have been nice to have radec = 19:39, -63:42 but then
+        # selection by description string does not work because the catalogue's
+        # description string pads it out to radec = 19:39:00.00, -63:42:00.0.
+        # (XXX Maybe fix Target comparison in katpoint to support this?)
+        Target('J1939-6342 | PKS1934-638, radec bpcal, 19:39:25.03, -63:42:45.6'),
+        Target('J1939-6342, radec gaincal, 19:39:25.03, -63:42:45.6'),
+        Target('J0408-6545 | PKS 0408-65, radec bpcal, 4:08:20.38, -65:45:09.1'),
+        Target('J1346-6024 | Cen B, radec, 13:46:49.04, -60:24:29.4'),
+    ]
+    # Ensure that len(timestamps) is an integer multiple of len(targets)
+    timestamps = 1234667890.0 + 1.0 * np.arange(12)
+    return MinimalDataSet(targets, timestamps)
+
+
+def test_selecting_antenna(dataset):
+    dataset.select(ants='~m000')
+    assert_array_equal(
+        dataset.corr_products,
+        [('m063h', 'm063h'), ('m063v', 'm063v')])
+
+
+@pytest.mark.parametrize(
+    'scans,expected_dumps',
+    [
+            ('track', [1, 2, 4, 5, 7, 8, 10, 11]),
+            ('slew', [0, 3, 6, 9]),
+            ('~slew', [1, 2, 4, 5, 7, 8, 10, 11]),
+            (('track', 'slew'), np.arange(12)),
+            (('track', '~slew'), [1, 2, 4, 5, 7, 8, 10, 11]),
+            (1, [1, 2]),
+            ([1, 3, 4], [1, 2, 4, 5, 6]),
+            # Empty selections
+            ('scan', []),
+            (-1, []),
+            (100, []),
+    ]
+)
+def test_select_scans(dataset, scans, expected_dumps):
+    dataset.select(scans=scans)
+    assert_array_equal(dataset.dumps, expected_dumps)
+
+
+@pytest.mark.parametrize(
+    'compscans,expected_dumps',
+    [
+            ('track', np.arange(12)),
+            (1, [3, 4, 5]),
+            ([1, 3], [3, 4, 5, 9, 10, 11]),
+            # Empty selections
+            ('~track', []),
+            (-1, []),
+            (100, []),
+    ]
+)
+def test_select_compscans(dataset, compscans, expected_dumps):
+    dataset.select(compscans=compscans)
+    assert_array_equal(dataset.dumps, expected_dumps)
+
+
+@pytest.mark.parametrize(
+    'targets,expected_dumps',
+    [
+        ('PKS1934-638', [0, 1, 2]),
+        ('J1939-6342', [0, 1, 2, 3, 4, 5]),
+        (('J0408-6545', 'Cen B'), [6, 7, 8, 9, 10, 11]),
+        ('J1939-6342, radec gaincal, 19:39:25.03, -63:42:45.6', [3, 4, 5]),
+        (Target('J1939-6342, radec gaincal, 19:39:25.03, -63:42:45.6'), [3, 4, 5]),
+        (
+            [
+                Target('J1939-6342, radec gaincal, 19:39:25.03, -63:42:45.6'),
+                'J1346-6024 | Cen B, radec, 13:46:49.04, -60:24:29.4',
+            ],
+            [3, 4, 5, 9, 10, 11]
+        ),
+        (1, [3, 4, 5]),
+        ([1, 3], [3, 4, 5, 9, 10, 11]),
+        # Empty selections
+        ('Moon', []),
+        ('J1939-6342, radec gaincal, 19:39, -63:42', []),
+        (Target('Sun, special'), []),
+    ]
+)
+def test_select_targets(caplog, dataset, targets, expected_dumps):
+    with caplog.at_level(logging.WARNING, logger='katdal.dataset'):
+        dataset.select(targets=targets)
+    # If the target is not found, check that a warning is logged
+    if len(expected_dumps) == 0:
+        assert 'Skipping unknown selected target' in caplog.text
+    assert_array_equal(dataset.dumps, expected_dumps)