Repository: great-expectations/great_expectations
Pull request #7066 (2023-02-07 16:35:28)

[FEATURE] Add `checkpoint` arg to `DataContext` `Checkpoint` CRUD
Changes proposed in this pull request:
- Allow users to pass in an existing checkpoint through `add_checkpoint` and `add_or_update_checkpoint`


### Definition of Done
- [x] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [x] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_checklist) of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] I have added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [x] I have run any local integration tests and made sure that nothing is broken.


netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | d4c467672972948c1dc7629da7e3803a90c2bafa |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63e27dd50ac3d4000715adf6 |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=7066&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations">Click on the image</a> for a new way to code review</b>
</summary>

  - Make big changes easier — review code in small groups of related files
  - Know where to start — see the whole change at a glance
  - Take a code tour — explore the change with an interactive tour
  - Make comments and review — all fully sync’ed with github

    [Try it now!](https://app.codesee.io/r/reviews?pr=7066&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)
</details>

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/great-expectations/great_expectations/7066/fc546d7c/d4c467672972948c1dc7629da7e3803a90c2bafa.svg)](https://app.codesee.io/r/reviews?pr=7066&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)

Patch:
diff --git a/great_expectations/data_context/data_context/abstract_data_context.py b/great_expectations/data_context/data_context/abstract_data_context.py
index ffa8eb347..f8c808e6a 100644
--- a/great_expectations/data_context/data_context/abstract_data_context.py
+++ b/great_expectations/data_context/data_context/abstract_data_context.py
@@ -1464,6 +1464,78 @@ class AbstractDataContext(ConfigPeer, ABC):
         self._cached_datasources.pop(datasource_name, None)
         self.config.datasources.pop(datasource_name, None)  # type: ignore[union-attr]
 
+    @overload
+    def add_checkpoint(
+        self,
+        name: str = ...,
+        config_version: int | float | None = ...,
+        template_name: str | None = ...,
+        module_name: str | None = ...,
+        class_name: str | None = ...,
+        run_name_template: str | None = ...,
+        expectation_suite_name: str | None = ...,
+        batch_request: dict | None = ...,
+        action_list: list[dict] | None = ...,
+        evaluation_parameters: dict | None = ...,
+        runtime_configuration: dict | None = ...,
+        validations: list[dict] | None = ...,
+        profilers: list[dict] | None = ...,
+        # Next two fields are for LegacyCheckpoint configuration
+        validation_operator_name: str | None = ...,
+        batches: list[dict] | None = ...,
+        # the following four arguments are used by SimpleCheckpoint
+        site_names: str | list[str] | None = ...,
+        slack_webhook: str | None = ...,
+        notify_on: str | None = ...,
+        notify_with: str | list[str] | None = ...,
+        ge_cloud_id: str | None = ...,
+        expectation_suite_ge_cloud_id: str | None = ...,
+        default_validation_id: str | None = ...,
+        id: str | None = ...,
+        expectation_suite_id: str | None = ...,
+        checkpoint: None = ...,
+    ) -> Checkpoint:
+        """
+        Individual constructor arguments are provided.
+        `checkpoint` should not be provided.
+        """
+        ...
+
+    @overload
+    def add_checkpoint(
+        self,
+        name: None = ...,
+        config_version: None = ...,
+        template_name: None = ...,
+        module_name: None = ...,
+        class_name: None = ...,
+        run_name_template: None = ...,
+        expectation_suite_name: None = ...,
+        batch_request: None = ...,
+        action_list: None = ...,
+        evaluation_parameters: None = ...,
+        runtime_configuration: None = ...,
+        validations: None = ...,
+        profilers: None = ...,
+        validation_operator_name: None = ...,
+        batches: None = ...,
+        site_names: None = ...,
+        slack_webhook: None = ...,
+        notify_on: None = ...,
+        notify_with: None = ...,
+        ge_cloud_id: None = ...,
+        expectation_suite_ge_cloud_id: None = ...,
+        default_validation_id: None = ...,
+        id: None = ...,
+        expectation_suite_id: None = ...,
+        checkpoint: Checkpoint = ...,
+    ) -> Checkpoint:
+        """
+        A `checkpoint` is provided.
+        Individual constructor arguments should not be provided.
+        """
+        ...
+
     @public_api
     @deprecated_argument(argument_name="validation_operator_name", version="0.14.0")
     @deprecated_argument(argument_name="batches", version="0.14.0")
@@ -1477,9 +1549,14 @@ class AbstractDataContext(ConfigPeer, ABC):
         version="0.15.48",
         message="To be used in place of `expectation_suite_ge_cloud_id`",
     )
