test: add asset-scanning (AM 2.0) integration coverage

Cover the asset scanning feature (DAM/AM 2.0) against both the normal org and
the AM 2.0 (DAM-enabled) org:

- normal-org scan tests (test_06_asset): upload returns _asset_scan_status
  'pending'; field absent unless include_asset_scan_status=true; clean file
  scans 'clean'; EICAR test file scans 'quarantined'; listing includes status
- test_31_am_assets: AM-org assets get 'am'-prefixed UIDs, full CRUD round-trip,
  the same scan lifecycle, publish() with the api_version: 3.2 header (publish-
  only; 404 on fetch)
- framework: am_stack fixture (stack in AM_ORG_UID; whole suite skips when
  unset), runtime-generated EICAR fixture (base64-encoded so the signature is
  not committed raw), wait_for_scan() polling helper, and api_version header
  reset for per-test isolation

Note: the correct query param is include_asset_scan_status (the response field
is _asset_scan_status); verified live.

Author: aniket-shikhare-cstk

tests/integration/api/test_06_asset.py

@@ -179,6 +179,52 @@ def test_specific_asset_type_requires_type(self, stack):
             stack.assets().specific_asset_type(None)
 
 
+class TestAssetScan:
+    """Asset scanning (AM 2.0) — verified enabled on the normal ORGANIZATION too.
+
+    The scan status is exposed only when include_asset_scan_status=true is passed;
+    the response field is _asset_scan_status with values pending -> clean | quarantined.
+    """
+
+    def test_upload_returns_pending(self, stack):
+        asset = stack.assets()
+        asset.add_param("include_asset_scan_status", "true")
+        resp = asset.upload(_ASSET_PATH)
+        h.assert_status(resp, 201)
+        status = h.body(resp).get("asset", {}).get("_asset_scan_status")
+        h.tracked_assert(status, "scan status on upload").equals("pending")
+
+    def test_scan_status_absent_without_param(self, stack):
+        # The field must be absent unless the include param is passed.
+        created = h.body(stack.assets().upload(_ASSET_PATH)).get("asset", {})
+        resp = stack.assets(created["uid"]).fetch()
+        h.assert_status(resp, 200)
+        h.tracked_assert(
+            "_asset_scan_status" not in h.body(resp).get("asset", {}), "field absent w/o param"
+        ).equals(True)
+
+    def test_clean_asset_scanned_clean(self, stack):
+        created = h.body(stack.assets().upload(_ASSET_PATH)).get("asset", {})
+        status = h.wait_for_scan(stack, created["uid"], "clean")
+        h.tracked_assert(status, "clean file scan result").equals("clean")
+
+    def test_malware_asset_quarantined(self, stack, eicar_file):
+        created = h.body(stack.assets().upload(eicar_file)).get("asset", {})
+        status = h.wait_for_scan(stack, created["uid"], "quarantined")
+        h.tracked_assert(status, "EICAR scan result").equals("quarantined")
+
+    def test_find_includes_scan_status(self, stack):
+        query = stack.assets()
+        query.add_param("include_asset_scan_status", "true")
+        resp = query.find()
+        h.assert_status(resp, 200)
+        assets = h.body(resp).get("assets", [])
+        if assets:
+            h.tracked_assert(
+                "_asset_scan_status" in assets[0], "scan status in listing"
+            ).equals(True)
+
+
 class TestAssetDelete:
     def test_delete(self, stack):
         created = h.body(stack.assets().upload(_ASSET_PATH))

tests/integration/api/test_31_am_assets.py

