From 06fa34d68828684d1b73766f97b7baaefea4e779 Mon Sep 17 00:00:00 2001
From: Bram Neijt <bram@neijt.nl>
Date: Fri, 10 Jun 2022 19:29:08 +0200
Subject: [PATCH] Happy flow

---
 core/dbt/task/run.py                          |  4 +-
 .../test_drop_dangling_models.py              | 53 +++++++++++--------
 2 files changed, 33 insertions(+), 24 deletions(-)

diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index c835977d334..4d8047f8ac6 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -512,7 +512,7 @@ def manage_schema(self, adapter, results: List[RunResult]):
             r.status in (NodeStatus.Error, NodeStatus.Fail, NodeStatus.Skipped) for r in results
         )
         if not was_successfull_complete_run and manage_schemas_config:
-            warn_or_error("One or more models failed, skipping schema management")
+            warn("One or more models failed, skipping schema management")
             return
 
         models_in_results: Set[Tuple[str, str, str]] = set(
@@ -526,6 +526,8 @@ def manage_schema(self, adapter, results: List[RunResult]):
                 (database, schema, relation.identifier): relation
                 for relation in adapter.list_relations(database, schema)
             }
+            if len(available_models) == 0:
+                warn_or_error(f"No modules in managed schema '{schema}' for database '{database}'")
             should_act_upon = available_models.keys() - models_in_results
             for (target_database, target_schema, target_identifier) in should_act_upon:
                 target_action = managed_schemas_actions_config[(target_database, target_schema)]
diff --git a/tests/functional/schema_management/test_drop_dangling_models.py b/tests/functional/schema_management/test_drop_dangling_models.py
index fe609057f51..576bdfb903f 100644
--- a/tests/functional/schema_management/test_drop_dangling_models.py
+++ b/tests/functional/schema_management/test_drop_dangling_models.py
@@ -21,50 +21,57 @@
 """
 
 
-@pytest.fixture(scope="class")
-def models():
-    return {
-        "numbers.sql": model,
-    }
 
+class TestDanglingModels:
 
-@pytest.fixture(scope="class")
-def dbt_profile_target():
-    return {
-        "type": "postgres",
-        "threads": 4,
-        "host": "localhost",
-        "port": int(os.getenv("POSTGRES_TEST_PORT", 5432)),
-        "user": os.getenv("POSTGRES_TEST_USER", "root"),
-        "pass": os.getenv("POSTGRES_TEST_PASS", "password"),
-        "dbname": os.getenv("POSTGRES_TEST_DATABASE", "dbt"),
-        "manage_schemas": True,
-    }
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_a.sql": model,
+            "model_b.sql": model,
+        }
 
 
-class TestDanglingModels:
     @pytest.fixture(scope="class")
-    def project_config_update(self):
+    def dbt_profile_target(self):
+        return {
+            "type": "postgres",
+            "threads": 4,
+            "host": "localhost",
+            "port": int(os.getenv("POSTGRES_TEST_PORT", 5432)),
+            "user": os.getenv("POSTGRES_TEST_USER", "root"),
+            "pass": os.getenv("POSTGRES_TEST_PASS", "password"),
+            "dbname": os.getenv("POSTGRES_TEST_DATABASE", "dbt"),
+            "manage_schemas": True,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self, unique_schema):
         return {
             "managed-schemas": [
                 {
                     "database": os.getenv("POSTGRES_TEST_DATABASE", "dbt"),
-                    "schema": "dbt",
+                    "schema": unique_schema,
                     "action": "drop",
                 }
             ]
         }
 
+
     def test_drop(
         self,
         project,
     ):
         # create numbers model
         run_dbt(["run"])
-        check_table_does_exist(project.adapter, "numbers")
+        check_table_does_exist(project.adapter, "model_a")
+        check_table_does_exist(project.adapter, "model_b")
         check_table_does_not_exist(project.adapter, "baz")
 
         # remove numbers model
-        project.update_models({})
+        project.update_models({
+            "model_b.sql": model,
+        })
         run_dbt(["run"])
-        check_table_does_not_exist(project.adapter, "numbers")
+        check_table_does_not_exist(project.adapter, "model_a")
+        check_table_does_exist(project.adapter, "model_b")