test(integration): dynamic-stack CMA SDK sanity suite + AM 2.0 coverage + asset Content-Type fix (v1.10.1)

.gitignore

@@ -133,9 +133,17 @@ tests/config/default.yml
 .vscode/settings.json
 run.py
 tests/resources/.DS_Store
-.talismanrc
 tests/.DS_Store
-tests/resources/.DS_Store
 .DS_Store
 */data/regions.json
-.talismanrc
+# Local backup of legacy tests (do not commit)
+
+# --- CMA integration suite: do not track ---
+docs/
+tests_backup_legacy/
+tests/integration/report/
+tests/integration/.env
+tests/integration/.env.example
+# Timestamped reports written at repo root
+cma-python-report-*.html
+api-requests-*.txt

AGENTS.md

@@ -16,7 +16,7 @@
 | Language | Python ≥ 3.9 (`setup.py` `python_requires`) |
 | Build | `setuptools` / `setup.py`; package `contentstack_management` |
 | HTTP | `requests`, `requests-toolbelt`, `urllib3` |
-| Tests | `pytest` — `tests/unit`, `tests/api`, `tests/mock` |
+| Tests | `pytest` — `tests/integration` (live e2e / sanity, dynamic stack), `tests/unit`, `tests/mock`, `tests/api` (legacy, superseded by `tests/integration`) |
 | Lint / coverage | `pylint`, `coverage` (see `requirements.txt`) |
 | Secrets / hooks | Talisman, Snyk (see `README.md` development setup) |
 
@@ -29,24 +29,42 @@
 | `contentstack_management/stack/stack.py` | **Stack**-scoped CMA operations |
 | `contentstack_management/*/` | Domain modules (entries, assets, webhooks, taxonomies, …) |
 | `contentstack_management/__init__.py` | Public exports |
-| `tests/cred.py` | **`get_credentials()`** — **dotenv** + env vars for API/mock tests |
+| `tests/integration/` | **Live e2e / sanity suite** (pytest). Self-contained: creates a fresh stack per run, exercises every SDK method (positive/negative/edge), tears it down. Own `framework/` + `data/`; config in `tests/integration/.env`. |
+| `tests/cred.py` | **`get_credentials()`** — **dotenv** + env vars for the legacy `tests/api` / `tests/mock` suites |
 
 ## Commands (quick reference)
 
 | Command Type | Command |
 |---|---|
 | Install | `pip install -e ".[dev]"` |
+| **Sanity / e2e (live)** | `pytest tests/integration` — dynamically creates a stack, runs the full suite, tears it down. Needs `tests/integration/.env` (`EMAIL`, `PASSWORD`, `HOST`, `ORGANIZATION`). Writes a timestamped HTML report + cURL log to the repo root. |
+| Sanity, keep stack | `DELETE_DYNAMIC_RESOURCES=false pytest tests/integration` (preserve the created stack for debugging) |
+| Sanity, one resource | `pytest tests/integration/api/test_12_content_type.py` |
 | Test (unit) | `pytest tests/unit/ -v` |
-| Test (API, live) | `pytest tests/api/ -v` (needs `.env` — see `tests/cred.py`) |
 | Test (mock) | `pytest tests/mock/ -v` |
-| Coverage | `coverage run -m pytest tests/unit/` |
+| Test (legacy API, live) | `pytest tests/api/ -v` (needs `.env` — see `tests/cred.py`) |
+| Coverage (CI) | `coverage run -m pytest tests/unit/` |
 | Lint | `pylint contentstack_management/` |
 
-## Environment variables (API / integration tests)
+> **CI note:** `.github/workflows/unit-test.yml` runs **only `tests/unit/`** (no credentials). The `tests/integration` sanity suite is run manually (or via a credential-gated job) because it provisions real stacks.
 
-Loaded via **`tests/cred.py`** (`load_dotenv()`). Examples include **`HOST`**, **`APIKEY`**, **`AUTHTOKEN`**, **`MANAGEMENT_TOKEN`**, **`ORG_UID`**, and resource UIDs (**`CONTENT_TYPE_UID`**, **`ENTRY_UID`**, …). See that file for the full list.
+## Environment variables
 
-Do not commit secrets.
+**Sanity / e2e suite** (`tests/integration`) — configured via **`tests/integration/.env`** (gitignored). No pre-existing stack/UIDs needed; the suite creates everything at runtime.
+
+| Var | Required | Purpose |
+|-----|----------|---------|
+| `EMAIL`, `PASSWORD` | ✅ | Login for the run (a **non-2FA** account) |
+| `HOST` | ✅ | API host (e.g. `api.contentstack.io`) |
+| `ORGANIZATION` | ✅ | Org the dynamic test stack is created in |
+| `MFA_SECRET` | — | TOTP secret (for the OAuth/2FA account, not the primary login) |
+| `DELETE_DYNAMIC_RESOURCES` | — | `false` keeps the created stack for debugging (default deletes) |
+| `CLIENT_ID`, `APP_ID`, `REDIRECT_URI` | — | OAuth tests |
+| `PERSONALIZE_HOST` | — | Personalize project for variant tests |
+
+**Legacy `tests/api` / `tests/mock`** — loaded via **`tests/cred.py`** (`load_dotenv()`): `HOST`, `APIKEY`, `AUTHTOKEN`, `MANAGEMENT_TOKEN`, `ORG_UID`, and resource UIDs. See that file for the full list.
+
+Do not commit secrets. `tests/integration/.env`, `docs/`, and the repo-root `cma-python-report-*.html` / `api-requests-*.txt` are gitignored.
 
 ## Where the documentation lives: skills
 

CHANGELOG.md

@@ -1,6 +1,14 @@
 # CHANGELOG
 
 ## Content Management SDK For Python
+---
+## v1.10.1
+
+#### Date: 26 June 2026
+
+- Fixed `Asset.update()` to send the JSON body with `Content-Type: application/json` instead of an invalid bare `multipart/form-data`, which the API rejected with 422.
+- Fixed `Asset.replace()` to let the HTTP layer set `multipart/form-data` with a proper boundary (a bare `multipart/form-data` header without a boundary previously caused a 422). Both fixes also remove a side effect that leaked the wrong `Content-Type` onto subsequent requests.
+
 ---
 ## v1.10.0
 

contentstack_management/__init__.py

@@ -102,7 +102,7 @@ def get_contentstack_endpoint(region='us', service='', omit_https=False):
 __author__ = 'dev-ex'
 __status__ = 'debug'
 __region__ = 'na'
-__version__ = '1.10.0'
+__version__ = '1.10.1'
 __host__ = 'api.contentstack.io'
 __protocol__ = 'https://'
 __api_version__ = 'v3'

contentstack_management/assets/assets.py

