code: def test_import_csv_enforced_schema(mock_event_logger):
      if utils.backend() == "sqlite":
          pytest.skip(
              "Sqlite doesn't support schema / database creation"
          )
  
      if utils.backend() == "mysql":
          pytest.skip("This test is flaky on MySQL")
  
      full_table_name = f"admin_database.{CSV_UPLOAD_TABLE_W_SCHEMA}"
  
      # Invalid table name
      resp = upload_csv(CSV_FILENAME1, full_table_name)
      assert "Table name cannot contain a schema" in resp
  
      # no schema specified, fail upload
      resp = upload_csv(
          CSV_FILENAME1,
          CSV_UPLOAD_TABLE_W_SCHEMA,
          extra={"schema": None},
      )
      assert (
          f"Database {escaped_double_quotes(CSV_UPLOAD_DATABASE)} schema"
          f" {escaped_double_quotes('None')} is not allowed for csv uploads"
          in resp
      )
  
      success_msg = f"CSV file {escaped_double_quotes(CSV_FILENAME1)} uploaded to table {escaped_double_quotes(full_table_name)}"
  
      resp = upload_csv(
          CSV_FILENAME1,
          CSV_UPLOAD_TABLE_W_SCHEMA,
          extra={"schema": "admin_database", "if_exists": "replace"},
      )
  
      assert success_msg in resp
      mock_event_logger.assert_called_with(
          action="successful_csv_upload",
          database=get_upload_db().name,
          schema="admin_database",
          table=CSV_UPLOAD_TABLE_W_SCHEMA,
      )
  
      with get_upload_db().get_sqla_engine_with_context() as engine:
          data = engine.execute(
              f"SELECT * from {ADMIN_SCHEMA_NAME}.{CSV_UPLOAD_TABLE_W_SCHEMA} ORDER BY b"
          ).fetchall()
          assert data == [("john", 1), ("paul", 2)]
  
      # user specified schema doesn't match, fail
      resp = upload_csv(
          CSV_FILENAME1,
          CSV_UPLOAD_TABLE_W_SCHEMA,
          extra={"schema": "gold"},
      )
      assert (
          f'Database {escaped_double_quotes(CSV_UPLOAD_DATABASE)} schema {escaped_double_quotes("gold")} is not allowed for csv uploads'
          in resp
      )
  
      # user specified schema matches the expected schema, append
      if utils.backend() == "hive":
          pytest.skip(
              "Hive database doesn't support append csv uploads."
          )
      resp = upload_csv(
          CSV_FILENAME1,
          CSV_UPLOAD_TABLE_W_SCHEMA,
          extra={"schema": "admin_database", "if_exists": "append"},
      )
      assert success_msg in resp
  
      # Clean up
      with get_upload_db().get_sqla_engine_with_context() as engine:
          engine.execute(f"DROP TABLE {full_table_name}")
  
imports: import json
  import logging
  import os
  import shutil
  from typing import Optional, Union
  from unittest import mock
  import pandas
  import pytest
  import superset.utils.database
  from superset.sql_parse import Table
  from tests.integration_tests.conftest import ADMIN_SCHEMA_NAME
  from superset import db
  from superset import security_manager
  from superset.models.core import Database
  from superset.utils import core
  from tests.integration_tests.test_app import app, login
  from tests.integration_tests.base_tests import get_resp, SupersetTestCase
label_desc: Use image node_id 14 for calling a global function with example usage: test_import_csv_enforced_schema(mock_event_logger) without return types