@@ -0,0 +1,90 @@
+"""
+AM 2.0 (DAM 2.0) asset tests — run only against the AM-enabled org (AM_ORG_UID).
+
+What's AM 2.0-specific vs the normal-org scan tests (test_06):
+  - asset UIDs are 'am'-prefixed (vs 'blt')
+  - the `api_version: 3.2` header is required on publish (single/bulk) and is
+    publish-only — applying it to fetch/upload returns 404
+
+Asset scanning itself behaves identically in both orgs (verified): the
+include_asset_scan_status=true param surfaces _asset_scan_status with values
+pending -> clean | quarantined. The whole file skips when AM_ORG_UID is unset.
+"""
+
+import os
+
+import pytest
+
+from framework import helpers as h
+
+pytestmark = pytest.mark.order(31)
+
+_ASSET_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "assets", "sample.png")
+
+
+class TestAMAssetBasics:
+    def test_upload_has_am_uid_prefix(self, am_stack):
+        resp = am_stack.assets().upload(_ASSET_PATH)
+        h.assert_status(resp, 201)
+        uid = h.body(resp).get("asset", {}).get("uid", "")
+        h.tracked_assert(uid[:2], "AM 2.0 asset uid prefix").equals("am")
+
+    def test_crud_round_trip(self, am_stack):
+        uid = h.body(am_stack.assets().upload(_ASSET_PATH)).get("asset", {}).get("uid")
+        h.wait(h.SHORT_DELAY)
+        h.assert_status(am_stack.assets(uid).fetch(), 200)
+        h.assert_status(am_stack.assets().find(), 200)
+        h.assert_status(am_stack.assets(uid).version(), 200)
+        h.assert_status(am_stack.assets(uid).delete(), 200)
+
+
+class TestAMAssetScan:
+    def test_upload_returns_pending(self, am_stack):
+        asset = am_stack.assets()
+        asset.add_param("include_asset_scan_status", "true")
+        resp = asset.upload(_ASSET_PATH)
+        h.assert_status(resp, 201)
+        h.tracked_assert(
+            h.body(resp).get("asset", {}).get("_asset_scan_status"), "scan status on upload"
+        ).equals("pending")
+
+    def test_clean_asset_scanned_clean(self, am_stack):
+        uid = h.body(am_stack.assets().upload(_ASSET_PATH)).get("asset", {}).get("uid")
+        status = h.wait_for_scan(am_stack, uid, "clean")
+        h.tracked_assert(status, "clean file scan result").equals("clean")
+
+    def test_malware_asset_quarantined(self, am_stack, eicar_file):
+        uid = h.body(am_stack.assets().upload(eicar_file)).get("asset", {}).get("uid")
+        status = h.wait_for_scan(am_stack, uid, "quarantined")
+        h.tracked_assert(status, "EICAR scan result").equals("quarantined")
+
+    def test_scan_status_absent_without_param(self, am_stack):
+        uid = h.body(am_stack.assets().upload(_ASSET_PATH)).get("asset", {}).get("uid")
+        asset = h.body(am_stack.assets(uid).fetch()).get("asset", {})
+        h.tracked_assert("_asset_scan_status" not in asset, "field absent w/o param").equals(True)
+
+
+class TestAMAssetPublish:
+    def test_publish_with_api_version_3_2(self, am_stack):
+        uid = h.body(am_stack.assets().upload(_ASSET_PATH)).get("asset", {}).get("uid")
+        # upload() pops Content-Type; restore it before the JSON requests below.
+        am_stack.client.headers["Content-Type"] = "application/json"
+        env = h.generate_valid_uid("env_am")
+        am_stack.environments().create(
+            {"environment": {"name": env, "urls": [{"url": "https://e.example.com", "locale": "en-us"}]}}
+        )
+        h.wait(h.SHORT_DELAY)
+        am_stack.client.headers["Content-Type"] = "application/json"
+        asset = am_stack.assets(uid)
+        asset.add_header("api_version", "3.2")
+        resp = asset.publish({"asset": {"locales": ["en-us"], "environments": [env]}, "version": 1})
+        # Publish always returns success; scan validation happens async on the CDA side.
+        h.assert_status(resp, 200, 201)
+
+    def test_api_version_3_2_is_publish_only(self, am_stack):
+        # The api_version: 3.2 header is publish-only — on fetch it 404s.
+        uid = h.body(am_stack.assets().upload(_ASSET_PATH)).get("asset", {}).get("uid")
+        asset = am_stack.assets(uid)
+        asset.add_header("api_version", "3.2")
+        resp = asset.fetch()
+        h.assert_status(resp, 404)

tests/integration/conftest.py

@@ -29,7 +29,7 @@
 from framework import capture, report
 from framework import setup as setup_mod
 from framework.context import reset_store, test_data