+    @new_argument(
+        argument_name="checkpoint",
+        version="0.15.48",
+        message="Pass in an existing checkpoint instead of individual constructor args",
+    )
     def add_checkpoint(
         self,
-        name: str,
+        name: str | None = None,
         config_version: int | float | None = None,
         template_name: str | None = None,
         module_name: str | None = None,
@@ -1505,6 +1582,7 @@ class AbstractDataContext(ConfigPeer, ABC):
         default_validation_id: str | None = None,
         id: str | None = None,
         expectation_suite_id: str | None = None,
+        checkpoint: Checkpoint | None = None,
     ) -> Checkpoint:
         """Add a Checkpoint to the DataContext.
 
@@ -1536,6 +1614,7 @@ class AbstractDataContext(ConfigPeer, ABC):
             default_validation_id: The default validation ID to use in generating this checkpoint.
             id: The ID to use in generating this checkpoint (preferred over `ge_cloud_id`).
             expectation_suite_id: The expectation suite ID to use in generating this checkpoint (preferred over `expectation_suite_ge_cloud_id`).
+            checkpoint: An existing checkpoint you wish to persist.
 
         Returns:
             The Checkpoint object created.
@@ -1548,12 +1627,9 @@ class AbstractDataContext(ConfigPeer, ABC):
         del ge_cloud_id
         del expectation_suite_ge_cloud_id
 
-        from great_expectations.checkpoint.checkpoint import Checkpoint
-
-        checkpoint: Checkpoint = Checkpoint.construct_from_config_args(
-            data_context=self,
-            checkpoint_store_name=self.checkpoint_store_name,  # type: ignore[arg-type]
+        checkpoint = self._resolve_add_checkpoint_args(
             name=name,
+            id=id,
             config_version=config_version,
             template_name=template_name,
             module_name=module_name,
@@ -1566,17 +1642,15 @@ class AbstractDataContext(ConfigPeer, ABC):
             runtime_configuration=runtime_configuration,
             validations=validations,
             profilers=profilers,
-            # Next two fields are for LegacyCheckpoint configuration
             validation_operator_name=validation_operator_name,
             batches=batches,
-            # the following four arguments are used by SimpleCheckpoint
             site_names=site_names,
             slack_webhook=slack_webhook,
             notify_on=notify_on,
             notify_with=notify_with,
-            ge_cloud_id=id,
-            expectation_suite_ge_cloud_id=expectation_suite_id,
+            expectation_suite_id=expectation_suite_id,
             default_validation_id=default_validation_id,
+            checkpoint=checkpoint,
         )
 
         return self.checkpoint_store.add_checkpoint(checkpoint)
@@ -1597,11 +1671,83 @@ class AbstractDataContext(ConfigPeer, ABC):
         """
         return self.checkpoint_store.update_checkpoint(checkpoint)
 
+    @overload
+    def add_or_update_checkpoint(
+        self,
+        name: str = ...,
+        config_version: int | float | None = ...,
+        template_name: str | None = ...,
+        module_name: str | None = ...,
+        class_name: str | None = ...,
+        run_name_template: str | None = ...,
+        expectation_suite_name: str | None = ...,
+        batch_request: dict | None = ...,
+        action_list: list[dict] | None = ...,
+        evaluation_parameters: dict | None = ...,
+        runtime_configuration: dict | None = ...,
+        validations: list[dict] | None = ...,
+        profilers: list[dict] | None = ...,
+        # Next two fields are for LegacyCheckpoint configuration
+        validation_operator_name: str | None = ...,
+        batches: list[dict] | None = ...,
+        # the following four arguments are used by SimpleCheckpoint
+        site_names: str | list[str] | None = ...,
+        slack_webhook: str | None = ...,
+        notify_on: str | None = ...,
+        notify_with: str | list[str] | None = ...,
+        ge_cloud_id: str | None = ...,
+        expectation_suite_ge_cloud_id: str | None = ...,
+        default_validation_id: str | None = ...,
+        id: str | None = ...,
+        expectation_suite_id: str | None = ...,
+        checkpoint: None = ...,
+    ) -> Checkpoint:
+        """
+        Individual constructor arguments are provided.
+        `checkpoint` should not be provided.
+        """
+        ...
+
+    @overload
+    def add_or_update_checkpoint(
+        self,
+        name: None = ...,
+        config_version: None = ...,
+        template_name: None = ...,
+        module_name: None = ...,
+        class_name: None = ...,
+        run_name_template: None = ...,
+        expectation_suite_name: None = ...,
+        batch_request: None = ...,
+        action_list: None = ...,
+        evaluation_parameters: None = ...,
+        runtime_configuration: None = ...,
+        validations: None = ...,
+        profilers: None = ...,
+        validation_operator_name: None = ...,
+        batches: None = ...,
+        site_names: None = ...,
+        slack_webhook: None = ...,
+        notify_on: None = ...,
+        notify_with: None = ...,
+        ge_cloud_id: None = ...,
+        expectation_suite_ge_cloud_id: None = ...,
+        default_validation_id: None = ...,
+        id: None = ...,
+        expectation_suite_id: None = ...,
+        checkpoint: Checkpoint = ...,
+    ) -> Checkpoint:
+        """
+        A `checkpoint` is provided.
+        Individual constructor arguments should not be provided.
+        """
+        ...
+
     @public_api
     @new_method_or_class(version="0.15.48")
     def add_or_update_checkpoint(  # noqa: C901 - Complexity 23
         self,
-        name: str,
+        name: str | None = None,
         id: str | None = None,
         config_version: int | float | None = None,
         template_name: str | None = None,
@@ -1622,6 +1768,7 @@ class AbstractDataContext(ConfigPeer, ABC):
         notify_with: str | list[str] | None = None,
         expectation_suite_id: str | None = None,
         default_validation_id: str | None = None,
+        checkpoint: Checkpoint | None = None,
     ) -> Checkpoint:
         """Add a new Checkpoint or update an existing one on the context depending on whether it already exists or not.
 
@@ -1646,16 +1793,14 @@ class AbstractDataContext(ConfigPeer, ABC):
             notify_with: The notify with setting to use in generating this checkpoint. This is only used for SimpleCheckpoint configuration.
             expectation_suite_id: The expectation suite GE Cloud ID to use in generating this checkpoint.
             default_validation_id: The default validation ID to use in generating this checkpoint.
+            checkpoint: An existing checkpoint you wish to persist.
 
         Returns:
             A new Checkpoint or an updated once (depending on whether or not it existed before this method call).
         """
