diff --git a/CHANGELOG.md b/CHANGELOG.md index ada53b3..670ac4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # CHANGELOG +## _v2.6.0_ + +### **Date: 18-May-2026** + +- Added optional branch support for entry variants on `Entry.variants()` and `ContentType.variants()`. + ## _v2.5.1_ ### **Date: 15-April-2026** diff --git a/contentstack/__init__.py b/contentstack/__init__.py index dea0db0..15fdcdc 100644 --- a/contentstack/__init__.py +++ b/contentstack/__init__.py @@ -22,7 +22,7 @@ __title__ = 'contentstack-delivery-python' __author__ = 'contentstack' __status__ = 'debug' -__version__ = 'v2.5.1' +__version__ = 'v2.6.0' __endpoint__ = 'cdn.contentstack.io' __email__ = 'support@contentstack.com' __developer_email__ = 'mobile@contentstack.com' diff --git a/contentstack/contenttype.py b/contentstack/contenttype.py index 1423216..007a63d 100644 --- a/contentstack/contenttype.py +++ b/contentstack/contenttype.py @@ -118,17 +118,20 @@ def find(self, params=None): result = self.http_instance.get(url) return result - def variants(self, variant_uid: str | list[str], params: dict = None): + def variants(self, variant_uid: str | list[str], branch: str = None, params: dict = None): """ Fetches the variants of the content type - :param variant_uid: {str} -- variant_uid - :return: Entry, so you can chain this call. + :param variant_uid: {str | list[str]} -- variant UID or list of variant UIDs + :param branch: {str} -- optional branch name to scope the variant request + :param params: {dict} -- optional query parameters + :return: Variants, so you can chain this call. """ return Variants( http_instance=self.http_instance, content_type_uid=self.__content_type_uid, entry_uid=None, variant_uid=variant_uid, + branch=branch, params=params, logger=None ) diff --git a/contentstack/entry.py b/contentstack/entry.py index b0bae6b..6b0b112 100644 --- a/contentstack/entry.py +++ b/contentstack/entry.py @@ -266,17 +266,20 @@ def _merged_response(self): return merged_response # Now correctly returns a dictionary raise ValueError(ErrorMessages.MISSING_LIVE_PREVIEW_KEYS) - def variants(self, variant_uid: str | list[str], params: dict = None): + def variants(self, variant_uid: str | list[str], branch: str = None, params: dict = None): """ Fetches the variants of the entry - :param variant_uid: {str} -- variant_uid - :return: Entry, so you can chain this call. + :param variant_uid: {str | list[str]} -- variant UID or list of variant UIDs + :param branch: {str} -- optional branch name to scope the variant request + :param params: {dict} -- optional query parameters + :return: Variants, so you can chain this call. """ return Variants( http_instance=self.http_instance, content_type_uid=self.content_type_id, entry_uid=self.entry_uid, variant_uid=variant_uid, + branch=branch, params=params, logger=self.logger ) diff --git a/contentstack/variants.py b/contentstack/variants.py index 1cdc92e..e59bcfb 100644 --- a/contentstack/variants.py +++ b/contentstack/variants.py @@ -20,6 +20,7 @@ def __init__(self, content_type_uid=None, entry_uid=None, variant_uid=None, + branch=None, params=None, logger=None): @@ -30,29 +31,47 @@ def __init__(self, self.content_type_id = content_type_uid self.entry_uid = entry_uid self.variant_uid = variant_uid + self.branch = branch self.logger = logger or logging.getLogger(__name__) self.entry_param = params or {} + def _prepare_variant_headers(self): + headers = self.http_instance.headers.copy() + if isinstance(self.variant_uid, str): + headers['x-cs-variant-uid'] = self.variant_uid + elif isinstance(self.variant_uid, list): + headers['x-cs-variant-uid'] = ','.join(self.variant_uid) + if self.branch is not None: + headers['branch'] = self.branch + return headers + + def _apply_variant_headers(self, headers): + self._original_branch = self.http_instance.headers.get('branch') + self.http_instance.headers.update(headers) + + def _cleanup_variant_headers(self): + self.http_instance.headers.pop('x-cs-variant-uid', None) + if self.branch is not None: + if self._original_branch is not None: + self.http_instance.headers['branch'] = self._original_branch + else: + self.http_instance.headers.pop('branch', None) + def find(self, params=None): """ find the variants of the entry of a particular content type :param self.variant_uid: {str} -- self.variant_uid :return: Entry, so you can chain this call. """ - headers = self.http_instance.headers.copy() # Create a local copy of headers - if isinstance(self.variant_uid, str): - headers['x-cs-variant-uid'] = self.variant_uid - elif isinstance(self.variant_uid, list): - headers['x-cs-variant-uid'] = ','.join(self.variant_uid) - + headers = self._prepare_variant_headers() if params is not None: self.entry_param.update(params) encoded_params = parse.urlencode(self.entry_param) endpoint = self.http_instance.endpoint url = f'{endpoint}/content_types/{self.content_type_id}/entries?{encoded_params}' - self.http_instance.headers.update(headers) + self._apply_variant_headers(headers) result = self.http_instance.get(url) - self.http_instance.headers.pop('x-cs-variant-uid', None) + self._cleanup_variant_headers() return result def fetch(self, params=None): @@ -77,18 +96,13 @@ def fetch(self, params=None): if self.entry_uid is None: raise ValueError(ErrorMessages.ENTRY_UID_REQUIRED) else: - headers = self.http_instance.headers.copy() # Create a local copy of headers - if isinstance(self.variant_uid, str): - headers['x-cs-variant-uid'] = self.variant_uid - elif isinstance(self.variant_uid, list): - headers['x-cs-variant-uid'] = ','.join(self.variant_uid) - + headers = self._prepare_variant_headers() if params is not None: self.entry_param.update(params) encoded_params = parse.urlencode(self.entry_param) endpoint = self.http_instance.endpoint url = f'{endpoint}/content_types/{self.content_type_id}/entries/{self.entry_uid}?{encoded_params}' - self.http_instance.headers.update(headers) + self._apply_variant_headers(headers) result = self.http_instance.get(url) - self.http_instance.headers.pop('x-cs-variant-uid', None) + self._cleanup_variant_headers() return result \ No newline at end of file diff --git a/tests/test_entry.py b/tests/test_entry.py index 6e388fc..65cf584 100644 --- a/tests/test_entry.py +++ b/tests/test_entry.py @@ -9,6 +9,7 @@ HOST = config.HOST FAQ_UID = config.FAQ_UID # Add this in your config.py VARIANT_UID = config.VARIANT_UID +BRANCH = getattr(config, 'BRANCH', None) class TestEntry(unittest.TestCase): @@ -318,6 +319,34 @@ def test_41_entry_variants_multiple_uids(self): result = entry.fetch() self.assertIn('variants', result['entry']['publish_details']) + @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests") + def test_41a_entry_variants_with_branch(self): + """Test single entry variants with branch""" + content_type = self.stack.content_type('faq') + result = content_type.entry(FAQ_UID).variants(VARIANT_UID, BRANCH).fetch() + self.assertIn('variants', result['entry']['publish_details']) + + @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests") + def test_41b_entry_variants_multiple_uids_with_branch(self): + """Test single entry variants with multiple UIDs and branch""" + content_type = self.stack.content_type('faq') + result = content_type.entry(FAQ_UID).variants([VARIANT_UID], BRANCH).fetch() + self.assertIn('variants', result['entry']['publish_details']) + + @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests") + def test_41c_content_type_variants_find_with_branch(self): + """Test entries query variants with branch""" + content_type = self.stack.content_type('faq') + result = content_type.variants(VARIANT_UID, BRANCH).find() + self.assertIn('variants', result['entries'][0]['publish_details']) + + @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests") + def test_41d_content_type_variants_multiple_uids_find_with_branch(self): + """Test entries query variants with multiple UIDs and branch""" + content_type = self.stack.content_type('faq') + result = content_type.variants([VARIANT_UID], BRANCH).find() + self.assertIn('variants', result['entries'][0]['publish_details']) + def test_42_entry_environment_removal(self): """Test entry remove_environment method""" entry = (self.stack.content_type('faq') diff --git a/tests/test_variants.py b/tests/test_variants.py new file mode 100644 index 0000000..d9395be --- /dev/null +++ b/tests/test_variants.py @@ -0,0 +1,190 @@ +""" +Unit tests for Variants branch support in contentstack.variants +""" + +import pytest +from unittest.mock import MagicMock +from urllib.parse import urlencode + +from contentstack.variants import Variants + + +@pytest.fixture +def mock_http_instance(): + mock = MagicMock() + mock.endpoint = "https://cdn.contentstack.io/v3" + mock.headers = { + "api_key": "api_key", + "access_token": "delivery_token", + "environment": "test_env", + } + mock.get = MagicMock(return_value={"entries": []}) + return mock + + +@pytest.fixture +def variants(mock_http_instance): + return Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid="variant_uid", + ) + + +def _capture_headers_on_get(mock_http_instance): + captured = {} + + def _get(url): + captured["headers"] = mock_http_instance.headers.copy() + return {"entries": []} + + mock_http_instance.get.side_effect = _get + return captured + + +class TestVariantsBranch: + def test_fetch_sets_variant_and_branch_headers(self, mock_http_instance): + captured = _capture_headers_on_get(mock_http_instance) + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid="variant_uid", + branch="dev_branch", + ) + variants.fetch() + + assert captured["headers"]["x-cs-variant-uid"] == "variant_uid" + assert captured["headers"]["branch"] == "dev_branch" + + def test_fetch_multiple_variant_uids_with_branch(self, mock_http_instance): + captured = _capture_headers_on_get(mock_http_instance) + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid=["variant1", "variant2"], + branch="dev_branch", + ) + variants.fetch() + + assert captured["headers"]["x-cs-variant-uid"] == "variant1,variant2" + assert captured["headers"]["branch"] == "dev_branch" + + def test_find_sets_variant_and_branch_headers(self, mock_http_instance): + captured = _capture_headers_on_get(mock_http_instance) + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid=None, + variant_uid="variant_uid", + branch="dev_branch", + ) + variants.find() + + assert captured["headers"]["x-cs-variant-uid"] == "variant_uid" + assert captured["headers"]["branch"] == "dev_branch" + + def test_fetch_restores_stack_branch_after_request(self, mock_http_instance): + mock_http_instance.headers["branch"] = "main" + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid="variant_uid", + branch="dev_branch", + ) + variants.fetch() + + assert "x-cs-variant-uid" not in mock_http_instance.headers + assert mock_http_instance.headers["branch"] == "main" + + def test_fetch_removes_branch_when_stack_had_none(self, mock_http_instance): + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid="variant_uid", + branch="dev_branch", + ) + variants.fetch() + + assert "branch" not in mock_http_instance.headers + assert "x-cs-variant-uid" not in mock_http_instance.headers + + def test_fetch_without_branch_uses_stack_branch(self, mock_http_instance): + mock_http_instance.headers["branch"] = "main" + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + variant_uid="variant_uid", + ) + variants.fetch() + + assert mock_http_instance.headers["branch"] == "main" + assert "x-cs-variant-uid" not in mock_http_instance.headers + + def test_fetch_cleans_up_variant_header_only(self, variants, mock_http_instance): + variants.fetch() + + assert "x-cs-variant-uid" not in mock_http_instance.headers + assert mock_http_instance.headers["environment"] == "test_env" + + def test_fetch_builds_expected_url(self, variants, mock_http_instance): + variants.fetch() + expected_url = ( + "https://cdn.contentstack.io/v3/content_types/faq/entries/entry_uid?" + ) + mock_http_instance.get.assert_called_once() + assert mock_http_instance.get.call_args[0][0].startswith(expected_url) + + def test_find_builds_expected_url(self, mock_http_instance): + variants = Variants( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid=None, + variant_uid="variant_uid", + branch="dev_branch", + ) + variants.find(params={"locale": "en-us"}) + expected_params = urlencode({"locale": "en-us"}) + expected_url = ( + f"https://cdn.contentstack.io/v3/content_types/faq/entries?{expected_params}" + ) + mock_http_instance.get.assert_called_once_with(expected_url) + + def test_entry_variants_passes_branch(self, mock_http_instance): + from contentstack.entry import Entry + + entry = Entry( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + ) + result = entry.variants("variant_uid", "dev_branch") + assert isinstance(result, Variants) + assert result.branch == "dev_branch" + assert result.variant_uid == "variant_uid" + + def test_content_type_variants_passes_branch(self, mock_http_instance): + from contentstack.contenttype import ContentType + + content_type = ContentType(mock_http_instance, "faq") + result = content_type.variants(["variant1", "variant2"], "dev_branch") + assert isinstance(result, Variants) + assert result.branch == "dev_branch" + assert result.variant_uid == ["variant1", "variant2"] + + def test_variants_backward_compatible_params_kwarg(self, mock_http_instance): + from contentstack.entry import Entry + + entry = Entry( + http_instance=mock_http_instance, + content_type_uid="faq", + entry_uid="entry_uid", + ) + result = entry.variants("variant_uid", params={"locale": "en-us"}) + assert result.branch is None + assert result.entry_param == {"locale": "en-us"}