test: add unit and integration tests for DataContract delete lifecycle (DQ-665)

- Add DataContractSpec unit tests (parsing, mutation, YAML roundtrip) and
  ContractClient unit tests (generate_initial_spec, delete, delete_latest_version
  header assertions) to both pyatlan and pyatlan_v9 test suites
- Add sync integration tests for generate_initial_spec, spec-based contract
  creation, delete_all, and delete_latest_version (xfail: Atlas NPE on dq-dev
  until backend version-chain schema is deployed) in data_mesh_test.py and v9
- Add async integration tests for the same scenarios in aio/test_client.py
  for both pyatlan and pyatlan_v9
- Move CONTRACT_DELETE_SCOPE_HEADER constant from contract.py / aio/contract.py
  into pyatlan/client/constants.py (single source of truth across 4 files)
- Add Optional[Dict[str, str]] type hints to extra_headers param in atlan.py,
  aio/client.py, and protocol.py (ApiCaller + AsyncApiCaller)

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: Aryamanz29

pyatlan/client/aio/client.py

@@ -16,7 +16,7 @@
 from contextlib import _AsyncGeneratorContextManager
 from http import HTTPStatus
 from types import SimpleNamespace
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
 import httpx
 from httpx_retries.retry import Retry
@@ -501,7 +501,7 @@ async def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response=False,
-        extra_headers=None,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         """
         Async version of _call_api - mirrors sync client structure.

pyatlan/client/aio/contract.py

@@ -7,14 +7,16 @@
 from pydantic.v1 import validate_arguments
 
 from pyatlan.client.common import AsyncApiCaller, ContractInit
-from pyatlan.client.constants import CONTRACT_INIT_API, DELETE_ENTITIES_BY_GUIDS
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan.model.assets import Asset
 from pyatlan.model.enums import AtlanDeleteType
 from pyatlan.model.response import AssetMutationResponse
 
-CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"
-
 
 class AsyncContractClient:
     """

pyatlan/client/atlan.py

@@ -792,7 +792,7 @@ def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response=False,
-        extra_headers=None,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         path = self._create_path(api)
         params = self._create_params(api, query_params, request_obj, exclude_unset)

pyatlan/client/constants.py

@@ -704,3 +704,5 @@
     HTTPStatus.ACCEPTED,
     endpoint=EndPoint.CHRONOS,
 )
+
+CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"

pyatlan/client/contract.py

@@ -5,14 +5,16 @@
 from pydantic.v1 import validate_arguments
 
 from pyatlan.client.common import ApiCaller, ContractInit
-from pyatlan.client.constants import CONTRACT_INIT_API, DELETE_ENTITIES_BY_GUIDS
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan.model.assets import Asset
 from pyatlan.model.enums import AtlanDeleteType
 from pyatlan.model.response import AssetMutationResponse
 
-CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"
-
 
 class ContractClient:
     """

pyatlan/client/protocol.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from contextlib import _AsyncGeneratorContextManager, _GeneratorContextManager
-from typing import Any, Protocol, runtime_checkable
+from typing import Any, Dict, Optional, Protocol, runtime_checkable
 
 from httpx_retries import Retry
 
@@ -26,7 +26,7 @@ def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response: bool = False,
-        extra_headers=None,
+        extra_headers: Optional[Dict[str, str]] = None,
     ):
         pass
 
@@ -57,7 +57,7 @@ async def _call_api(
         request_obj=None,
         exclude_unset: bool = True,
         text_response: bool = False,
-        extra_headers=None,
+        extra_headers: Optional[Dict[str, str]] = None,
     ) -> Any:
         pass
 

pyatlan_v9/client/aio/contract.py

@@ -5,7 +5,11 @@
 from typing import Optional
 
 from pyatlan.client.common import AsyncApiCaller
-from pyatlan.client.constants import CONTRACT_INIT_API, DELETE_ENTITIES_BY_GUIDS
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan_v9.client.asset import _parse_mutation_response
 from pyatlan_v9.model.assets import Asset
@@ -14,8 +18,6 @@
 from pyatlan_v9.model.response import AssetMutationResponse
 from pyatlan_v9.validate import validate_arguments
 
-CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"
-
 
 class V9AsyncContractClient:
     """

pyatlan_v9/client/contract.py

@@ -3,7 +3,11 @@
 from typing import Optional
 
 from pyatlan.client.common import ApiCaller
-from pyatlan.client.constants import CONTRACT_INIT_API, DELETE_ENTITIES_BY_GUIDS
+from pyatlan.client.constants import (
+    CONTRACT_DELETE_SCOPE_HEADER,
+    CONTRACT_INIT_API,
+    DELETE_ENTITIES_BY_GUIDS,
+)
 from pyatlan.errors import ErrorCode
 from pyatlan_v9.client.asset import _parse_mutation_response
 from pyatlan_v9.model.assets import Asset
