22166837280

Committed 19 Feb 2026 02:47AM UTC coverage: 87.003% (-0.004%) from 87.007%

Build # 22166837280

Build Type

push

github

Committed by

web-flow

Commit Message

SNS: Move Tagging Functionality to Provider Methods (#13608)

Run Details

31 of 31 new or added lines in 1 file covered. (100.0%)

122 existing lines in 11 files now uncovered.

69763 of 80185 relevant lines covered (87.0%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.1

/localstack-core/localstack/utils/catalog/catalog_loader.py

import json
import logging
from json import JSONDecodeError
from pathlib import Path

import requests
from pydantic import BaseModel

from localstack import config, constants
from localstack.utils.catalog.common import AwsRemoteCatalog
from localstack.utils.http import get_proxies
from localstack.utils.json import FileMappedDocument

LOG = logging.getLogger(__name__)

AWS_CATALOG_FILE_NAME = "aws_catalog.json"


class RemoteCatalogVersionResponse(BaseModel):
    emulator_type: str
    version: str


class AwsCatalogLoaderException(Exception):
    def __init__(self, msg: str, *args):
        super().__init__(msg, *args)


class RemoteCatalogLoader:
    supported_schema_version = "v1"
    api_endpoint_catalog = f"{constants.API_ENDPOINT}/license/catalog"
    catalog_file_path = Path(config.dirs.cache) / AWS_CATALOG_FILE_NAME

    def get_remote_catalog(self) -> AwsRemoteCatalog:
        catalog_doc = FileMappedDocument(self.catalog_file_path)
        cached_catalog = AwsRemoteCatalog(**catalog_doc) if catalog_doc else None
        if cached_catalog:
            cached_catalog_version = cached_catalog.localstack.version
            if not self._should_update_cached_catalog(cached_catalog_version):
                return cached_catalog
        catalog = self._get_catalog_from_platform()
        self._save_catalog_to_cache(catalog_doc, catalog)
        return catalog

    def _get_latest_localstack_version(self) -> str:
        try:
            proxies = get_proxies()
            response = requests.get(
                f"{self.api_endpoint_catalog}/aws/version",
                verify=not config.is_env_true("SSL_NO_VERIFY"),
                proxies=proxies,
            )
            if response.ok:
                return RemoteCatalogVersionResponse.model_validate(response.content).version
            self._raise_server_error(response)
        except requests.exceptions.RequestException as e:
            raise AwsCatalogLoaderException(
                f"An unexpected network error occurred when trying to fetch latest localstack version: {e}"
            ) from e

    def _should_update_cached_catalog(self, current_catalog_version: str) -> bool:
        try:
            latest_version = self._get_latest_localstack_version()
            return latest_version != current_catalog_version
        except Exception as e:
            LOG.warning(
                "Failed to retrieve the latest catalog version, cached catalog update skipped: %s",
                e,
            )
            return False

    def _save_catalog_to_cache(self, catalog_doc: FileMappedDocument, catalog: AwsRemoteCatalog):
        catalog_doc.clear()
        catalog_doc.update(catalog.model_dump())
        catalog_doc.save()

    def _get_catalog_from_platform(self) -> AwsRemoteCatalog:
        try:
            proxies = get_proxies()
            response = requests.post(
                self.api_endpoint_catalog,
                verify=not config.is_env_true("SSL_NO_VERIFY"),
                proxies=proxies,
            )

            if response.ok:
                return self._parse_catalog(response.content)
            self._raise_server_error(response)
        except requests.exceptions.RequestException as e:
            raise AwsCatalogLoaderException(
                f"An unexpected network error occurred when trying to fetch remote catalog: {e}"
            ) from e

    def _parse_catalog(self, document: bytes) -> AwsRemoteCatalog | None:
        try:
            catalog_json = json.loads(document)
        except JSONDecodeError as e:
            raise AwsCatalogLoaderException(f"Could not de-serialize json catalog: {e}") from e
        remote_catalog = AwsRemoteCatalog.model_validate(catalog_json)
        if remote_catalog.schema_version != self.supported_schema_version:
            raise AwsCatalogLoaderException(
                f"Unsupported schema version: '{remote_catalog.schema_version}'. Only '{self.supported_schema_version}' is supported"
            )
        return remote_catalog

    def _raise_server_error(self, response: requests.Response):
        try:
            server_error = response.json()
            if error_message := server_error.get("message"):
                raise AwsCatalogLoaderException(
                    f"Unexpected AWS catalog server error: {response.text}"
                )
            raise AwsCatalogLoaderException(
                f"A server error occurred while calling remote catalog API (HTTP {response.status_code}): {error_message}"
            )
        except Exception:
            raise AwsCatalogLoaderException(
                f"An unexpected server error occurred while calling remote catalog API (HTTP {response.status_code}): {response.text}"
            )

1	import json	1✔
2	import logging	1✔
3	from json import JSONDecodeError	1✔
4	from pathlib import Path	1✔
5
6	import requests	1✔
7	from pydantic import BaseModel	1✔
8
9	from localstack import config, constants	1✔
10	from localstack.utils.catalog.common import AwsRemoteCatalog	1✔
11	from localstack.utils.http import get_proxies	1✔
12	from localstack.utils.json import FileMappedDocument	1✔
13
14	LOG = logging.getLogger(__name__)	1✔
15
16	AWS_CATALOG_FILE_NAME = "aws_catalog.json"	1✔
17
18
19	class RemoteCatalogVersionResponse(BaseModel):	1✔
20	emulator_type: str	1✔
21	version: str	1✔
22
23
24	class AwsCatalogLoaderException(Exception):	1✔
25	def __init__(self, msg: str, *args):	1✔
26	super().__init__(msg, *args)	1✔
27
28
29	class RemoteCatalogLoader:	1✔
30	supported_schema_version = "v1"	1✔
31	api_endpoint_catalog = f"{constants.API_ENDPOINT}/license/catalog"	1✔
32	catalog_file_path = Path(config.dirs.cache) / AWS_CATALOG_FILE_NAME	1✔
33
34	def get_remote_catalog(self) -> AwsRemoteCatalog:	1✔
35	catalog_doc = FileMappedDocument(self.catalog_file_path)	1✔
36	cached_catalog = AwsRemoteCatalog(**catalog_doc) if catalog_doc else None	1✔
37	if cached_catalog:	1✔
38	cached_catalog_version = cached_catalog.localstack.version	×
39	if not self._should_update_cached_catalog(cached_catalog_version):	×
40	return cached_catalog	×
41	catalog = self._get_catalog_from_platform()	1✔
42	self._save_catalog_to_cache(catalog_doc, catalog)	1✔
43	return catalog	1✔
44
45	def _get_latest_localstack_version(self) -> str:	1✔
46	try:	×
47	proxies = get_proxies()	×
48	response = requests.get(	×
49	f"{self.api_endpoint_catalog}/aws/version",
50	verify=not config.is_env_true("SSL_NO_VERIFY"),
51	proxies=proxies,
52	)
53	if response.ok:	×
54	return RemoteCatalogVersionResponse.model_validate(response.content).version	×
55	self._raise_server_error(response)	×
56	except requests.exceptions.RequestException as e:	×
57	raise AwsCatalogLoaderException(	×
58	f"An unexpected network error occurred when trying to fetch latest localstack version: {e}"
59	) from e
60
61	def _should_update_cached_catalog(self, current_catalog_version: str) -> bool:	1✔
62	try:	×
63	latest_version = self._get_latest_localstack_version()	×
64	return latest_version != current_catalog_version	×
65	except Exception as e:	×
66	LOG.warning(	×
67	"Failed to retrieve the latest catalog version, cached catalog update skipped: %s",
68	e,
69	)
70	return False	×
71
72	def _save_catalog_to_cache(self, catalog_doc: FileMappedDocument, catalog: AwsRemoteCatalog):	1✔
73	catalog_doc.clear()	1✔
74	catalog_doc.update(catalog.model_dump())	1✔
75	catalog_doc.save()	1✔
76
77	def _get_catalog_from_platform(self) -> AwsRemoteCatalog:	1✔
78	try:	1✔
79	proxies = get_proxies()	1✔
80	response = requests.post(	1✔
81	self.api_endpoint_catalog,
82	verify=not config.is_env_true("SSL_NO_VERIFY"),
83	proxies=proxies,
84	)
85
86	if response.ok:	1✔
87	return self._parse_catalog(response.content)	1✔
UNCOV 88	self._raise_server_error(response)	×
UNCOV 89	except requests.exceptions.RequestException as e:	×
90	raise AwsCatalogLoaderException(	×
91	f"An unexpected network error occurred when trying to fetch remote catalog: {e}"
92	) from e
93
94	def _parse_catalog(self, document: bytes) -> AwsRemoteCatalog \| None:	1✔
95	try:	1✔
96	catalog_json = json.loads(document)	1✔
97	except JSONDecodeError as e:	1✔
98	raise AwsCatalogLoaderException(f"Could not de-serialize json catalog: {e}") from e	1✔
99	remote_catalog = AwsRemoteCatalog.model_validate(catalog_json)	1✔
100	if remote_catalog.schema_version != self.supported_schema_version:	1✔
101	raise AwsCatalogLoaderException(	×
102	f"Unsupported schema version: '{remote_catalog.schema_version}'. Only '{self.supported_schema_version}' is supported"
103	)
104	return remote_catalog	1✔
105
106	def _raise_server_error(self, response: requests.Response):	1✔
UNCOV 107	try:	×
UNCOV 108	server_error = response.json()	×
UNCOV 109	if error_message := server_error.get("message"):	×
UNCOV 110	raise AwsCatalogLoaderException(	×
111	f"Unexpected AWS catalog server error: {response.text}"
112	)
113	raise AwsCatalogLoaderException(	×
114	f"A server error occurred while calling remote catalog API (HTTP {response.status_code}): {error_message}"
115	)
UNCOV 116	except Exception:	×
UNCOV 117	raise AwsCatalogLoaderException(	×
118	f"An unexpected server error occurred while calling remote catalog API (HTTP {response.status_code}): {response.text}"
119	)

localstack / localstack / 22166837280

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous