Remove test_store.py

xcube-dev · b-yogesh · Jan 21, 2025 · Nov 4, 2024 · Nov 7, 2024 · Nov 7, 2024
commit da3928e60af2e24ba55644824b19a2273d149ae6
diff --git a/test/test_store.py b/test/test_store.py
diff --git a/xcube_clms/api_token_handler.py b/xcube_clms/api_token_handler.py
                LOG.error("Token refresh failed: ", e)
                raise e
        else:
            LOG.info("Current token valid. Reusing it.")

        return self.api_token

diff --git a/xcube_clms/clms.py b/xcube_clms/clms.py
                f"got {data_id}"
            )
        if not self.file_store:
            raise ValueError(
                "File store does not exist yet. Please preload "
                "data first using the preload_data() method."
            )

        data_id_file = os.listdir(cache_entry)
        if len(data_id_file) != 1:
            LOG.warning(
                f"Expected 1 file in the folder {cache_entry}, "
                f"got {len(data_id_file)}. Opening the first file."
            )
        time_range = (item.get(START_TIME_KEY), item.get(END_TIME_KEY))

        if len(crs) > 1:
            LOG.warning(
                f"Expected 1 crs, got {len(crs)}. Outputting the first element."
            )

        Raises:
            ValueError: If any data ID is invalid.
        """
        data_id_maps = {
            data_id: {
                ITEM_KEY: self._access_item(data_id),
                PRODUCT_KEY: self._access_item(data_id.split(DATA_ID_SEPARATOR)[0]),
            }
            for data_id in data_ids
        }
        self._preload_data.initiate_preload(data_id_maps)

    def _create_data_ids(
        self,
        """
        dataset = self._get_item(data_id)
        if len(dataset) != 1:
            raise ValueError(
                f"Expected one dataset for data_id: {data_id}, found"
                f" {len(dataset)}."
            )
diff --git a/xcube_clms/constants.py b/xcube_clms/constants.py
 LEVEL = logging.INFO
 LOG.setLevel(LEVEL)
 if not LOG.hasHandlers():
    handler = logging.StreamHandler()
    handler.setLevel(LEVEL)
    formatter = logging.Formatter(
        "%(name)s - %(asctime)s -  %(levelname)s - %(message)s"
    )
    handler.setFormatter(formatter)
    LOG.addHandler(handler)

 # DataOpener IDs
 DATA_OPENER_IDS = (
diff --git a/xcube_clms/download_manager.py b/xcube_clms/download_manager.py
        Raises:
            AssertionError: If the dataset is unsupported.
        """
        self._token_handler.refresh_token()

        # This is to make sure that there are pre-packaged files available for
        # download. Without this, the API throws the following error: Error,
        # We check for path and source based on the API code here:
        # https://github.com/eea/clms.downloadtool/blob/master/clms/downloadtool/api/services/datarequest_post/post.py#L177-L196

        path = item.get(PATH_KEY, "")
        source = item.get(SOURCE_KEY, "")

        if (path == "") and (source == ""):
            LOG.warning(f"No prepackaged downloadable items available for {data_id}")

        # This is to make sure that we do not send requests for currently for
        # unsupported datasets.

        full_source = (
            product.get(DATASET_DOWNLOAD_INFORMATION_KEY)
            .get(ITEMS_KEY)[0]
            .get(FULL_SOURCE_KEY, "")
        )

        assert full_source not in NOT_SUPPORTED_LIST, (
            f"This data product: {product[TITLE_KEY]} is not yet supported in "
            f"this plugin yet."
        )

        status, task_id = self.get_current_requests_status(
            dataset_id=product[UID_KEY], file_id=item[ID_KEY]
        )

        if status == COMPLETE or status == PENDING:
            LOG.info(
                f"Download request with task id {task_id} "
                f"{'already completed' if status == 'COMPLETE' 
                else 'is in queue'} for data id: {data_id}"
            )
            return task_id
        if status == UNDEFINED:
            LOG.info(
                f"Download request does not exists or has expired for "
                f"data id:"
                f" {data_id}"
            )

        if status == CANCELLED:
            LOG.info(
                f"Download request was cancelled for "
                f"data id:"
                f" {data_id}. Re-requesting now."
            )

        download_request_url, headers, json = self._prepare_download_request(
            data_id, item, product
        )

        response_data = make_api_request(
            method="POST", url=download_request_url, headers=headers, json=json
        )
        response = get_response_of_type(response_data, JSON_TYPE)
        task_ids = response.get(TASK_IDS_KEY)
        assert (
            len(task_ids) == 1
        ), f"Expected API response with 1 task_id, got {len(task_ids)}"
        task_id = task_ids[0].get(TASK_ID_KEY)
        LOG.info(f"Download Requested with Task ID : {task_id}")
        return task_id

    def get_download_url(self, task_id: str) -> tuple[str, int]:
        """
        Raises:
            Exception: If the task has not completed or no download URL is available.
        """
        self._token_handler.refresh_token()

        headers = ACCEPT_HEADER.copy()
        headers.update(get_authorization_header(self._api_token))

        url = build_api_url(self._url, TASK_STATUS_ENDPOINT, datasets_request=False)
        response_data = make_api_request(url=url, headers=headers)
        response = get_response_of_type(response_data, JSON_TYPE)

        for key in response:
            if key == task_id:
                status = response[key][STATUS_KEY]
                if status in STATUS_COMPLETE:
                    return (
                        response[key][DOWNLOAD_URL_KEY],
                        response[key]["FileSize"],
                    )
                else:
                    raise Exception(
                        f"Task ID {task_id} has not yet finished. No download url available yet."
                    )

            tuple[str, dict, dict]: A tuple containing the request URL,
                headers, and JSON payload.
        """
        LOG.info(f"Preparing download request for {data_id}")

        dataset_id = product[UID_KEY]
        file_id = item[ID_KEY]
        json = get_dataset_download_info(
            dataset_id=dataset_id,
            file_id=file_id,
        )
        url = build_api_url(self._url, DOWNLOAD_ENDPOINT, datasets_request=False)
        if not self._api_token:
            self._token_handler.refresh_token()
        headers = ACCEPT_HEADER.copy()
        headers.update(CONTENT_TYPE_HEADER)
        headers.update(get_authorization_header(self._api_token))

        return url, headers, json

    def get_current_requests_status(
        self,

        Notes:
        """
        self._token_handler.refresh_token()
        headers = ACCEPT_HEADER.copy()
        headers.update(get_authorization_header(self._api_token))

        url = build_api_url(self._url, TASK_STATUS_ENDPOINT, datasets_request=False)
        response_data = make_api_request(url=url, headers=headers)
        response = get_response_of_type(response_data, JSON_TYPE)

        status_priority = {
            "Finished_ok": 1,  # Complete
            "Queued": 2,  # Pending
            "In_progress": 2,  # Pending
            "Cancelled": 3,  # Cancelled
        }

        latest_entries = {status: {} for status in status_priority.keys()}

        for key in response:
            status = response[key][STATUS_KEY]
            datasets = response[key][DATASETS_KEY]
            requested_data_id = datasets[0][DATASET_ID_KEY]
            requested_file_id = datasets[0][FILE_ID_KEY]
            timestamp = (
                response[key][DOWNLOAD_AVAILABLE_TIME_KEY]
                if status in {"Finished_ok", "Cancelled"}
                else None
            )  # Only get timestamp for Completed or Cancelled
            condition = (
                ((dataset_id == requested_data_id) and (file_id == requested_file_id))
                if dataset_id
                else (key == task_id)
            )
            if condition:
                current_entry = {
                    "status": status,
                    "key": key,
                    "timestamp": timestamp,
                    "response": response[key],
                }

                if status in latest_entries:
                    existing_entry = latest_entries[status]
                    if not existing_entry or (
                        timestamp and (timestamp > existing_entry.get("timestamp", ""))
                    ):
                        latest_entries[status] = current_entry

        for status in sorted(
            status_priority, key=lambda s: status_priority[s], reverse=False
        ):
            latest_entry = latest_entries[status]
            if latest_entry:
                key = latest_entry["key"]
                entry_response = latest_entry["response"]
                if status in STATUS_COMPLETE:
                    if not has_expired(entry_response[DOWNLOAD_AVAILABLE_TIME_KEY]):
                        return COMPLETE, key
                elif status in STATUS_PENDING:
                    return PENDING, key
                elif status in STATUS_CANCELLED:
                    return CANCELLED, key

        return UNDEFINED, ""

    def download_data(
        self, download_url: str, file_size: int, task_id: str, data_id: str
            task_id: Task ID associated with the dataset.
            data_id: Unique identifier of the dataset.
        """
        LOG.info(f"Downloading zip file from {download_url}")

        response = make_api_request(download_url, timeout=600, stream=True)
        chunk_size = 1024 * 1024  # 1 MB chunks

        with tempfile.NamedTemporaryFile(mode="wb", delete=True) as temp_file:
            temp_file_path = temp_file.name
            LOG.info(f"Temporary file created at {temp_file_path}")

            progress_bar = tqdm(
                response.iter_content(chunk_size=chunk_size),
                desc=f"Downloading zip file for task: {task_id}",
                total=file_size // chunk_size,
                unit_scale=True,
            )

            for chunk in progress_bar:
                temp_file.write(chunk)
                progress_bar.update(len(chunk))
                del chunk
            progress_bar.close()

            fs = fsspec.filesystem("zip", fo=temp_file_path)
            zip_contents = fs.ls(RESULTS)
            actual_zip_file = None
            if len(zip_contents) == 1:
                if ".zip" in zip_contents[0][FILENAME_KEY]:
                    actual_zip_file = zip_contents[0]
            elif len(zip_contents) > 1:
                LOG.warn("Cannot handle more than one zip files at the moment.")
            else:
                LOG.warn("No downloadable zip file found inside.")
            if actual_zip_file:
                LOG.info(
                    f"Found one zip file "
                    f"{actual_zip_file.get(ORIGINAL_FILENAME_KEY)}."
                )
                with fs.open(actual_zip_file[NAME_KEY], "rb") as f:
                    zip_fs = fsspec.filesystem("zip", fo=f)

                    geo_files = DownloadTaskManager._find_geo_in_dir(
                        "/",
                        zip_fs,
                    )
                    if geo_files:
                        target_folder = os.path.join(self.path, data_id + "/").__str__()
                        os.makedirs(
                            os.path.dirname(target_folder),
                            exist_ok=True,
                        )
                        for geo_file in tqdm(
                            geo_files,
                            desc="Extracting geo files for task_id {task_id}",
                        ):
                            try:
                                with zip_fs.open(geo_file, "rb") as source_file:
                                    geo_file_name = geo_file.split("/")[-1]
                                    geo_file_path = os.path.join(
                                        target_folder, geo_file_name
                                    )
                                    with open(
                                        geo_file_path,
                                        "wb",
                                    ) as dest_file:
                                        for chunk in tqdm(
                                            iter(
                                                lambda: source_file.read(chunk_size),
                                                b"",
                                            ),
                                            desc=f"Extracting geo file {geo_file_name}",
                                        ):
                                            dest_file.write(chunk)
                                LOG.info(
                                    f"The file {geo_file_name} has been successfully "
                                    f"downloaded to {geo_file_path}"
                                )
                            except OSError as e:
                                raise OSError(f"Error occurred while writing data. {e}")
                            except UnicodeDecodeError as e:
                                raise ValueError(
                                    f"Decoding error: {e}. File might not be text "
                                    f"or encoding is incorrect."
                                )
                            except Exception as e:
                                raise Exception(f"An unexpected error occurred: {e}")

                    else:
                        raise FileNotFoundError(
                            "No file found in the downloaded zip file to load"
                        )

                in `GEO_FILE_EXTS`.
            - Logs details of each discovered geo file.
        """
        geo_file: list[str] = []
        contents = zip_fs.ls(path)
        for item in contents:
            if zip_fs.isdir(item[NAME_KEY]):
                geo_file.extend(
                    DownloadTaskManager._find_geo_in_dir(
                        item[NAME_KEY],
                        zip_fs,
                    )
                )
            else:
                if item[NAME_KEY].endswith(GEO_FILE_EXTS):
                    LOG.info(f"Found geo file: {item[NAME_KEY]}")
                    filename = item[NAME_KEY]
                    geo_file.append(filename)
        return geo_file
diff --git a/xcube_clms/preload.py b/xcube_clms/preload.py
                we use the item and product keys mapped to their data ids to
                preload the data.
        """
        self.refresh_cache()
        # We create a status event for each of the tasks so that we can
        # signal the thread that it can proceed further from the queue.
        for data_id_map_key in data_id_maps.keys():
            self._task_control[data_id_map_key] = {
                "status_event": threading.Event(),
            }

        executor = ThreadPoolExecutor()
        for data_id_map in data_id_maps.items():
            executor.submit(
                self._initiate_preload,
                data_id_map,
                self._task_control[data_id_map[0]]["status_event"],
            status_event: Event to manage task status and the visibility of
                the spinner.
        """
        data_id = data_id_map[0]
        if data_id in self.view_cache().keys():
            LOG.info(f"The data for {data_id} is already cached at {self.path}")
            return

        task_id = self._download_manager.request_download(
            data_id=data_id,
            item=data_id_map[1].get("item"),
            product=data_id_map[1].get("product"),
        )

        spinner_thread = threading.Thread(
            target=spinner, args=(status_event, f"{task_id}")
        )
        status_event.set()
        spinner_thread.start()

        while status_event.is_set():
            status, _ = self._download_manager.get_current_requests_status(
                task_id=task_id
            )
            if status == COMPLETE:
                LOG.info(f"Status: {status} for {data_id} with task ID {task_id}.")
                status_event.clear()
                spinner_thread.join()
                download_url, file_size = self._download_manager.get_download_url(
                    task_id
                )
                self.download_data(download_url, file_size, task_id, data_id)
                self._file_processor.postprocess(data_id)
            if status in PENDING:
                LOG.info(
                    f"Status: {status} for {data_id} with task ID {task_id}. Will "
                    f"recheck status in "
                    f"{RETRY_TIMEOUT} seconds"
                )
            if status in CANCELLED:
                LOG.info(
                    f"Status: {status} for {data_id} with task ID {task_id}. Exiting now"
                )
                status_event.clear()
                spinner_thread.join()
                return

            time.sleep(RETRY_TIMEOUT)

    def download_data(
        self, download_url: str, file_size: int, task_id: str, data_id: str
            task_id: Task ID of the download request.
            data_id: Identifier of the data being downloaded.
        """
        self._download_manager.download_data(download_url, file_size, task_id, data_id)

    def view_cache(self) -> dict[str, str]:
        """
        Returns:
            dict: Cached map.
        """
        return self._cache_manager.cache

    def refresh_cache(self) -> None:
        """
        Refreshes the cache map.
        """
        self._cache_manager.refresh_cache()
diff --git a/xcube_clms/processor.py b/xcube_clms/processor.py
                    "supported."
                )
                return
            self._merge_and_save(en_map, data_id)
            if self.cleanup:
                cleanup_dir(target_folder)

    def _prepare_merge(
        self, files: list[str], data_id: str
diff --git a/xcube_clms/store.py b/xcube_clms/store.py
 from typing import Tuple, Iterator, Container, Any, Union

 import xarray as xr
 from xcube.core.store import (
    DataDescriptor,
    DataStore,
    DataTypeLike,
    JsonBooleanSchema,
 )

 from .clms import Clms
 from .constants import DATA_OPENER_IDS
 from .utils import assert_valid_data_type


 class ClmsDataStore(DataStore, ABC):
    """CLMS implementation of the data store defined in the ``xcube_clms``
    plugin."""

    def __init__(self, **clms_kwargs):
        self._clms = Clms(**clms_kwargs)

    @classmethod
    def get_data_store_params_schema(cls) -> JsonObjectSchema:
        credentials_params = dict(
            client_id=JsonStringSchema(),
            issued=JsonStringSchema(),
            private_key=JsonStringSchema(),

    @classmethod
    def get_data_types(cls) -> Tuple[str, ...]:
        return (DATASET_TYPE.alias,)

    def get_data_types_for_data(self, data_id: str) -> Tuple[str, ...]:
        return self.get_data_types()

    def get_data_ids(
        self,
        data_type: DataTypeLike = None,
        include_attrs: Container[str] | bool | None = None,
    ) -> Union[Iterator[str], Iterator[tuple[str, dict[str, Any]]]]:
        assert_valid_data_type(data_type)
        data_ids = self._clms.get_data_ids(include_attrs)
        for data_id in data_ids:
            if ((include_attrs is not None) and (include_attrs != False)) or (
                include_attrs == True
            ):
                yield data_id[0], data_id[1]
            else:
                yield data_id

    def has_data(self, data_id: str, data_type: DataTypeLike = None) -> bool:
        return self._clms.has_data(data_id, data_type)

    def describe_data(
        self, data_id: str, data_type: DataTypeLike = None
    ) -> DataDescriptor:
        assert_valid_data_type(data_type)
        metadata = self._clms.get_extent(data_id)
        return DatasetDescriptor(data_id, **metadata)

    def get_data_opener_ids(
        self, data_id: str = None, data_type: DataTypeLike = None
    ) -> Tuple[str, ...]:
        return DATA_OPENER_IDS

    def get_open_data_params_schema(
        self, data_id: str = None, opener_id: str = None
    ) -> JsonObjectSchema:
        # We do not support any open_data_params yet
        return JsonObjectSchema()

    def open_data(
        self,
        resolution: str = "",
        **open_params,
    ) -> xr.Dataset:
        return self._clms.open_data(data_id)

    def search_data(
        self, data_type: DataTypeLike = None, **search_params
    ) -> JsonObjectSchema:
        pass

    def preload_data(self, *data_ids: str, **preload_params):
        schema = self.get_preload_params_schema()
        schema.validate_instance(preload_params)
        return self._clms.preload_data(*data_ids, **preload_params)

    @classmethod
    def get_preload_params_schema(cls):
        return JsonObjectSchema(additional_properties=False)
diff --git a/xcube_clms/utils.py b/xcube_clms/utils.py
        DataStoreError: Error, if *data_type* is not
            supported by the store.
    """
    if not is_valid_data_type(data_type):
        raise DataStoreError(
            f"Data type must be {DATASET_TYPE.alias!r} or but got {data_type!r}."
        )

    Returns:
        True if *data_type* is supported by the store, otherwise False
    """
    return data_type is None or DATASET_TYPE.is_super_type_of(data_type)


 ResponseType = Literal["json", "text", "bytes"]
	LOG.error("Token refresh failed: ", e)
	raise e
	else:
	LOG.info("Current token valid. Reusing it.")
Check warning on line 77 in xcube_clms/api_token_handler.py Codecov / codecov/patch xcube_clms/api_token_handler.py#L77 `Added line #L77 was not covered by tests`

	return self.api_token
	f"got {data_id}"
	)
	if not self.file_store:
	raise ValueError(
Check warning on line 118 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L118 `Added line #L118 was not covered by tests`
	"File store does not exist yet. Please preload "
	"data first using the preload_data() method."
	)

	data_id_file = os.listdir(cache_entry)
	if len(data_id_file) != 1:
	LOG.warning(
Check warning on line 130 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L130 `Added line #L130 was not covered by tests`
	f"Expected 1 file in the folder {cache_entry}, "
	f"got {len(data_id_file)}. Opening the first file."
	)
	time_range = (item.get(START_TIME_KEY), item.get(END_TIME_KEY))

	if len(crs) > 1:
	LOG.warning(
Check warning on line 189 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L189 `Added line #L189 was not covered by tests`
	f"Expected 1 crs, got {len(crs)}. Outputting the first element."
	)

	Raises:
	ValueError: If any data ID is invalid.
	"""
	data_id_maps = {
Check warning on line 211 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L211 `Added line #L211 was not covered by tests`
	data_id: {
	ITEM_KEY: self._access_item(data_id),
	PRODUCT_KEY: self._access_item(data_id.split(DATA_ID_SEPARATOR)[0]),
	}
	for data_id in data_ids
	}
	self._preload_data.initiate_preload(data_id_maps)
Check warning on line 218 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L218 `Added line #L218 was not covered by tests`

	def _create_data_ids(
	self,
	"""
	dataset = self._get_item(data_id)
	if len(dataset) != 1:
	raise ValueError(
Check warning on line 296 in xcube_clms/clms.py Codecov / codecov/patch xcube_clms/clms.py#L296 `Added line #L296 was not covered by tests`
	f"Expected one dataset for data_id: {data_id}, found"
	f" {len(dataset)}."
	)
	LEVEL = logging.INFO
	LOG.setLevel(LEVEL)
	if not LOG.hasHandlers():
	handler = logging.StreamHandler()
	handler.setLevel(LEVEL)
	formatter = logging.Formatter(
Check warning on line 113 in xcube_clms/constants.py Codecov / codecov/patch xcube_clms/constants.py#L111-L113 `Added lines #L111 - L113 were not covered by tests`
	"%(name)s - %(asctime)s - %(levelname)s - %(message)s"
	)
	handler.setFormatter(formatter)
	LOG.addHandler(handler)
Check warning on line 117 in xcube_clms/constants.py Codecov / codecov/patch xcube_clms/constants.py#L116-L117 `Added lines #L116 - L117 were not covered by tests`

	# DataOpener IDs
	DATA_OPENER_IDS = (
	Raises:
	AssertionError: If the dataset is unsupported.
	"""
	self._token_handler.refresh_token()
Check warning on line 114 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L114 `Added line #L114 was not covered by tests`

	# This is to make sure that there are pre-packaged files available for
	# download. Without this, the API throws the following error: Error,
	# We check for path and source based on the API code here:
	# https://github.com/eea/clms.downloadtool/blob/master/clms/downloadtool/api/services/datarequest_post/post.py#L177-L196

	path = item.get(PATH_KEY, "")
	source = item.get(SOURCE_KEY, "")
Check warning on line 123 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L122-L123 `Added lines #L122 - L123 were not covered by tests`

	if (path == "") and (source == ""):
	LOG.warning(f"No prepackaged downloadable items available for {data_id}")
Check warning on line 126 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L125-L126 `Added lines #L125 - L126 were not covered by tests`

	# This is to make sure that we do not send requests for currently for
	# unsupported datasets.

	full_source = (
Check warning on line 131 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L131 `Added line #L131 was not covered by tests`
	product.get(DATASET_DOWNLOAD_INFORMATION_KEY)
	.get(ITEMS_KEY)[0]
	.get(FULL_SOURCE_KEY, "")
	)

	assert full_source not in NOT_SUPPORTED_LIST, (
Check warning on line 137 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L137 `Added line #L137 was not covered by tests`
	f"This data product: {product[TITLE_KEY]} is not yet supported in "
	f"this plugin yet."
	)

	status, task_id = self.get_current_requests_status(
Check warning on line 142 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L142 `Added line #L142 was not covered by tests`
	dataset_id=product[UID_KEY], file_id=item[ID_KEY]
	)

	if status == COMPLETE or status == PENDING:
	LOG.info(
Check warning on line 147 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L146-L147 `Added lines #L146 - L147 were not covered by tests`
	f"Download request with task id {task_id} "
	f"{'already completed' if status == 'COMPLETE'
	else 'is in queue'} for data id: {data_id}"
	)
	return task_id
	if status == UNDEFINED:
	LOG.info(
Check warning on line 154 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L152-L154 `Added lines #L152 - L154 were not covered by tests`
	f"Download request does not exists or has expired for "
	f"data id:"
	f" {data_id}"
	)

	if status == CANCELLED:
	LOG.info(
Check warning on line 161 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L160-L161 `Added lines #L160 - L161 were not covered by tests`
	f"Download request was cancelled for "
	f"data id:"
	f" {data_id}. Re-requesting now."
	)

	download_request_url, headers, json = self._prepare_download_request(
Check warning on line 167 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L167 `Added line #L167 was not covered by tests`
	data_id, item, product
	)

	response_data = make_api_request(
Check warning on line 171 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L171 `Added line #L171 was not covered by tests`
	method="POST", url=download_request_url, headers=headers, json=json
	)
	response = get_response_of_type(response_data, JSON_TYPE)
	task_ids = response.get(TASK_IDS_KEY)
	assert (
Check warning on line 176 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L174-L176 `Added lines #L174 - L176 were not covered by tests`
	len(task_ids) == 1
	), f"Expected API response with 1 task_id, got {len(task_ids)}"
	task_id = task_ids[0].get(TASK_ID_KEY)
	LOG.info(f"Download Requested with Task ID : {task_id}")
	return task_id
Check warning on line 181 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L179-L181 `Added lines #L179 - L181 were not covered by tests`

	def get_download_url(self, task_id: str) -> tuple[str, int]:
	"""
	Raises:
	Exception: If the task has not completed or no download URL is available.
	"""
	self._token_handler.refresh_token()
Check warning on line 196 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L196 `Added line #L196 was not covered by tests`

	headers = ACCEPT_HEADER.copy()
	headers.update(get_authorization_header(self._api_token))
Check warning on line 199 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L198-L199 `Added lines #L198 - L199 were not covered by tests`

	url = build_api_url(self._url, TASK_STATUS_ENDPOINT, datasets_request=False)
	response_data = make_api_request(url=url, headers=headers)
	response = get_response_of_type(response_data, JSON_TYPE)
Check warning on line 203 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L201-L203 `Added lines #L201 - L203 were not covered by tests`

	for key in response:
	if key == task_id:
	status = response[key][STATUS_KEY]
	if status in STATUS_COMPLETE:
	return (
Check warning on line 209 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L205-L209 `Added lines #L205 - L209 were not covered by tests`
	response[key][DOWNLOAD_URL_KEY],
	response[key]["FileSize"],
	)
	else:
	raise Exception(
Check warning on line 214 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L214 `Added line #L214 was not covered by tests`
	f"Task ID {task_id} has not yet finished. No download url available yet."
	)

	tuple[str, dict, dict]: A tuple containing the request URL,
	headers, and JSON payload.
	"""
	LOG.info(f"Preparing download request for {data_id}")
Check warning on line 233 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L233 `Added line #L233 was not covered by tests`

	dataset_id = product[UID_KEY]
	file_id = item[ID_KEY]
	json = get_dataset_download_info(
Check warning on line 237 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L235-L237 `Added lines #L235 - L237 were not covered by tests`
	dataset_id=dataset_id,
	file_id=file_id,
	)
	url = build_api_url(self._url, DOWNLOAD_ENDPOINT, datasets_request=False)
	if not self._api_token:
	self._token_handler.refresh_token()
	headers = ACCEPT_HEADER.copy()
	headers.update(CONTENT_TYPE_HEADER)
	headers.update(get_authorization_header(self._api_token))
Check warning on line 246 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L241-L246 `Added lines #L241 - L246 were not covered by tests`

	return url, headers, json
Check warning on line 248 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L248 `Added line #L248 was not covered by tests`

	def get_current_requests_status(
	self,

	Notes:
	"""
	self._token_handler.refresh_token()
	headers = ACCEPT_HEADER.copy()
	headers.update(get_authorization_header(self._api_token))
Check warning on line 277 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L275-L277 `Added lines #L275 - L277 were not covered by tests`

	url = build_api_url(self._url, TASK_STATUS_ENDPOINT, datasets_request=False)
	response_data = make_api_request(url=url, headers=headers)
	response = get_response_of_type(response_data, JSON_TYPE)
Check warning on line 281 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L279-L281 `Added lines #L279 - L281 were not covered by tests`

	status_priority = {
Check warning on line 283 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L283 `Added line #L283 was not covered by tests`
	"Finished_ok": 1, # Complete
	"Queued": 2, # Pending
	"In_progress": 2, # Pending
	"Cancelled": 3, # Cancelled
	}

	latest_entries = {status: {} for status in status_priority.keys()}
Check warning on line 290 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L290 `Added line #L290 was not covered by tests`

	for key in response:
	status = response[key][STATUS_KEY]
	datasets = response[key][DATASETS_KEY]
	requested_data_id = datasets[0][DATASET_ID_KEY]
	requested_file_id = datasets[0][FILE_ID_KEY]
	timestamp = (
Check warning on line 297 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L292-L297 `Added lines #L292 - L297 were not covered by tests`
	response[key][DOWNLOAD_AVAILABLE_TIME_KEY]
	if status in {"Finished_ok", "Cancelled"}
	else None
	) # Only get timestamp for Completed or Cancelled
	condition = (
Check warning on line 302 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L302 `Added line #L302 was not covered by tests`
	((dataset_id == requested_data_id) and (file_id == requested_file_id))
	if dataset_id
	else (key == task_id)
	)
	if condition:
	current_entry = {
Check warning on line 308 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L307-L308 `Added lines #L307 - L308 were not covered by tests`
	"status": status,
	"key": key,
	"timestamp": timestamp,
	"response": response[key],
	}

	if status in latest_entries:
	existing_entry = latest_entries[status]
	if not existing_entry or (
Check warning on line 317 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L315-L317 `Added lines #L315 - L317 were not covered by tests`
	timestamp and (timestamp > existing_entry.get("timestamp", ""))
	):
	latest_entries[status] = current_entry
Check warning on line 320 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L320 `Added line #L320 was not covered by tests`

	for status in sorted(
Check warning on line 322 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L322 `Added line #L322 was not covered by tests`
	status_priority, key=lambda s: status_priority[s], reverse=False
	):
	latest_entry = latest_entries[status]
	if latest_entry:
	key = latest_entry["key"]
	entry_response = latest_entry["response"]
	if status in STATUS_COMPLETE:
	if not has_expired(entry_response[DOWNLOAD_AVAILABLE_TIME_KEY]):
	return COMPLETE, key
	elif status in STATUS_PENDING:
	return PENDING, key
	elif status in STATUS_CANCELLED:
	return CANCELLED, key
Check warning on line 335 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L325-L335 `Added lines #L325 - L335 were not covered by tests`

	return UNDEFINED, ""
Check warning on line 337 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L337 `Added line #L337 was not covered by tests`

	def download_data(
	self, download_url: str, file_size: int, task_id: str, data_id: str
	task_id: Task ID associated with the dataset.
	data_id: Unique identifier of the dataset.
	"""
	LOG.info(f"Downloading zip file from {download_url}")
Check warning on line 352 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L352 `Added line #L352 was not covered by tests`

	response = make_api_request(download_url, timeout=600, stream=True)
	chunk_size = 1024 * 1024 # 1 MB chunks
Check warning on line 355 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L354-L355 `Added lines #L354 - L355 were not covered by tests`

	with tempfile.NamedTemporaryFile(mode="wb", delete=True) as temp_file:
	temp_file_path = temp_file.name
	LOG.info(f"Temporary file created at {temp_file_path}")
Check warning on line 359 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L357-L359 `Added lines #L357 - L359 were not covered by tests`

	progress_bar = tqdm(
Check warning on line 361 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L361 `Added line #L361 was not covered by tests`
	response.iter_content(chunk_size=chunk_size),
	desc=f"Downloading zip file for task: {task_id}",
	total=file_size // chunk_size,
	unit_scale=True,
	)

	for chunk in progress_bar:
	temp_file.write(chunk)
	progress_bar.update(len(chunk))
	del chunk
	progress_bar.close()
Check warning on line 372 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L368-L372 `Added lines #L368 - L372 were not covered by tests`

	fs = fsspec.filesystem("zip", fo=temp_file_path)
	zip_contents = fs.ls(RESULTS)
	actual_zip_file = None
	if len(zip_contents) == 1:
	if ".zip" in zip_contents[0][FILENAME_KEY]:
	actual_zip_file = zip_contents[0]
	elif len(zip_contents) > 1:
	LOG.warn("Cannot handle more than one zip files at the moment.")
Check warning on line 381 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L374-L381 `Added lines #L374 - L381 were not covered by tests`
	else:
	LOG.warn("No downloadable zip file found inside.")
	if actual_zip_file:
	LOG.info(
Check warning on line 385 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L383-L385 `Added lines #L383 - L385 were not covered by tests`
	f"Found one zip file "
	f"{actual_zip_file.get(ORIGINAL_FILENAME_KEY)}."
	)
	with fs.open(actual_zip_file[NAME_KEY], "rb") as f:
	zip_fs = fsspec.filesystem("zip", fo=f)
Check warning on line 390 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L389-L390 `Added lines #L389 - L390 were not covered by tests`

	geo_files = DownloadTaskManager._find_geo_in_dir(
Check warning on line 392 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L392 `Added line #L392 was not covered by tests`
	"/",
	zip_fs,
	)
	if geo_files:
	target_folder = os.path.join(self.path, data_id + "/").__str__()
	os.makedirs(
Check warning on line 398 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L396-L398 `Added lines #L396 - L398 were not covered by tests`
	os.path.dirname(target_folder),
	exist_ok=True,
	)
	for geo_file in tqdm(
Check warning on line 402 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L402 `Added line #L402 was not covered by tests`
	geo_files,
	desc="Extracting geo files for task_id {task_id}",
	):
	try:
	with zip_fs.open(geo_file, "rb") as source_file:
	geo_file_name = geo_file.split("/")[-1]
	geo_file_path = os.path.join(
Check warning on line 409 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L406-L409 `Added lines #L406 - L409 were not covered by tests`
	target_folder, geo_file_name
	)
	with open(
Check warning on line 412 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L412 `Added line #L412 was not covered by tests`
	geo_file_path,
	"wb",
	) as dest_file:
	for chunk in tqdm(
Check warning on line 416 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L416 `Added line #L416 was not covered by tests`
	iter(
	lambda: source_file.read(chunk_size),
	b"",
	),
	desc=f"Extracting geo file {geo_file_name}",
	):
	dest_file.write(chunk)
	LOG.info(
Check warning on line 424 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L423-L424 `Added lines #L423 - L424 were not covered by tests`
	f"The file {geo_file_name} has been successfully "
	f"downloaded to {geo_file_path}"
	)
	except OSError as e:
	raise OSError(f"Error occurred while writing data. {e}")
	except UnicodeDecodeError as e:
	raise ValueError(
Check warning on line 431 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L428-L431 `Added lines #L428 - L431 were not covered by tests`
	f"Decoding error: {e}. File might not be text "
	f"or encoding is incorrect."
	)
	except Exception as e:
	raise Exception(f"An unexpected error occurred: {e}")
Check warning on line 436 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L435-L436 `Added lines #L435 - L436 were not covered by tests`

	else:
	raise FileNotFoundError(
Check warning on line 439 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L439 `Added line #L439 was not covered by tests`
	"No file found in the downloaded zip file to load"
	)

	in `GEO_FILE_EXTS`.
	- Logs details of each discovered geo file.
	"""
	geo_file: list[str] = []
	contents = zip_fs.ls(path)
	for item in contents:
	if zip_fs.isdir(item[NAME_KEY]):
	geo_file.extend(
Check warning on line 466 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L462-L466 `Added lines #L462 - L466 were not covered by tests`
	DownloadTaskManager._find_geo_in_dir(
	item[NAME_KEY],
	zip_fs,
	)
	)
	else:
	if item[NAME_KEY].endswith(GEO_FILE_EXTS):
	LOG.info(f"Found geo file: {item[NAME_KEY]}")
	filename = item[NAME_KEY]
	geo_file.append(filename)
	return geo_file
Check warning on line 477 in xcube_clms/download_manager.py Codecov / codecov/patch xcube_clms/download_manager.py#L473-L477 `Added lines #L473 - L477 were not covered by tests`
	we use the item and product keys mapped to their data ids to
	preload the data.
	"""
	self.refresh_cache()
Check warning on line 92 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L92 `Added line #L92 was not covered by tests`
	# We create a status event for each of the tasks so that we can
	# signal the thread that it can proceed further from the queue.
	for data_id_map_key in data_id_maps.keys():
	self._task_control[data_id_map_key] = {
Check warning on line 96 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L95-L96 `Added lines #L95 - L96 were not covered by tests`
	"status_event": threading.Event(),
	}

	executor = ThreadPoolExecutor()
	for data_id_map in data_id_maps.items():
	executor.submit(
Check warning on line 102 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L100-L102 `Added lines #L100 - L102 were not covered by tests`
	self._initiate_preload,
	data_id_map,
	self._task_control[data_id_map[0]]["status_event"],
	status_event: Event to manage task status and the visibility of
	the spinner.
	"""
	data_id = data_id_map[0]
	if data_id in self.view_cache().keys():
	LOG.info(f"The data for {data_id} is already cached at {self.path}")
	return
Check warning on line 124 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L121-L124 `Added lines #L121 - L124 were not covered by tests`

	task_id = self._download_manager.request_download(
Check warning on line 126 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L126 `Added line #L126 was not covered by tests`
	data_id=data_id,
	item=data_id_map[1].get("item"),
	product=data_id_map[1].get("product"),
	)

	spinner_thread = threading.Thread(
Check warning on line 132 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L132 `Added line #L132 was not covered by tests`
	target=spinner, args=(status_event, f"{task_id}")
	)
	status_event.set()
	spinner_thread.start()
Check warning on line 136 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L135-L136 `Added lines #L135 - L136 were not covered by tests`

	while status_event.is_set():
	status, _ = self._download_manager.get_current_requests_status(
Check warning on line 139 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L138-L139 `Added lines #L138 - L139 were not covered by tests`
	task_id=task_id
	)
	if status == COMPLETE:
	LOG.info(f"Status: {status} for {data_id} with task ID {task_id}.")
	status_event.clear()
	spinner_thread.join()
	download_url, file_size = self._download_manager.get_download_url(
Check warning on line 146 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L142-L146 `Added lines #L142 - L146 were not covered by tests`
	task_id
	)
	self.download_data(download_url, file_size, task_id, data_id)
	self._file_processor.postprocess(data_id)
	if status in PENDING:
	LOG.info(
Check warning on line 152 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L149-L152 `Added lines #L149 - L152 were not covered by tests`
	f"Status: {status} for {data_id} with task ID {task_id}. Will "
	f"recheck status in "
	f"{RETRY_TIMEOUT} seconds"
	)
	if status in CANCELLED:
	LOG.info(
Check warning on line 158 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L157-L158 `Added lines #L157 - L158 were not covered by tests`
	f"Status: {status} for {data_id} with task ID {task_id}. Exiting now"
	)
	status_event.clear()
	spinner_thread.join()
	return
Check warning on line 163 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L161-L163 `Added lines #L161 - L163 were not covered by tests`

	time.sleep(RETRY_TIMEOUT)
Check warning on line 165 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L165 `Added line #L165 was not covered by tests`

	def download_data(
	self, download_url: str, file_size: int, task_id: str, data_id: str
	task_id: Task ID of the download request.
	data_id: Identifier of the data being downloaded.
	"""
	self._download_manager.download_data(download_url, file_size, task_id, data_id)
Check warning on line 181 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L181 `Added line #L181 was not covered by tests`

	def view_cache(self) -> dict[str, str]:
	"""
	Returns:
	dict: Cached map.
	"""
	return self._cache_manager.cache
Check warning on line 190 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L190 `Added line #L190 was not covered by tests`

	def refresh_cache(self) -> None:
	"""
	Refreshes the cache map.
	"""
	self._cache_manager.refresh_cache()
Check warning on line 196 in xcube_clms/preload.py Codecov / codecov/patch xcube_clms/preload.py#L196 `Added line #L196 was not covered by tests`
	"supported."
	)
	return
	self._merge_and_save(en_map, data_id)
	if self.cleanup:
	cleanup_dir(target_folder)
Check warning on line 85 in xcube_clms/processor.py Codecov / codecov/patch xcube_clms/processor.py#L83-L85 `Added lines #L83 - L85 were not covered by tests`

	def _prepare_merge(
	self, files: list[str], data_id: str
	from typing import Tuple, Iterator, Container, Any, Union

	import xarray as xr
	from xcube.core.store import (
Check warning on line 25 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L25 `Added line #L25 was not covered by tests`
	DataDescriptor,
	DataStore,
	DataTypeLike,
	JsonBooleanSchema,
	)

	from .clms import Clms
	from .constants import DATA_OPENER_IDS
	from .utils import assert_valid_data_type
Check warning on line 40 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L38-L40 `Added lines #L38 - L40 were not covered by tests`


	class ClmsDataStore(DataStore, ABC):
Check warning on line 43 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L43 `Added line #L43 was not covered by tests`
	"""CLMS implementation of the data store defined in the ``xcube_clms``
	plugin."""

	def __init__(self, **clms_kwargs):
	self._clms = Clms(**clms_kwargs)
Check warning on line 48 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L48 `Added line #L48 was not covered by tests`

	@classmethod
	def get_data_store_params_schema(cls) -> JsonObjectSchema:
	credentials_params = dict(
Check warning on line 52 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L52 `Added line #L52 was not covered by tests`
	client_id=JsonStringSchema(),
	issued=JsonStringSchema(),
	private_key=JsonStringSchema(),

	@classmethod
	def get_data_types(cls) -> Tuple[str, ...]:
	return (DATASET_TYPE.alias,)
Check warning on line 84 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L84 `Added line #L84 was not covered by tests`

	def get_data_types_for_data(self, data_id: str) -> Tuple[str, ...]:
	return self.get_data_types()
Check warning on line 87 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L87 `Added line #L87 was not covered by tests`

	def get_data_ids(
	self,
	data_type: DataTypeLike = None,
	include_attrs: Container[str] \| bool \| None = None,
	) -> Union[Iterator[str], Iterator[tuple[str, dict[str, Any]]]]:
	assert_valid_data_type(data_type)
	data_ids = self._clms.get_data_ids(include_attrs)
	for data_id in data_ids:
	if ((include_attrs is not None) and (include_attrs != False)) or (
Check warning on line 97 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L94-L97 `Added lines #L94 - L97 were not covered by tests`
	include_attrs == True
	):
	yield data_id[0], data_id[1]
Check warning on line 100 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L100 `Added line #L100 was not covered by tests`
	else:
	yield data_id
Check warning on line 102 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L102 `Added line #L102 was not covered by tests`

	def has_data(self, data_id: str, data_type: DataTypeLike = None) -> bool:
	return self._clms.has_data(data_id, data_type)
Check warning on line 105 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L104-L105 `Added lines #L104 - L105 were not covered by tests`

	def describe_data(
	self, data_id: str, data_type: DataTypeLike = None
	) -> DataDescriptor:
	assert_valid_data_type(data_type)
	metadata = self._clms.get_extent(data_id)
	return DatasetDescriptor(data_id, **metadata)
Check warning on line 112 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L110-L112 `Added lines #L110 - L112 were not covered by tests`

	def get_data_opener_ids(
	self, data_id: str = None, data_type: DataTypeLike = None
	) -> Tuple[str, ...]:
	return DATA_OPENER_IDS
Check warning on line 117 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L117 `Added line #L117 was not covered by tests`

	def get_open_data_params_schema(
	self, data_id: str = None, opener_id: str = None
	) -> JsonObjectSchema:
	# We do not support any open_data_params yet
	return JsonObjectSchema()
Check warning on line 123 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L123 `Added line #L123 was not covered by tests`

	def open_data(
	self,
	resolution: str = "",
	**open_params,
	) -> xr.Dataset:
	return self._clms.open_data(data_id)
Check warning on line 133 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L133 `Added line #L133 was not covered by tests`

	def search_data(
	self, data_type: DataTypeLike = None, **search_params
	) -> JsonObjectSchema:
	pass

	def preload_data(self, data_ids: str, *preload_params):
	schema = self.get_preload_params_schema()
	schema.validate_instance(preload_params)
	return self._clms.preload_data(data_ids, *preload_params)
Check warning on line 149 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L146-L149 `Added lines #L146 - L149 were not covered by tests`

	@classmethod
	def get_preload_params_schema(cls):
	return JsonObjectSchema(additional_properties=False)
Check warning on line 153 in xcube_clms/store.py Codecov / codecov/patch xcube_clms/store.py#L151-L153 `Added lines #L151 - L153 were not covered by tests`
	DataStoreError: Error, if data_type is not
	supported by the store.
	"""
	if not is_valid_data_type(data_type):
	raise DataStoreError(
Check warning on line 63 in xcube_clms/utils.py Codecov / codecov/patch xcube_clms/utils.py#L62-L63 `Added lines #L62 - L63 were not covered by tests`
	f"Data type must be {DATASET_TYPE.alias!r} or but got {data_type!r}."
	)

	Returns:
	True if data_type is supported by the store, otherwise False
	"""
	return data_type is None or DATASET_TYPE.is_super_type_of(data_type)
Check warning on line 78 in xcube_clms/utils.py Codecov / codecov/patch xcube_clms/utils.py#L78 `Added line #L78 was not covered by tests`


	ResponseType = Literal["json", "text", "bytes"]