-        from great_expectations.checkpoint.checkpoint import Checkpoint
-
-        checkpoint = Checkpoint.construct_from_config_args(
-            data_context=self,
-            checkpoint_store_name=self.checkpoint_store_name,  # type: ignore[arg-type]
+        checkpoint = self._resolve_add_checkpoint_args(
             name=name,
+            id=id,
             config_version=config_version,
             template_name=template_name,
             module_name=module_name,
@@ -1668,17 +1813,83 @@ class AbstractDataContext(ConfigPeer, ABC):
             runtime_configuration=runtime_configuration,
             validations=validations,
             profilers=profilers,
-            # the following four arguments are used by SimpleCheckpoint
             site_names=site_names,
             slack_webhook=slack_webhook,
             notify_on=notify_on,
             notify_with=notify_with,
-            ge_cloud_id=id,
-            expectation_suite_ge_cloud_id=expectation_suite_id,
+            expectation_suite_id=expectation_suite_id,
             default_validation_id=default_validation_id,
+            checkpoint=checkpoint,
         )
+
         return self.checkpoint_store.add_or_update_checkpoint(checkpoint)
 
+    def _resolve_add_checkpoint_args(
+        self,
+        name: str | None = None,
+        id: str | None = None,
+        config_version: int | float | None = None,
+        template_name: str | None = None,
+        module_name: str | None = None,
+        class_name: str | None = None,
+        run_name_template: str | None = None,
+        expectation_suite_name: str | None = None,
+        batch_request: dict | None = None,
+        action_list: list[dict] | None = None,
+        evaluation_parameters: dict | None = None,
+        runtime_configuration: dict | None = None,
+        validations: list[dict] | None = None,
+        profilers: list[dict] | None = None,
+        validation_operator_name: str | None = None,
+        batches: list[dict] | None = None,
+        site_names: str | list[str] | None = None,
+        slack_webhook: str | None = None,
+        notify_on: str | None = None,
+        notify_with: str | list[str] | None = None,
+        expectation_suite_id: str | None = None,
+        default_validation_id: str | None = None,
+        checkpoint: Checkpoint | None = None,
+    ) -> Checkpoint:
+        from great_expectations.checkpoint.checkpoint import Checkpoint
+
+        if not ((checkpoint is None) ^ (name is None)):
+            raise ValueError(
+                "Must either pass in an existing checkpoint or individual constructor arguments (but not both)"
+            )
+
+        if not checkpoint:
+            assert (
+                name
+            ), "Guaranteed to have a non-null name if constructing Checkpoint with individual args"
+            checkpoint = Checkpoint.construct_from_config_args(
+                data_context=self,
+                checkpoint_store_name=self.checkpoint_store_name,  # type: ignore[arg-type]
+                name=name,
+                config_version=config_version,
+                template_name=template_name,
+                module_name=module_name,
+                class_name=class_name,
+                run_name_template=run_name_template,
+                expectation_suite_name=expectation_suite_name,
+                batch_request=batch_request,
+                action_list=action_list,
+                evaluation_parameters=evaluation_parameters,
+                runtime_configuration=runtime_configuration,
+                validations=validations,
+                profilers=profilers,
+                validation_operator_name=validation_operator_name,
+                batches=batches,
+                site_names=site_names,
+                slack_webhook=slack_webhook,
+                notify_on=notify_on,
+                notify_with=notify_with,
+                ge_cloud_id=id,
+                expectation_suite_ge_cloud_id=expectation_suite_id,
+                default_validation_id=default_validation_id,
+            )
+
+        return checkpoint
+
     @public_api
     @new_argument(
         argument_name="id",
diff --git a/great_expectations/data_context/data_context/cloud_data_context.py b/great_expectations/data_context/data_context/cloud_data_context.py
index 0645d0a8a..ac036ebb8 100644
--- a/great_expectations/data_context/data_context/cloud_data_context.py
+++ b/great_expectations/data_context/data_context/cloud_data_context.py
@@ -632,41 +632,48 @@ class CloudDataContext(SerializableDataContext):
 
     def add_checkpoint(
         self,
-        name: str,
-        config_version: Optional[Union[int, float]] = None,
-        template_name: Optional[str] = None,
-        module_name: Optional[str] = None,
-        class_name: Optional[str] = None,
-        run_name_template: Optional[str] = None,
-        expectation_suite_name: Optional[str] = None,
-        batch_request: Optional[dict] = None,
-        action_list: Optional[List[dict]] = None,
-        evaluation_parameters: Optional[dict] = None,
-        runtime_configuration: Optional[dict] = None,
-        validations: Optional[List[dict]] = None,
-        profilers: Optional[List[dict]] = None,
+        name: str | None = None,
+        config_version: int | float | None = None,
+        template_name: str | None = None,
+        module_name: str | None = None,
+        class_name: str | None = None,
+        run_name_template: str | None = None,
+        expectation_suite_name: str | None = None,
+        batch_request: dict | None = None,
+        action_list: list[dict] | None = None,
+        evaluation_parameters: dict | None = None,
+        runtime_configuration: dict | None = None,
+        validations: list[dict] | None = None,
+        profilers: list[dict] | None = None,
         # Next two fields are for LegacyCheckpoint configuration
-        validation_operator_name: Optional[str] = None,
-        batches: Optional[List[dict]] = None,
+        validation_operator_name: str | None = None,
+        batches: list[dict] | None = None,
         # the following four arguments are used by SimpleCheckpoint
-        site_names: Optional[Union[str, List[str]]] = None,
-        slack_webhook: Optional[str] = None,
-        notify_on: Optional[str] = None,
-        notify_with: Optional[Union[str, List[str]]] = None,
-        ge_cloud_id: Optional[str] = None,
-        expectation_suite_ge_cloud_id: Optional[str] = None,
-        default_validation_id: Optional[str] = None,
+        site_names: str | list[str] | None = None,
+        slack_webhook: str | None = None,
+        notify_on: str | None = None,
+        notify_with: str | list[str] | None = None,
+        ge_cloud_id: str | None = None,
+        expectation_suite_ge_cloud_id: str | None = None,
+        default_validation_id: str | None = None,
+        id: str | None = None,
+        expectation_suite_id: str | None = None,
+        checkpoint: Checkpoint | None = None,
     ) -> Checkpoint:
         """
         See `AbstractDataContext.add_checkpoint` for more information.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        expectation_suite_id = self._resolve_id_and_ge_cloud_id(
+            id=expectation_suite_id, ge_cloud_id=expectation_suite_ge_cloud_id
+        )
+        del ge_cloud_id
+        del expectation_suite_ge_cloud_id
 
-        from great_expectations.checkpoint.checkpoint import Checkpoint
-
-        checkpoint: Checkpoint = Checkpoint.construct_from_config_args(
-            data_context=self,
-            checkpoint_store_name=self.checkpoint_store_name,  # type: ignore[arg-type]
+        checkpoint = self._resolve_add_checkpoint_args(
             name=name,
+            id=id,
             config_version=config_version,
             template_name=template_name,
             module_name=module_name,
@@ -679,27 +686,26 @@ class CloudDataContext(SerializableDataContext):
             runtime_configuration=runtime_configuration,
             validations=validations,
             profilers=profilers,
-            # Next two fields are for LegacyCheckpoint configuration
             validation_operator_name=validation_operator_name,
             batches=batches,
-            # the following four arguments are used by SimpleCheckpoint
             site_names=site_names,
             slack_webhook=slack_webhook,
             notify_on=notify_on,
             notify_with=notify_with,
-            ge_cloud_id=ge_cloud_id,
-            expectation_suite_ge_cloud_id=expectation_suite_ge_cloud_id,
+            expectation_suite_id=expectation_suite_id,
             default_validation_id=default_validation_id,
+            checkpoint=checkpoint,
         )
 
         checkpoint_config = self.checkpoint_store.create(
             checkpoint_config=checkpoint.config
         )
 
-        checkpoint = Checkpoint.instantiate_from_config_with_runtime_args(
+        from great_expectations.checkpoint.checkpoint import Checkpoint
+
+        return Checkpoint.instantiate_from_config_with_runtime_args(
             checkpoint_config=checkpoint_config, data_context=self  # type: ignore[arg-type]
         )
-        return checkpoint
 
     def list_checkpoints(self) -> Union[List[str], List[ConfigurationIdentifier]]:
         return self.checkpoint_store.list_checkpoints(ge_cloud_mode=True)


Test patch:
diff --git a/tests/data_context/test_data_context_state_management.py b/tests/data_context/test_data_context_state_management.py
index 1c41601e3..7d0d63013 100644
--- a/tests/data_context/test_data_context_state_management.py
+++ b/tests/data_context/test_data_context_state_management.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from typing import Mapping
+from unittest import mock
 
 import pytest
 
@@ -688,6 +689,53 @@ def test_add_or_update_profiler_updates_successfully(
     assert context.profiler_store.save_count == 2
 
 
+@pytest.mark.unit
+def test_add_checkpoint_with_existing_checkpoint(
+    in_memory_data_context: EphemeralDataContextSpy,
+):
+    context = in_memory_data_context
+    checkpoint_name = "my_checkpoint"
+    checkpoint = Checkpoint(name=checkpoint_name, data_context=context)
+
+    persisted_checkpoint = context.add_checkpoint(checkpoint=checkpoint)
+
+    assert checkpoint == persisted_checkpoint
+    assert context.checkpoint_store.save_count == 1
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    "checkpoint, checkpoint_name",
+    [
+        pytest.param(
+            mock.MagicMock(),
+            "my_checkpoint_name",
+            id="both checkpoint and checkpoint_name",
+        ),
+        pytest.param(None, None, id="neither checkpoint nor checkpoint_name"),
+    ],
+)
+def test_add_checkpoint_conflicting_args_failure(
+    in_memory_data_context: EphemeralDataContextSpy,
+    # Only care about the presence of the value (no need to construct a full Checkpoint obj)
+    checkpoint: mock.MagicMock | None,
+    checkpoint_name: str | None,
+):
+    context = in_memory_data_context
+
+    with pytest.raises(ValueError) as e:
+        context.add_checkpoint(
+            checkpoint=checkpoint,
+            name=checkpoint_name,
+        )
+
+    assert (
+        "an existing checkpoint or individual constructor arguments (but not both)"
+        in str(e.value)
+    )
+    assert context.checkpoint_store.save_count == 0
+
+
 @pytest.mark.unit
 def test_update_checkpoint_success(
     in_memory_data_context: EphemeralDataContextSpy,
@@ -744,19 +792,24 @@ def test_update_checkpoint_failure(in_memory_data_context: EphemeralDataContextS
 
 
 @pytest.mark.unit
+@pytest.mark.parametrize("use_existing_checkpoint", [True, False])
 def test_add_or_update_checkpoint_adds_successfully(
     in_memory_data_context: EphemeralDataContextSpy,
     checkpoint_config: dict,
+    use_existing_checkpoint: bool,
 ):
     context = in_memory_data_context
-    name = "my_checkpoint"
-    checkpoint_config["name"] = name
 
-    checkpoint = context.add_or_update_checkpoint(**checkpoint_config)
+    if use_existing_checkpoint:
+        checkpoint_config.pop("class_name")
+        checkpoint = Checkpoint(**checkpoint_config, data_context=context)
+        checkpoint = context.add_or_update_checkpoint(checkpoint=checkpoint)
+    else:
+        checkpoint = context.add_or_update_checkpoint(**checkpoint_config)
 
     actual_config = checkpoint.config
 
-    assert actual_config.name == name
+    assert actual_config.name == checkpoint_config["name"]
     assert (
         actual_config.expectation_suite_name
         == checkpoint_config["expectation_suite_name"]
@@ -854,7 +907,7 @@ def test_add_or_update_checkpoint_adds_successfully(
         ),
     ],
 )
-def test_add_or_update_checkpoint_updates_successfully(
+def test_add_or_update_checkpoint_individual_args_updates_successfully(
     in_memory_data_context: EphemeralDataContextSpy,
     checkpoint_config: dict,
     update_kwargs: dict,
@@ -872,3 +925,59 @@ def test_add_or_update_checkpoint_updates_successfully(
 
     assert checkpoint.config.to_dict() == expected_config.to_dict()
     assert context.checkpoint_store.save_count == 2
+
+
+@pytest.mark.unit
+def test_add_or_update_checkpoint_existing_checkpoint_updates_successfully(
+    in_memory_data_context: EphemeralDataContextSpy,
+    checkpoint_config: dict,
+):
+    context = in_memory_data_context
+
+    name = "my_checkpoint"
+    checkpoint_config["name"] = name
+
+    checkpoint = context.add_checkpoint(name=name, class_name="Checkpoint")
+
+    assert len(checkpoint.validations) == 0
+    assert context.checkpoint_store.save_count == 1
+
+    checkpoint_config.pop("class_name")
+    checkpoint = Checkpoint(**checkpoint_config, data_context=context)
+    checkpoint = context.add_or_update_checkpoint(checkpoint=checkpoint)
+
+    assert len(checkpoint.validations) == len(checkpoint_config["validations"])
+    assert context.checkpoint_store.save_count == 2
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    "checkpoint, checkpoint_name",
+    [
+        pytest.param(
+            mock.MagicMock(),
+            "my_checkpoint_name",
+            id="both checkpoint and checkpoint_name",
+        ),
+        pytest.param(None, None, id="neither checkpoint nor checkpoint_name"),
+    ],
+)
+def test_add_or_update_checkpoint_conflicting_args_failure(
+    in_memory_data_context: EphemeralDataContextSpy,
+    # Only care about the presence of the value (no need to construct a full Checkpoint obj)
+    checkpoint: mock.MagicMock | None,
+    checkpoint_name: str | None,
+):
+    context = in_memory_data_context
+
+    with pytest.raises(ValueError) as e:
+        context.add_or_update_checkpoint(
+            checkpoint=checkpoint,
+            name=checkpoint_name,
+        )
+
+    assert (
+        "an existing checkpoint or individual constructor arguments (but not both)"
+        in str(e.value)
+    )
+    assert context.checkpoint_store.save_count == 0