@@ -12,8 +16,6 @@
 from pyatlan_v9.model.response import AssetMutationResponse
 from pyatlan_v9.validate import validate_arguments
 
-CONTRACT_DELETE_SCOPE_HEADER = "x-atlan-contract-delete-scope"
-
 
 class V9ContractClient:
     """

tests/integration/aio/test_client.py

@@ -27,13 +27,23 @@
     AtlasGlossary,
     AtlasGlossaryCategory,
     AtlasGlossaryTerm,
+    Connection,
     Database,
+    DataContract,
     Schema,
     Table,
 )
 from pyatlan.model.audit import AuditSearchRequest
+from pyatlan.model.contract import DataContractSpec
 from pyatlan.model.core import Announcement
-from pyatlan.model.enums import AnnouncementType, AtlanConnectorType, SortOrder, UTMTags
+from pyatlan.model.enums import (
+    AnnouncementType,
+    AtlanConnectorType,
+    DataContractStatus,
+    EntityStatus,
+    SortOrder,
+    UTMTags,
+)
 from pyatlan.model.fluent_search import CompoundQuery, FluentSearch
 from pyatlan.model.search import (
     DSL,
@@ -1691,3 +1701,215 @@ async def doit(asset: Asset):
 
     processed_count = await client.asset.process_assets(search=search, func=doit)
     assert processed_count == expected_count
+
+
+# ---------------------------------------------------------------------------
+# Async contract tests — client.contracts (generate_initial_spec, delete,
+# delete_latest_version)
+# ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture(scope="module")
+async def async_spec_contract(
+    client: AsyncAtlanClient,
+    table: Table,
+    connection: Connection,
+):
+    """Contract created via DataContractSpec model using the async client."""
+    assert table and table.qualified_name and table.type_name
+
+    spec = DataContractSpec(
+        kind="DataContract",
+        template_version="0.0.2",
+        status=DataContractStatus.DRAFT,
+        type=table.type_name,
+        dataset=table.name or table.qualified_name.split("/")[-1],
+        data_source=connection.name,
+        description="Automated testing of the Python SDK - async spec-based.",
+    )
+    contract = DataContract.creator(
+        asset_qualified_name=table.qualified_name,
+        contract_spec=spec,
+    )
+    response = await client.asset.save(contract)
+    created = response.assets_created(asset_type=DataContract)
+    updated = response.assets_updated(asset_type=DataContract)
+    result = (created or updated)[0]
+    yield result
+    await delete_asset_async(client, guid=result.guid, asset_type=DataContract)
+
+
+@pytest.mark.order(before="test_async_contract_from_spec")
+async def test_async_generate_initial_spec(client: AsyncAtlanClient, table: Table):
+    """client.contracts.generate_initial_spec() returns parseable YAML (async)."""
+    assert table and table.qualified_name
+
+    yaml_spec = await client.contracts.generate_initial_spec(table)
+
+    assert yaml_spec, "Expected a non-empty YAML string from generate_initial_spec"
+    spec = DataContractSpec.from_yaml(yaml_spec)
+    assert spec.kind == "DataContract"
+    assert spec.type
+    assert spec.dataset
+
+
+async def test_async_contract_from_spec(
+    client: AsyncAtlanClient, table: Table, async_spec_contract: DataContract
+):
+    """Async: DataContract created from DataContractSpec has the right asset linkage."""
+    assert async_spec_contract and async_spec_contract.guid
+    assert async_spec_contract.qualified_name
+    assert table.qualified_name
+    assert async_spec_contract.qualified_name.startswith(table.qualified_name)
+    assert "/contract" in async_spec_contract.qualified_name
+
+    fetched = await client.asset.get_by_guid(
+        async_spec_contract.guid, asset_type=DataContract, ignore_relationships=False
+    )
+    assert fetched
+    assert fetched.data_contract_asset_guid == table.guid
+    assert fetched.data_contract_spec or fetched.data_contract_json
+    assert fetched.data_contract_version and fetched.data_contract_version >= 1
+
+    # hasContract / dataContractLatest / dataContractLatestCertified
+    table_state = await client.asset.get_by_guid(
+        table.guid, asset_type=Table, ignore_relationships=False
+    )
+    assert table_state.has_contract is True
+    assert table_state.data_contract_latest is not None
+    assert table_state.data_contract_latest.guid == async_spec_contract.guid
+    assert table_state.data_contract_latest_certified is None  # DRAFT not certified
+
+
+@pytest_asyncio.fixture(scope="module")
+async def async_multi_version_contract(
+    client: AsyncAtlanClient,
+    table: Table,
+    connection: Connection,
+):
+    """V1 DRAFT → V1 VERIFIED → V2 DRAFT lifecycle (async)."""
+    import asyncio
+
+    assert table and table.qualified_name and table.type_name
+    asset_qn = table.qualified_name
+    dataset_name = table.name or asset_qn.split("/")[-1]
+
+    table_with_rels = await client.asset.get_by_guid(
+        table.guid, asset_type=Table, ignore_relationships=False
+    )
+    if table_with_rels.data_contract_latest:
+        try:
+            await client.contracts.delete(table_with_rels.data_contract_latest.guid)
+            await asyncio.sleep(2)
+        except Exception:
+            pass
+
+    async def _save_spec(status_str: str, description: str) -> DataContract:
+        spec = DataContractSpec(
+            kind="DataContract",
+            template_version="0.0.2",
+            status=status_str,
+            type=table.type_name,
+            dataset=dataset_name,
+            data_source=connection.name,
+            description=description,
+        )
+        contract = DataContract.creator(
+            asset_qualified_name=asset_qn,
+            contract_spec=spec,
+        )
+        response = await client.asset.save(contract)
+        created = response.assets_created(asset_type=DataContract)
+        updated = response.assets_updated(asset_type=DataContract)
+        return (created or updated)[0]
+
+    v1 = await _save_spec("draft", "E2E test - DRAFT v1")
+    await asyncio.sleep(3)
+    await _save_spec("VERIFIED", "E2E test - VERIFIED v1")
+    await asyncio.sleep(2)
+    v2 = await _save_spec("draft", "E2E test - DRAFT v2")
+
+    yield {"v1_guid": v1.guid, "v2_guid": v2.guid}
+
+    try:
+        await client.contracts.delete(v2.guid)
+    except Exception:
+        pass
+
+
+@pytest.mark.order(before="test_async_delete_all_contract_versions")
+@pytest.mark.xfail(
+    strict=False,
+    reason=(
+        "delete_latest_version triggers an Atlas NPE on dq-dev: "
+        "getRelationshipEdgeLabel() is null for contract version-chain attributes "
+        "not yet deployed in this environment. Remove xfail once backend is updated."
+    ),
+)
+async def test_async_delete_latest_version_restores_previous(
+    client: AsyncAtlanClient, table: Table, async_multi_version_contract: dict
+):
+    """Async: deleting latest DRAFT restores VERIFIED v1 as dataContractLatest."""
+    v2_guid = async_multi_version_contract["v2_guid"]
+    assert v2_guid
+
+    response = await client.contracts.delete_latest_version(v2_guid)
+
+    assert response
+    deleted = response.assets_deleted(asset_type=DataContract)
+    assert deleted and len(deleted) == 1
+    assert deleted[0].guid == v2_guid
+    assert deleted[0].status == EntityStatus.DELETED
+
+    table_after = await client.asset.get_by_guid(
+        table.guid, asset_type=Table, ignore_relationships=False
+    )
+    assert table_after.has_contract is True
+    assert table_after.data_contract_latest is not None
+    assert table_after.data_contract_latest.guid != v2_guid
+    assert table_after.data_contract_latest_certified is not None
+    assert table_after.data_contract_latest_certified.guid != v2_guid
+
+
+@pytest.mark.order(before="test_async_contract_from_spec")
+async def test_async_delete_all_contract_versions(
+    client: AsyncAtlanClient, table: Table, connection: Connection
+):
+    """Async: client.contracts.delete() purges all versions and clears asset attrs."""
+    assert table and table.qualified_name and table.type_name
+    dataset_name = table.name or table.qualified_name.split("/")[-1]
+
+    spec = DataContractSpec(
+        kind="DataContract",
+        template_version="0.0.2",
+        status=DataContractStatus.DRAFT,
+        type=table.type_name,
+        dataset=dataset_name,
+        data_source=connection.name,
+        description="Automated testing - async delete-all scenario.",
+    )
+    contract = DataContract.creator(
+        asset_qualified_name=table.qualified_name,
+        contract_spec=spec,
+    )
+    response = await client.asset.save(contract)
+    created = response.assets_created(asset_type=DataContract)
+    updated = response.assets_updated(asset_type=DataContract)
+    saved = (created or updated)[0]
+    assert saved and saved.guid
+
+    del_response = await client.contracts.delete(saved.guid)
+
+    assert del_response
+    deleted = del_response.assets_deleted(asset_type=DataContract)
+    assert deleted and len(deleted) >= 1
+    assert saved.guid in {a.guid for a in deleted}
+    for asset in deleted:
+        assert asset.status == EntityStatus.DELETED
+
+    table_after = await client.asset.get_by_guid(
+        table.guid, asset_type=Table, ignore_relationships=False
+    )
+    assert not table_after.has_contract
+    assert table_after.data_contract_latest is None
+    assert table_after.data_contract_latest_certified is None