fixes remaining tests

dlt/pipeline/pipeline.py

@@ -1002,10 +1002,11 @@ def _managed_state(self, *, extract_state: bool = False) -> Iterator[TPipelineSt
             # restore original pipeline props
             self._state_to_props(backup_state)
             # synchronize schema storage with initial list of schemas, note that we'll not be able to synchronize the schema content
-            # TODO: we should restore schemas backup here
-            for existing_schema_name in self._schema_storage.list_schemas():
-                if existing_schema_name not in self.schema_names:
-                    self._schema_storage.remove_schema(existing_schema_name)
+            if self._schema_storage:
+                # TODO: we should restore schemas backup here
+                for existing_schema_name in self._schema_storage.list_schemas():
+                    if existing_schema_name not in self.schema_names:
+                        self._schema_storage.remove_schema(existing_schema_name)
             # raise original exception
             raise
         else:

tests/pipeline/test_pipeline.py

@@ -212,9 +212,9 @@ def i_fail():
 
     # nothing to normalize
     assert len(storage.list_files_to_normalize_sorted()) == 0
-    # but the schemas are stored in the pipeline
-    p.schemas["default_3"]
-    p.schemas["default_4"]
+    # pipeline state is successfully rollbacked after the last extract and default_3 and 4 schemas are not present
+    assert set(p.schema_names) == {"default", "default_2"}
+    assert set(p._schema_storage.list_schemas()) == {"default", "default_2"}
 
 
 def test_restore_state_on_dummy() -> None: