content: #  Copyright (c) ZenML GmbH 2022. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  import atexit
  import json
  import os
  import random
  import time
  import uuid
  from contextlib import ExitStack as does_not_raise
  from datetime import datetime
  from string import ascii_lowercase
  from threading import Thread
  from typing import Dict, List, Optional, Tuple, Union
  from unittest.mock import patch
  from uuid import UUID, uuid4
  
  import pytest
  from pydantic import ValidationError
  from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
  
  from tests.integration.functional.utils import sample_name
  from tests.integration.functional.zen_stores.utils import (
      CodeRepositoryContext,
      ComponentContext,
      CrudTestConfig,
      LoginContext,
      ModelContext,
      PipelineRunContext,
      SecretContext,
      ServiceAccountContext,
      ServiceConnectorContext,
      ServiceConnectorTypeContext,
      StackContext,
      UserContext,
      list_of_entities,
  )
  from tests.unit.pipelines.test_build_utils import (
      StubLocalRepositoryContext,
  )
  from zenml import Model, Tag, add_tags, log_metadata, pipeline, step
  from zenml.client import Client
  from zenml.code_repositories import BaseCodeRepository
  from zenml.config.pipeline_configurations import PipelineConfiguration
  from zenml.config.source import Source, SourceType
  from zenml.config.step_configurations import Step, StepConfiguration, StepSpec
  from zenml.constants import (
      ACTIVATE,
      DEACTIVATE,
      DEFAULT_PROJECT_NAME,
      DEFAULT_STACK_AND_COMPONENT_NAME,
      DEFAULT_USERNAME,
      USERS,
  )
  from zenml.enums import (
      ArtifactSaveType,
      ArtifactType,
      ColorVariants,
      CuratedVisualizationSize,
      ExecutionStatus,
      MetadataResourceTypes,
      ModelStages,
      StackComponentType,
      StoreType,
      TaggableResourceTypes,
      VisualizationResourceTypes,
      VisualizationType,
  )
  from zenml.exceptions import (
      AuthorizationException,
      EntityExistsError,
      IllegalOperationError,
  )
  from zenml.metadata.metadata_types import MetadataTypeEnum
  from zenml.models import (
      APIKeyFilter,
      APIKeyRequest,
      APIKeyRotateRequest,
      APIKeyUpdate,
      ArtifactRequest,
      ArtifactVersionFilter,
      ArtifactVersionRequest,
      ArtifactVersionResponse,
      ArtifactVisualizationRequest,
      ComponentFilter,
      ComponentRequest,
      ComponentUpdate,
      CuratedVisualizationRequest,
      CuratedVisualizationUpdate,
      DeploymentRequest,
      ModelFilter,
      ModelRequest,
      ModelUpdate,
      ModelVersionArtifactFilter,
      ModelVersionArtifactRequest,
      ModelVersionFilter,
      ModelVersionPipelineRunFilter,
      ModelVersionPipelineRunRequest,
      ModelVersionRequest,
      ModelVersionUpdate,
      PipelineRequest,
      PipelineRunFilter,
      PipelineRunRequest,
      PipelineRunResponse,
      PipelineSnapshotRequest,
      ProjectFilter,
      ProjectUpdate,
      RunMetadataRequest,
      RunMetadataResource,
      ScheduleRequest,
      ServiceAccountFilter,
      ServiceAccountRequest,
      ServiceAccountUpdate,
      ServiceConnectorConfiguration,
      ServiceConnectorFilter,
      ServiceConnectorUpdate,
      StackFilter,
      StackRequest,
      StackUpdate,
      StepRunFilter,
      StepRunRequest,
      StepRunUpdate,
      TagResourceRequest,
      UserFilter,
      UserRequest,
      UserResponse,
      UserUpdate,
  )
  from zenml.utils import code_repository_utils, source_utils
  from zenml.utils.enum_utils import StrEnum
  from zenml.zen_stores.rest_zen_store import RestZenStore
  from zenml.zen_stores.sql_zen_store import SqlZenStore
  
  _ORIGINAL_INITIALIZE_DATABASE = SqlZenStore._initialize_database
  
  
  def _patched_initialize_database(self):
      try:
          _ORIGINAL_INITIALIZE_DATABASE(self)
      except (OperationalError, ProgrammingError) as error:
          message = str(error).lower()
          if (
              "no such column: stack.environment" in message
              or "unknown column 'stack.environment'" in message
          ):
              self.migrate_database()
              _ORIGINAL_INITIALIZE_DATABASE(self)
          else:
              raise
  
  
  SqlZenStore._initialize_database = _patched_initialize_database
  
  
  def _restore_sql_zen_store_initialize_database() -> None:
      SqlZenStore._initialize_database = _ORIGINAL_INITIALIZE_DATABASE
  
  
  atexit.register(_restore_sql_zen_store_initialize_database)
  
  DEFAULT_NAME = "default"
  
  # .--------------.
  # | GENERIC CRUD |
  # '--------------'
  
  
  @pytest.mark.parametrize(
      "crud_test_config",
      list_of_entities,
      ids=[e.entity_name for e in list_of_entities],
  )
  def test_basic_crud_for_entity(crud_test_config: CrudTestConfig):
      """Tests the basic crud operations for a given entity."""
  
      zs = Client().zen_store
  
      if not isinstance(zs, tuple(crud_test_config.supported_zen_stores)):
          pytest.skip(
              f"Test only applies to "
              f"{[c.__name__ for c in crud_test_config.supported_zen_stores]}"
          )
  
      # Test the creation
      created_entity = crud_test_config.create()
  
      # Test that the create method returns a hydrated model, if applicable
      if hasattr(created_entity, "metadata"):
          assert created_entity.metadata is not None
  
      # Test the list method
      entities_list = crud_test_config.list_method(
          crud_test_config.filter_model(id=created_entity.id)
      )
      assert entities_list.total == 1
      entity = entities_list.items[0]
      assert entity == created_entity
  
      # Test that the list method returns a non-hydrated model, if applicable
      if hasattr(entity, "metadata"):
          assert entity.metadata is None
  
          # Try to hydrate the entity
          entity.get_metadata()
          assert entity.metadata is not None
  
          # Test that the list method has a `hydrate` argument
          entities_list = crud_test_config.list_method(
              crud_test_config.filter_model(id=created_entity.id),
              hydrate=True,
          )
          assert entities_list.total == 1
          entity = entities_list.items[0]
          assert entity.metadata is not None
  
      # Test filtering by name if applicable
      if "name" in created_entity.model_fields and created_entity.name:
          entities_list = crud_test_config.list_method(
              crud_test_config.filter_model(name=created_entity.name)
          )
          assert entities_list.total == 1
          entity = entities_list.items[0]
          assert entity == created_entity
  
      # Test the get method
      entity = crud_test_config.get_method(created_entity.id)
      assert entity == created_entity
  
      # Test that the get method returns a hydrated model, if applicable
      if hasattr(entity, "metadata"):
          assert entity.metadata is not None
  
          # Test that the get method has a `hydrate` argument
          unhydrated_entity = crud_test_config.get_method(
              created_entity.id, hydrate=False
          )
          assert unhydrated_entity.metadata is None
  
          # Try to hydrate the entity
          unhydrated_entity.get_metadata()
          assert unhydrated_entity.metadata is not None
  
      # Test the update method if applicable
      if crud_test_config.update_model:
          updated_entity = crud_test_config.update()
          # Ids should remain the same
          assert updated_entity.id == created_entity.id
          # Something in the Model should have changed
          assert (
              updated_entity.model_dump_json()
              != created_entity.model_dump_json()
          )
  
          # Test that the update method returns a hydrated model, if applicable
          if hasattr(updated_entity, "metadata"):
              assert updated_entity.metadata is not None
  
      # Test the delete method
      crud_test_config.delete()
      with pytest.raises(KeyError):
          crud_test_config.get_method(created_entity.id)
      entities_list = crud_test_config.list_method(
          crud_test_config.filter_model(id=created_entity.id)
      )
      assert entities_list.total == 0
  
      # Cleanup
      crud_test_config.cleanup()
  
  
  @pytest.mark.parametrize(
      "crud_test_config",
      list_of_entities,
      ids=[e.entity_name for e in list_of_entities],
  )
  def test_create_entity_twice_fails(crud_test_config: CrudTestConfig):
      """Tests getting a non-existent entity by id."""
      zs = Client().zen_store
  
      if not isinstance(zs, tuple(crud_test_config.supported_zen_stores)):
          pytest.skip(
              f"Test only applies to "
              f"{[c.__name__ for c in crud_test_config.supported_zen_stores]}"
          )
  
      entity_name = crud_test_config.entity_name
      if entity_name in {"build", "snapshot"}:
          pytest.skip(f"Duplicates of {entity_name} are allowed.")
  
      # First creation is successful
      crud_test_config.create()
  
      # Second one fails
      with pytest.raises(EntityExistsError):
          crud_test_config.create()
  
      # Cleanup
      crud_test_config.cleanup()
  
  
  @pytest.mark.parametrize(
      "crud_test_config",
      list_of_entities,
      ids=[e.entity_name for e in list_of_entities],
  )
  def test_get_nonexistent_entity_fails(crud_test_config: CrudTestConfig):
      """Tests getting a non-existent entity by id."""
      zs = Client().zen_store
  
      if not isinstance(zs, tuple(crud_test_config.supported_zen_stores)):
          pytest.skip(
              f"Test only applies to "
              f"{[c.__name__ for c in crud_test_config.supported_zen_stores]}"
          )
  
      with pytest.raises(KeyError):
          crud_test_config.get_method(uuid.uuid4())
  
  
  @pytest.mark.parametrize(
      "crud_test_config",
      list_of_entities,
      ids=[e.entity_name for e in list_of_entities],
  )
  def test_updating_nonexisting_entity_raises_error(
      crud_test_config: CrudTestConfig,
  ):
      """Tests updating a nonexistent entity raises an error."""
      zs = Client().zen_store
  
      if not isinstance(zs, tuple(crud_test_config.supported_zen_stores)):
          pytest.skip(
              f"Test only applies to "
              f"{[c.__name__ for c in crud_test_config.supported_zen_stores]}"
          )
  
      if crud_test_config.update_model:
          # Update the created entity
          update_model = crud_test_config.update_model
          with pytest.raises(KeyError):
              crud_test_config.update_method(uuid.uuid4(), update_model)
      else:
          pytest.skip(
              "For entities that do not support updates, this test is not run."
          )
  
  
  @pytest.mark.parametrize(
      "crud_test_config",
      list_of_entities,
      ids=[e.entity_name for e in list_of_entities],
  )
  def test_deleting_nonexistent_entity_raises_error(
      crud_test_config: CrudTestConfig,
  ):
      """Tests deleting a nonexistent resource raises an error."""
      zs = Client().zen_store
  
      if not isinstance(zs, tuple(crud_test_config.supported_zen_stores)):
          pytest.skip(
              f"Test only applies to "
              f"{[c.__name__ for c in crud_test_config.supported_zen_stores]}"
          )
  
      with pytest.raises(KeyError):
          crud_test_config.delete_method(uuid.uuid4())
  
  
  # .----------.
  # | PROJECTS |
  # '----------'
  
  
  def test_only_one_default_project_present():
      """Tests that one and only one default project is present."""
      client = Client()
      assert (
          len(client.zen_store.list_projects(ProjectFilter(name="default"))) == 1
      )
  
  
  def test_updating_default_project_fails():
      """Tests updating the default project."""
      client = Client()
  
      default_project = client.zen_store.get_project(DEFAULT_PROJECT_NAME)
      assert default_project.name == DEFAULT_PROJECT_NAME
      project_update = ProjectUpdate(
          name="aria_project",
          description="Aria has taken possession of this project.",
      )
      with pytest.raises(IllegalOperationError):
          client.zen_store.update_project(
              project_id=default_project.id,
              project_update=project_update,
          )
  
  
  def test_deleting_default_project_fails():
      """Tests deleting the default project."""
      client = Client()
      with pytest.raises(IllegalOperationError):
          client.zen_store.delete_project(DEFAULT_NAME)
  
  
  #  .------.
  # | USERS |
  # '-------'
  
  
  class TestAdminUser:
      default_pwd = "".join(random.choices(ascii_lowercase, k=10))
  
      def test_creation_as_admin_and_non_admin(self):
          """Tests creating a user as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          with UserContext(login=True, is_admin=False):
              zen_store: RestZenStore = Client().zen_store
              # this is not allowed for non-admin users
              with pytest.raises(IllegalOperationError):
                  zen_store.create_user(
                      UserRequest(
                          name=sample_name("test_user"),
                          password=self.default_pwd,
                          is_admin=False,
                      )
                  )
  
      def test_listing_users(self):
          """Tests listing users as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              zen_store: RestZenStore = Client().zen_store
              users = zen_store.list_users(UserFilter())
              assert users.total >= 2
  
              # this is limited to self only for non-admin users
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  zen_store = Client().zen_store
                  users = zen_store.list_users(UserFilter())
                  assert users.total == 1
  
      def test_get_users(self):
          """Tests getting users as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              zen_store: RestZenStore = Client().zen_store
  
              user = zen_store.get_user(test_user.name)
              assert user.id == test_user.id
  
              # this is not allowed for non-admin users
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  zen_store = Client().zen_store
                  with pytest.raises(IllegalOperationError):
                      zen_store.get_user(DEFAULT_USERNAME)
  
      def test_update_users(self):
          """Tests updating users as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              zen_store: RestZenStore = Client().zen_store
  
              user = zen_store.update_user(
                  test_user.id,
                  UserUpdate(full_name="foo@bar.ai"),
              )
              assert user.full_name == "foo@bar.ai"
  
              with UserContext(login=True, is_admin=False):
                  zen_store = Client().zen_store
  
                  # this is not allowed for non-admin users
                  with pytest.raises(IllegalOperationError):
                      zen_store.update_user(
                          test_user.id,
                          UserUpdate(full_name="bar@foo.io"),
                      )
  
              # user is allowed to update itself
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  zen_store = Client().zen_store
                  user = zen_store.update_user(
                      test_user.id,
                      UserUpdate(full_name="bar@foo.io"),
                  )
  
                  assert user.full_name == "bar@foo.io"
  
      def test_deactivate_users(self):
          """Tests deactivating users as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          zen_store: RestZenStore = Client().zen_store
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              with UserContext(is_admin=False) as test_user2:
                  # this is not allowed for non-admin users
                  with LoginContext(
                      user_name=test_user.name, password=self.default_pwd
                  ):
                      new_zen_store: RestZenStore = Client().zen_store
                      with pytest.raises(IllegalOperationError):
                          new_zen_store.put(
                              f"{USERS}/{str(test_user2.id)}{DEACTIVATE}"
                          )
  
                  response_body = zen_store.put(
                      f"{USERS}/{str(test_user2.id)}{DEACTIVATE}",
                  )
                  deactivated_user = UserResponse.model_validate(response_body)
                  assert deactivated_user.name == test_user2.name
  
      def test_delete_users(self):
          """Tests deleting users as an admin and as a non-admin."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          zen_store: RestZenStore = Client().zen_store
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              with UserContext(is_admin=False) as test_user2:
                  # this is not allowed for non-admin users
                  with LoginContext(
                      user_name=test_user.name, password=self.default_pwd
                  ):
                      new_zen_store: RestZenStore = Client().zen_store
                      with pytest.raises(IllegalOperationError):
                          new_zen_store.delete_user(test_user2.id)
  
              zen_store.delete_user(test_user.id)
  
      def test_update_self_via_normal_endpoint(self):
          """Tests updating self in admin and non-admin setting."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          zen_store: RestZenStore = Client().zen_store
          default_user = zen_store.get_user(DEFAULT_USERNAME)
          with pytest.raises(IllegalOperationError):
              # cannot update admin status for default user
              zen_store.update_user(
                  default_user.id,
                  UserUpdate(name=default_user.name, is_admin=False),
              )
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              # this is not allowed for non-admin users
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  new_zen_store: RestZenStore = Client().zen_store
                  with pytest.raises(IllegalOperationError):
                      new_zen_store.update_user(
                          test_user.id,
                          UserUpdate(
                              name=test_user.name,
                              is_admin=True,
                          ),
                      )
  
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              zen_store.update_user(
                  test_user.id,
                  UserUpdate(name=test_user.name, is_admin=True),
              )
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  new_zen_store: RestZenStore = Client().zen_store
                  with pytest.raises(IllegalOperationError):
                      new_zen_store.update_user(
                          test_user.id,
                          UserUpdate(
                              name=test_user.name,
                              is_admin=False,
                          ),
                      )
  
      def test_update_self_via_current_user_endpoint(self):
          """Tests updating self in admin and non-admin setting."""
          if Client().zen_store.type == StoreType.SQL:
              pytest.skip("SQL ZenStore does not support admin users.")
  
          zen_store: RestZenStore = Client().zen_store
          default_user = zen_store.get_user(DEFAULT_USERNAME)
          assert default_user.is_admin
          zen_store.put(
              "/current-user",
              body=UserUpdate(name=default_user.name, full_name="Axl"),
          )
          default_user = zen_store.get_user(DEFAULT_USERNAME)
          assert default_user.full_name == "Axl"
          assert default_user.is_admin  # admin status is not changed
  
          with UserContext(
              password=self.default_pwd, is_admin=False
          ) as test_user:
              with LoginContext(
                  user_name=test_user.name, password=self.default_pwd
              ):
                  new_zen_store: RestZenStore = Client().zen_store
                  assert not test_user.is_admin
                  new_zen_store.put(
                      "/current-user",
                      body=UserUpdate(
                          name=test_user.name,
                          full_name="Axl",
                      ),
                  )
                  user = zen_store.get_user(test_user.id)
                  assert not user.is_admin
                  assert user.full_name == "Axl"
  
                  # self update does not change admin status
                  new_zen_store.put(
                      "/current-user",
                      body=UserUpdate(
                          name=test_user.name,
                          is_admin=True,
                      ),
                  )
                  user = zen_store.get_user(test_user.id)
                  assert not user.is_admin
  
  
  def test_active_user():
      """Tests the active user can be queried with .get_user()."""
      zen_store = Client().zen_store
      active_user = zen_store.get_user()
      assert active_user is not None
      # The SQL zen_store only supports the default user as active user
      if zen_store.type == StoreType.SQL:
          assert active_user.name == DEFAULT_USERNAME
      else:
          # TODO: Implement this
          assert True
  
  
  def test_creating_user_with_existing_name_fails():
      """Tests creating a user with an existing username fails."""
      zen_store = Client().zen_store
  
      with UserContext() as existing_user:
          with pytest.raises(EntityExistsError):
              zen_store.create_user(
                  UserRequest(
                      name=existing_user.name,
                      password="password",
                      is_admin=False,
                  )
              )
  
      with ServiceAccountContext() as existing_service_account:
          with does_not_raise():
              user = zen_store.create_user(
                  UserRequest(
                      name=existing_service_account.name,
                      password="password",
                      is_admin=False,
                  )
              )
              # clean up
              zen_store.delete_user(user.id)
  
  
  def test_creating_users_in_parallel_do_not_duplicate_fails(
      clean_client: "Client",
  ):
      """Tests creating a user with an existing username fails in parallel mode."""
  
      def silent_create_user(user_request: UserRequest):
          """This function attempts to create a user and silently passes
          if a duplicate user exists. This is used to simulate race
          conditions in parallel user creation.
          """
          try:
              clean_client.zen_store.create_user(user_request)
          except (EntityExistsError, IntegrityError):
              pass
  
      user_name = "test_user"
      password = "P@ssw0rd"
      count = 100
  
      threads: List[Thread] = []
      for _ in range(count):
          t = Thread(
              target=silent_create_user,
              args=(
                  UserRequest(name=user_name, password=password, is_admin=False),
              ),
          )
          threads.append(t)
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      users = clean_client.zen_store.list_users(
          user_filter_model=UserFilter(name=user_name)
      )
      assert users.total == 1
  
  
  def test_creating_service_accounts_in_parallel_do_not_duplicate_fails(
      clean_client: "Client",
  ):
      """Tests creating a user with an existing username fails in parallel mode."""
  
      def silent_create_service_account(
          service_account_request: ServiceAccountRequest,
      ):
          """This function attempts to create a service account and silently
          passes if a duplicate user exists. This is used to simulate race
          conditions in parallel user creation.
          """
          try:
              clean_client.zen_store.create_service_account(
                  service_account_request
              )
          except (EntityExistsError, IntegrityError):
              pass
  
      user_name = "test_user"
      count = 100
  
      threads: List[Thread] = []
      for _ in range(count):
          t = Thread(
              target=silent_create_service_account,
              args=(ServiceAccountRequest(name=user_name, active=True),),
          )
          threads.append(t)
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      service_accounts = clean_client.zen_store.list_service_accounts(
          filter_model=ServiceAccountFilter(name=user_name)
      )
      assert service_accounts.total == 1
  
  
  def test_get_user():
      """Tests getting a user account."""
      zen_store = Client().zen_store
      with UserContext() as user_account:
          user = zen_store.get_user(user_account.name)
          assert user.id == user_account.id
          assert user.name == user_account.name
          assert user.active is True
          assert user.email == user_account.email
          assert user.is_service_account is False
          assert user.full_name == user_account.full_name
          assert user.email_opted_in == user_account.email_opted_in
  
          # Get a user account as a service account by ID is not possible
          with pytest.raises(KeyError):
              zen_store.get_service_account(user_account.id)
  
          # Get a user account as a service account by name is not possible
          with pytest.raises(KeyError):
              zen_store.get_service_account(user_account.name)
  
          with ServiceAccountContext(name=user_account.name) as service_account:
              # Get the service account as a user account by ID is allowed
              # for backwards compatibility
              user = zen_store.get_user(service_account.id)
              assert user.id == service_account.id
              assert user.name == service_account.name
              assert user.is_service_account is True
  
              # Getting the user by name returns the user, not the service account
              user = zen_store.get_user(user_account.name)
              assert user.id == user_account.id
              assert user.name == user_account.name
              assert user.is_service_account is False
  
  
  def test_delete_user_with_resources_fails():
      """Tests deleting a user with resources fails."""
      zen_store = Client().zen_store
  
      if Client().zen_store.type == StoreType.SQL:
          pytest.skip("SQL ZenStore does not support switching users.")
  
      with UserContext(delete=False, login=True) as user:
          component_context = ComponentContext(
              c_type=StackComponentType.ORCHESTRATOR,
              flavor="local",
              config={},
              delete=False,
          )
          with component_context as orchestrator:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      component_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
      with UserContext(delete=False, login=True) as user:
          orchestrator_context = ComponentContext(
              c_type=StackComponentType.ORCHESTRATOR,
              flavor="local",
              config={},
              delete=False,
          )
          artifact_store_context = ComponentContext(
              c_type=StackComponentType.ARTIFACT_STORE,
              flavor="local",
              config={},
              delete=False,
          )
  
          with orchestrator_context as orchestrator:
              # We only use the context as a shortcut to create the resource
              pass
          with artifact_store_context as artifact_store:
              # We only use the context as a shortcut to create the resource
              pass
  
          components = {
              StackComponentType.ORCHESTRATOR: [orchestrator.id],
              StackComponentType.ARTIFACT_STORE: [artifact_store.id],
          }
          stack_context = StackContext(components=components, delete=False)
          with stack_context:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      stack_context.cleanup()
      artifact_store_context.cleanup()
      orchestrator_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
      with UserContext(delete=False, login=True) as user:
          secret_context = SecretContext(delete=False)
          with secret_context:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      secret_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
      with UserContext(delete=False, login=True) as user:
          code_repo_context = CodeRepositoryContext(delete=False)
          with code_repo_context:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      code_repo_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
      with UserContext(delete=False, login=True) as user:
          service_connector_context = ServiceConnectorContext(
              connector_type="cat'o'matic",
              auth_method="paw-print",
              resource_types=["cat"],
              resource_id="aria",
              configuration={
                  "language": "meow",
                  "foods": "tuna",
              },
              delete=False,
          )
          with service_connector_context:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      service_connector_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
      with UserContext(delete=False, login=True) as user:
          model_context = ModelContext(create_version=True, delete=False)
          with model_context:
              # We only use the context as a shortcut to create the resource
              pass
  
      # Can't delete because owned resources exist
      with pytest.raises(IllegalOperationError):
          zen_store.delete_user(user.id)
  
      model_context.cleanup()
  
      # Can delete because owned resources have been removed
      with does_not_raise():
          zen_store.delete_user(user.id)
  
  
  def test_updating_user_with_existing_name_fails():
      """Tests updating a user with an existing account name fails."""
      zen_store = Client().zen_store
  
      with UserContext() as user:
          with UserContext() as existing_user:
              with pytest.raises(EntityExistsError):
                  zen_store.update_user(
                      user_id=user.id,
                      user_update=UserUpdate(name=existing_user.name),
                  )
  
          with ServiceAccountContext() as existing_service_account:
              with does_not_raise():
  