-from framework.helpers import set_active_tracker
+from framework.helpers import set_active_tracker, short_id, wait
 from framework.report import TestRecord
 
 load_dotenv()
@@ -72,6 +72,60 @@ def store():
     return test_data
 
 
+@pytest.fixture(scope="session")
+def am_stack(ctx):
+    """A stack created in the AM 2.0 (DAM-enabled) org from AM_ORG_UID.
+
+    Skips the whole AM suite when AM_ORG_UID is not configured. Reuses the
+    authenticated session client; only the stack lives in the AM org.
+    """
+    am_org = os.getenv("AM_ORG_UID")
+    if not am_org:
+        pytest.skip("AM_ORG_UID not set — AM 2.0 tests require a DAM-enabled org")
+    client = ctx.client
+    # This session fixture runs before the per-test header reset, so restore a
+    # clean JSON Content-Type (a prior asset upload may have mutated/popped it).
+    client.client.headers["Content-Type"] = "application/json"
+    client.client.headers.pop("api_version", None)
+    client.client.headers["organization_uid"] = am_org  # Stack.create org-header workaround
+    resp = client.stack().create(am_org, {"stack": {
+        "name": f"SDK_Py_AM_{short_id()}",
+        "description": "Automated AM 2.0 test stack",
+        "master_locale": "en-us",
+    }})
+    if resp.status_code not in (200, 201):
+        client.client.headers["organization_uid"] = ctx.organization_uid
+        pytest.skip(f"could not create AM stack ({resp.status_code}): {resp.text[:120]}")
+    api_key = resp.json()["stack"]["api_key"]
+    wait(5)
+    yield client.stack(api_key)
+    # teardown: delete the AM stack, then restore the normal org header
+    if setup_mod.should_delete_resources():
+        try:
+            client.stack(api_key).delete()
+        except Exception:  # noqa: BLE001
+            pass
+    client.client.headers["organization_uid"] = ctx.organization_uid
+
+
+@pytest.fixture(scope="session")
+def eicar_file(tmp_path_factory):
+    """Path to an EICAR antivirus test file, written at runtime (never committed).
+
+    The asset scanner quarantines this standard test signature, letting us assert
+    the 'quarantined' scan status. The signature is stored base64-encoded (not as a
+    raw literal) so the source file itself isn't flagged by antivirus / repo scanners.
+    """
+    import base64
+
+    signature = base64.b64decode(
+        "WDVPIVAlQEFQWzRcUFpYNTQoUF4pN0NDKTd9JEVJQ0FSLVNUQU5EQVJELUFOVElWSVJVUy1URVNULUZJTEUhJEgrSCo="
+    )
+    path = tmp_path_factory.mktemp("am_scan") / "eicar.com"
+    path.write_bytes(signature)
+    return str(path)
+
+
 # ---------------------------------------------------------------------------
 # Per-test capture wiring
 # ---------------------------------------------------------------------------
@@ -101,6 +155,7 @@ def _reset_client_headers(request):
     headers = context.client.client.headers
     headers["Content-Type"] = "application/json"
     headers.pop("branch", None)
+    headers.pop("api_version", None)  # AM 2.0 publish header leaks otherwise (breaks later calls)
     yield
 
 

tests/integration/framework/helpers.py

@@ -92,6 +92,24 @@ def body(response) -> dict:
         return {}
 
 
+def wait_for_scan(stack, asset_uid, expected, timeout=40, interval=3):
+    """Poll an asset's _asset_scan_status (AM 2.0) until it reaches `expected`.
+
+    Requires the include_asset_scan_status=true query param to surface the field.
+    Returns the last observed status (the caller asserts == expected).
+    """
+    deadline = time.time() + timeout
+    last = None
+    while time.time() < deadline:
+        asset = stack.assets(asset_uid)
+        asset.add_param("include_asset_scan_status", "true")
+        last = body(asset.fetch()).get("asset", {}).get("_asset_scan_status")
+        if last == expected:
+            return last
+        time.sleep(interval)
+    return last
+
+
 # ---------------------------------------------------------------------------
 # Status / error assertions (Python SDK does NOT raise on HTTP errors)
 # ---------------------------------------------------------------------------