diff --git a/CHANGELOG.md b/CHANGELOG.md
index dc3a38152..f1d0df1d1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ### Fixes
 
 - Fix for issue where long-running python models led to invalid session errors ([544](https://github.com/databricks/dbt-databricks/pull/544))
+- Allow schema to be specified in testing (thanks @case-k-git!) ([538](https://github.com/databricks/dbt-databricks/pull/538))
 
 ## dbt-databricks 1.7.3 (Dec 12, 2023)
 
diff --git a/test.env.example b/test.env.example
index 7f94101e3..cd3aef4fc 100644
--- a/test.env.example
+++ b/test.env.example
@@ -7,6 +7,9 @@ DBT_DATABRICKS_HTTP_PATH=
 # Set the default database for the `databricks` profile
 DBT_DATABRICKS_UC_INITIAL_CATALOG=
 
+# Set the default schema for the `databricks` profile
+DBT_DATABRICKS_UC_INITIAL_SCHEMA=
+
 ## Also able to set the http-path values for each profile
 ## Server http-path value for `databricks_cluster` profile
 # DBT_DATABRICKS_CLUSTER_HTTP_PATH=
@@ -27,3 +30,7 @@ DBT_TEST_USER_3=
 
 # The default is INFO. Log levels can be: DEBUG, INFO, WARNING, ERROR, or CRITICAL
 # DBT_DATABRICKS_CONNECTOR_LOG_LEVEL=DEBUG
+
+
+# The default is INFO. Log levels can be: DEBUG, INFO, WARNING, ERROR, or CRITICAL
+# DBT_DATABRICKS_CONNECTOR_LOG_LEVEL=DEBUG
diff --git a/tests/integration/base.py b/tests/integration/base.py
index a6522a899..8d27fb5b8 100644
--- a/tests/integration/base.py
+++ b/tests/integration/base.py
@@ -166,12 +166,12 @@ def packages_config(self):
     def selectors_config(self):
         return None
 
-    def unique_schema(self):
-        schema = self.schema
-
-        to_return = "{}_{}".format(self.prefix, schema)
-
-        return to_return
+    # To test schema-related aspects, whether lower or upper, we can choose to use either the
+    # default schema or the test-defined schema.
+    def unique_schema(self, default_schema=None):
+        if not hasattr(self, "schema"):
+            self.schema = default_schema
+        return "{}_{}".format(self.prefix, self.schema)
 
     @property
     def default_database(self):
@@ -183,19 +183,23 @@ def alternative_database(self):
         return None
 
     def get_profile(self, adapter_type):
-        return {
+        profile = {
             "config": {"send_anonymous_usage_stats": False},
             "test": {
                 "outputs": {
                     "dev": dict(
                         **get_databricks_cluster_target(adapter_type),
-                        schema=self.unique_schema(),
                     ),
                 },
                 "target": "dev",
             },
         }
 
+        profile["test"]["outputs"]["dev"]["schema"] = self.unique_schema(
+            default_schema=profile["test"]["outputs"]["dev"]["schema"]
+        )
+        return profile
+
     def _pick_profile(self):
         test_name = self.id().split(".")[-1]
         return _profile_from_test_name(test_name)
diff --git a/tests/integration/incremental_on_schema_change/test_incremental_on_schema_change.py b/tests/integration/incremental_on_schema_change/test_incremental_on_schema_change.py
index b19e8e2db..323c9d908 100644
--- a/tests/integration/incremental_on_schema_change/test_incremental_on_schema_change.py
+++ b/tests/integration/incremental_on_schema_change/test_incremental_on_schema_change.py
@@ -2,10 +2,6 @@
 
 
 class TestIncrementalOnSchemaChange(DBTIntegrationTest):
-    @property
-    def schema(self):
-        return "incremental_on_schema_change"
-
     @property
     def models(self):
         return "models"
diff --git a/tests/profiles.py b/tests/profiles.py
index 73c660ff4..c00cb8985 100644
--- a/tests/profiles.py
+++ b/tests/profiles.py
@@ -16,6 +16,7 @@ def get_databricks_cluster_target(profile_type: str):
 def _build_databricks_cluster_target(
     http_path: str,
     catalog: Optional[str] = None,
+    schema: Optional[str] = None,
     session_properties: Optional[Dict[str, str]] = None,
 ):
     profile: Dict[str, Any] = {
@@ -32,6 +33,8 @@ def _build_databricks_cluster_target(
     }
     if catalog is not None:
         profile["catalog"] = catalog
+    if schema is not None:
+        profile["schema"] = schema
     if session_properties is not None:
         profile["session_properties"] = session_properties
     return profile
@@ -41,7 +44,8 @@ def databricks_cluster_target():
     return _build_databricks_cluster_target(
         http_path=os.getenv(
             "DBT_DATABRICKS_CLUSTER_HTTP_PATH", os.getenv("DBT_DATABRICKS_HTTP_PATH")
-        )
+        ),
+        schema=os.getenv("DBT_DATABRICKS_UC_INITIAL_SCHEMA", "default_schema"),
     )
 
 
@@ -51,6 +55,7 @@ def databricks_uc_cluster_target():
             "DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH", os.getenv("DBT_DATABRICKS_HTTP_PATH")
         ),
         catalog=os.getenv("DBT_DATABRICKS_UC_INITIAL_CATALOG", "main"),
+        schema=os.getenv("DBT_DATABRICKS_UC_INITIAL_SCHEMA", "default_schema"),
     )
 
 
@@ -61,4 +66,5 @@ def databricks_uc_sql_endpoint_target():
             os.getenv("DBT_DATABRICKS_HTTP_PATH"),
         ),
         catalog=os.getenv("DBT_DATABRICKS_UC_INITIAL_CATALOG", "main"),
+        schema=os.getenv("DBT_DATABRICKS_UC_INITIAL_SCHEMA", "default_schema"),
     )