@@ -174,7 +174,10 @@ def replace(self, file_path):
         """
 
         url = f"assets/{self.asset_uid}"
-        Parameter.add_header(self, "Content-Type", "multipart/form-data")
+        # Let requests build the multipart body and set Content-Type WITH a boundary.
+        # Setting a bare "multipart/form-data" (no boundary) makes the API reject the
+        # upload with 422 "Please send a valid multipart/form-data payload".
+        self.client.headers.pop("Content-Type", None)
         files = {"asset": open(f"{file_path}",'rb')}
         return self.client.put(url, headers = self.client.headers, params = self.params, files = files)
 
@@ -407,7 +410,9 @@ def update(self, data):
         if self.asset_uid is None or '':
             raise Exception(ASSET_UID_REQUIRED)
         url = f"assets/{self.asset_uid}"
-        Parameter.add_header(self, "Content-Type", "multipart/form-data")
+        # Updating an asset's title/description sends a JSON body, so it must use
+        # application/json. Forcing multipart/form-data here makes the API reject
+        # the request with 422 "Please send a valid multipart/form-data payload".
         return self.client.put(url, headers = self.client.headers, params = self.params, data = data)
 
     def publish(self, data):

requirements.txt

@@ -7,3 +7,6 @@ pylint>=2.0.0
 requests-toolbelt>=1.0.0,<2.0.0
 pyotp==2.9.0
 packaging>=24.0
+# Integration test suite
+pytest>=7.0
+pytest-order>=1.2.0

tests/integration/api/test_01_user.py

@@ -0,0 +1,42 @@
+"""User API tests — profile fetch and update."""
+
+import pytest
+
+from framework import helpers as h
+
+pytestmark = pytest.mark.order(1)
+
+
+class TestUser:
+    def test_fetch(self, ctx):
+        resp = ctx.client.user().fetch()
+        h.assert_status(resp, 200)
+        user = h.body(resp).get("user", {})
+        h.tracked_assert(user.get("uid"), "user uid").truthy()
+
+    def test_update_noop(self, ctx):
+        # Send the current first_name back — a harmless update that exercises PUT /user.
+        current = h.body(ctx.client.user().fetch()).get("user", {})
+        payload = {"user": {"first_name": current.get("first_name", "Test")}}
+        resp = ctx.client.user().update(payload)
+        h.assert_status(resp, 200, 201)
+
+
+class TestUserAuthOps:
+    """Account auth endpoints exercised safely (bogus tokens / non-real email)."""
+
+    def test_activate_bogus_token(self, ctx):
+        resp = ctx.client.user().activate("bogus_activation_token", {"user": {"password": "Test@12345"}})
+        h.assert_status(resp, 400, 404, 422)
+
+    def test_reset_password_bogus_token(self, ctx):
+        resp = ctx.client.user().reset_password(
+            {"user": {"reset_password_token": "bogus", "password": "Test@12345", "password_confirmation": "Test@12345"}}
+        )
+        h.assert_status(resp, 400, 404, 422)
+
+    def test_forgot_password(self, ctx):
+        # Triggers a reset email to a non-real address; APIs typically return 200
+        # regardless (to avoid email enumeration) or a 422.
+        resp = ctx.client.user().forgot_password({"user": {"email": "noreply+test@example.com"}})
+        h.assert_status(resp, 200, 201, 422, 429)

tests/integration/api/test_02_organization.py

@@ -0,0 +1,50 @@
+"""Organization API tests — fetch, roles, stacks, logs, negative cases."""
+
+import pytest
+
+from framework import helpers as h
+
+pytestmark = pytest.mark.order(2)
+
+
+class TestOrganization:
+    def test_find_all(self, ctx):
+        resp = ctx.client.organizations().find()
+        h.assert_status(resp, 200)
+        h.tracked_assert(h.body(resp).get("organizations"), "orgs list").is_type(list)
+
+    def test_fetch(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).fetch()
+        h.assert_status(resp, 200)
+        org = h.body(resp).get("organization", {})
+        h.tracked_assert(org.get("uid"), "org uid").equals(ctx.organization_uid)
+
+    def test_roles(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).roles()
+        h.assert_status(resp, 200)
+
+    def test_stacks(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).stacks()
+        h.assert_status(resp, 200)
+
+    def test_logs(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).logs()
+        h.assert_status(resp, 200)
+
+
+class TestOrganizationOwnership:
+    """Exercised safely with invalid data so no real invite/transfer occurs."""
+
+    def test_add_users_invalid(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).add_users({"share": {}})
+        h.assert_status(resp, 400, 403, 422)
+
+    def test_transfer_ownership_invalid(self, ctx):
+        resp = ctx.client.organizations(ctx.organization_uid).transfer_ownership({"transfer_to": "not-an-email"})
+        h.assert_status(resp, 400, 403, 422)
+
+
+class TestOrganizationNegative:
+    def test_fetch_nonexistent(self, ctx):
+        resp = ctx.client.organizations("org_does_not_exist").fetch()
+        h.assert_status(resp, 404, 422, 403)

tests/integration/api/test_03_stack.py

@@ -0,0 +1,93 @@
+"""Stack API tests — fetch, settings, users, share/unshare."""
+
+import os
+
+import pytest
+
+from framework import helpers as h
+
+pytestmark = pytest.mark.order(3)
+
+
+class TestStack:
+    def test_fetch(self, ctx):
+        resp = ctx.client.stack(ctx.stack_api_key).fetch()
+        h.assert_status(resp, 200)
+        stack = h.body(resp).get("stack", {})
+        h.tracked_assert(stack.get("api_key"), "api_key").equals(ctx.stack_api_key)
+
+    def test_settings(self, ctx):
+        resp = ctx.client.stack(ctx.stack_api_key).settings()
+        h.assert_status(resp, 200)
+
+    def test_create_settings(self, ctx):
+        data = {
+            "stack_settings": {
+                "stack_variables": {"enforce_unique_urls": "true"},
+            }
+        }
+        resp = ctx.client.stack(ctx.stack_api_key).create_settings(data)
+        h.assert_status(resp, 200, 201)
+
+    def test_users(self, ctx):
+        resp = ctx.client.stack(ctx.stack_api_key).users()
+        h.assert_status(resp, 200)
+
+    def test_update(self, ctx):
+        data = {"stack": {"description": "updated by integration suite"}}
+        resp = ctx.client.stack(ctx.stack_api_key).update(data)
+        h.assert_status(resp, 200, 201)
+
+    def test_reset_settings(self, ctx):
+        resp = ctx.client.stack(ctx.stack_api_key).reset_settings({"stack_settings": {}})
+        h.assert_status(resp, 200, 201)
+
+
+class TestStackOwnership:
+    """Ownership/role operations exercised safely (no real transfer occurs)."""
+
+    def test_update_user_role(self, ctx):
+        # Map the current user to a role; on a fresh single-user stack this may
+        # be accepted (200) or rejected (422) — both confirm the SDK call works.
+        roles = h.body(ctx.client.stack(ctx.stack_api_key).roles().find()).get("roles", [])
+        role_uid = next((r["uid"] for r in roles), None)
+        if not (role_uid and ctx.user_uid):
+            pytest.skip("no role/user available")
+        resp = ctx.client.stack(ctx.stack_api_key).update_user_role({"users": {ctx.user_uid: [role_uid]}})
+        # 404 when the user isn't a separately-added stack member (owner self-assign).
+        h.assert_status(resp, 200, 201, 404, 422)
+
+    def test_transfer_ownership_invalid(self, ctx):
+        # Transferring to an invalid address must fail — exercises the endpoint
+        # without actually handing the stack to anyone.
+        resp = ctx.client.stack(ctx.stack_api_key).transfer_ownership({"transfer_to": "not-an-email"})
+        h.assert_status(resp, 400, 422)
+
+    def test_accept_ownership_bogus_token(self, ctx):
+        # Accepting with a bogus token must fail.
+        resp = ctx.client.stack(ctx.stack_api_key).accept_ownership(ctx.user_uid or "uid", "bogus_token")
+        h.assert_status(resp, 400, 404, 422)
+
+
+class TestStackSharing:
+    def test_share(self, ctx):
+        member = os.getenv("MEMBER_EMAIL")
+        if not member:
+            pytest.skip("MEMBER_EMAIL not set")
+        # Sharing requires a valid role mapping per email — an empty roles object
+        # is rejected with 422 "roles is not valid".
+        roles = h.body(ctx.client.stack(ctx.stack_api_key).roles().find()).get("roles", [])
+        role_uid = next((r["uid"] for r in roles if r.get("name") != "Admin"),
+                        roles[0]["uid"] if roles else None)
+        if not role_uid:
+            pytest.skip("no role available to share with")
+        data = {"emails": [member], "roles": {member: [role_uid]}}
+        resp = ctx.client.stack(ctx.stack_api_key).share(data)
+        h.assert_status(resp, 200, 201)
+
+    def test_unshare(self, ctx):
+        member = os.getenv("MEMBER_EMAIL")
+        if not member:
+            pytest.skip("MEMBER_EMAIL not set")
+        resp = ctx.client.stack(ctx.stack_api_key).unshare({"email": member})
+        h.assert_status(resp, 200, 201)

tests/integration/api/test_04_locale.py

@@ -0,0 +1,64 @@
+"""Locale API tests — CRUD, fallback, negative cases."""
+
+import pytest
+
+from framework import helpers as h
+
+pytestmark = pytest.mark.order(4)
+
+# A non-master locale code to create/manipulate.
+_CODE = "fr-fr"
+
+
+class TestLocaleCRUD:
+    def test_create(self, stack, store):
+        data = {"locale": {"name": "French", "code": _CODE}}
+        resp = stack.locale().create(data)
+        h.assert_status(resp, 201)
+        store["locales"]["custom"] = _CODE
+        h.wait(h.SHORT_DELAY)
+
+    def test_find_all(self, stack):
+        resp = stack.locale().find()
+        h.assert_status(resp, 200)
+        h.tracked_assert(h.body(resp).get("locales"), "locales list").is_type(list)
+
+    def test_fetch(self, stack):
+        resp = stack.locale(_CODE).fetch()
+        h.assert_status(resp, 200)
+        h.validate_locale_response(resp)
+
+    def test_update(self, stack):
+        data = {"locale": {"name": "French (FR)"}}
+        resp = stack.locale(_CODE).update(data)
+        h.assert_status(resp, 200, 201)
+
+    def test_set_fallback(self, stack):
+        data = {"locale": {"name": "German", "code": "de-de", "fallback_locale": "en-us"}}
+        resp = stack.locale().set_fallback(data)
+        h.assert_status(resp, 200, 201)
+
+    def test_update_fallback(self, stack):
+        # Ensure de-de exists, then update its fallback configuration.
+        stack.locale().create({"locale": {"name": "German", "code": "de-de"}})
+        h.wait(h.SHORT_DELAY)
+        data = {"locale": {"name": "German", "code": "de-de", "fallback_locale": "en-us"}}
+        resp = stack.locale("de-de").update_fallback(data)
+        h.assert_status(resp, 200, 201)
+
+
+class TestLocaleNegative:
+    def test_fetch_nonexistent(self, stack):
+        resp = stack.locale("zz-zz").fetch()
+        h.assert_status(resp, 404, 422)
+
+    def test_fetch_without_code_raises(self, stack):
+        # Locale guards on a missing locale_code before the HTTP call.
+        with pytest.raises(Exception):
+            stack.locale().fetch()
+
+
+class TestLocaleDelete:
+    def test_delete(self, stack):
+        resp = stack.locale("de-de").delete()
+        h.assert_status(resp, 200)