Replace DB codenames by DB manager's database name registry

Data stores are no longer responsible of Database editor's tab
names. They merely register their names as candidates for
database display names but it is up to Toolbox what it chooses
to do with the information. This simplifies code and responsibilities
in spine_items.

spine_items/data_store/data_store.py

@@ -22,6 +22,8 @@
 from spinedb_api.helpers import remove_credentials_from_url, vacuum
 from spinetoolbox.helpers import create_dir
 from spinetoolbox.project_item.project_item import ProjectItem
+from spinetoolbox.spine_db_editor.editors import db_editor_registry
+from spinetoolbox.spine_db_editor.widgets.multi_spine_db_editor import open_db_editor
 from spinetoolbox.widgets.custom_qwidgets import SelectDatabaseItemsDialog
 from ..database_validation import DatabaseConnectionValidator
 from ..utils import convert_to_sqlalchemy_url, database_label
@@ -81,7 +83,7 @@ def get_db_map(self):
         sa_url = self.sql_alchemy_url()
         if sa_url is None:
             return None
-        return self._toolbox.db_mngr.get_db_map(sa_url, self._logger, codename=None)
+        return self._toolbox.db_mngr.get_db_map(sa_url, self._logger)
 
     @staticmethod
     def item_type():
@@ -104,7 +106,7 @@ def parse_url(self, url):
         if isinstance(url, dict):
             if url.get("dialect") == "sqlite" and (database := url.get("database")):
                 # Convert relative database path back to absolute
-                url["database"] = os.path.abspath(os.path.join(self._project.project_dir, database))
+                url["database"] = os.path.normcase(os.path.abspath(os.path.join(self._project.project_dir, database)))
             for key, value in url.items():
                 if value is not None:
                     base_url[key] = value
@@ -203,6 +205,8 @@ def do_update_url(self, **kwargs):
         was_valid = self._url_validated
         self._url_validated = False
         old_url = convert_to_sqlalchemy_url(self._url, self.name)
+        if old_url is not None and was_valid:
+            self._toolbox.db_mngr.name_registry.unregister(old_url, self.name)
         new_dialect = kwargs.get("dialect")
         if new_dialect == "sqlite":
             kwargs.update({"username": "", "password": "", "host": "", "port": "", "schema": ""})
@@ -271,7 +275,7 @@ def _clean_up_purge_dialog(self):
         self._purge_dialog = None
 
     def actions(self):
-        self._multi_db_editors_open = {x.name(): x for x in self._toolbox.db_mngr.get_all_multi_spine_db_editors()}
+        self._multi_db_editors_open = {x.name(): x for x in db_editor_registry.windows()}
         if not self._multi_db_editors_open:
             return super().actions()
         self._open_url_menu.clear()
@@ -311,8 +315,7 @@ def _open_spine_db_editor(self, reuse_existing):
             return
         sa_url = self.sql_alchemy_url()
         if sa_url is not None:
-            db_url_codenames = {sa_url: self.name}
-            self._toolbox.db_mngr.open_db_editor(db_url_codenames, reuse_existing)
+            open_db_editor([sa_url], self._toolbox.db_mngr, reuse_existing)
         self._check_notifications()
 
     def _open_url_in_existing_db_editor(self, db_editor):
@@ -321,7 +324,7 @@ def _open_url_in_existing_db_editor(self, db_editor):
             return
         sa_url = self.sql_alchemy_url()
         if sa_url is not None:
-            db_editor.add_new_tab({sa_url: self.name})
+            db_editor.add_new_tab([sa_url])
             db_editor.raise_()
             db_editor.activateWindow()
 
@@ -429,6 +432,7 @@ def _accept_url(self, url):
         self._update_actions_enabled()
         db_map = self.get_db_map()
         if db_map:
+            self._toolbox.db_mngr.name_registry.register(db_map.sa_url, self.name)
             clean = not self._toolbox.db_mngr.is_dirty(db_map)
             self._notify_about_dirtiness(clean)
 
@@ -491,19 +495,23 @@ def from_dict(name, item_dict, toolbox, project):
 
     def rename(self, new_name, rename_data_dir_message):
         """See base class."""
-        old_data_dir = os.path.abspath(self.data_dir)  # Old data_dir before rename
+        old_data_dir = os.path.abspath(self.data_dir)
         old_name = self.name
         if not super().rename(new_name, rename_data_dir_message):
             return False
+        if self._url_validated:
+            sa_url = self.sql_alchemy_url()
+            self._toolbox.db_mngr.name_registry.unregister(sa_url, old_name)
         # If dialect is sqlite and db line edit refers to a file in the old data_dir, db line edit needs updating
         if self._url["dialect"] == "sqlite":
             db_dir, db_filename = os.path.split(os.path.abspath(self._url["database"].strip()))
-            if db_dir == old_data_dir:
+            if db_dir == os.path.normcase(old_data_dir):
                 database = os.path.join(self.data_dir, db_filename)  # NOTE: data_dir has been updated at this point
                 # Check that the db was moved successfully to the new data_dir
                 if os.path.exists(database):
                     self._url.update(database=database)
                     self.load_url_into_selections(self._url)
+            self._check_notifications()
         return True
 
     def notify_destination(self, source_item):
@@ -543,6 +551,9 @@ def resources_for_direct_predecessors(self):
 
     def tear_down(self):
         """See base class"""
+        if self._url_validated:
+            sa_url = convert_to_sqlalchemy_url(self._url, self.name)
+            self._toolbox.db_mngr.name_registry.unregister(sa_url, self.name)
         self._toolbox.db_mngr.database_clean_changed.disconnect(self._set_database_clean)
         self._database_validator.wait_for_finish()
         super().tear_down()

spine_items/utils.py

@@ -84,7 +84,7 @@ def _convert_url(url):
         if dialect == "sqlite":
             database = url.get("database", "")
             if database:
-                url["database"] = os.path.abspath(database)
+                url["database"] = os.path.normcase(os.path.abspath(database))
             return URL("sqlite", **url)  # pylint: disable=unexpected-keyword-arg
         db_api = spinedb_api.SUPPORTED_DIALECTS.get(dialect)
         if db_api is None:

spine_items/view/view.py

@@ -20,6 +20,7 @@
 from spinetoolbox.helpers import busy_effect
 from spinetoolbox.plotting import PlottingError, plot_db_mngr_items
 from spinetoolbox.project_item.project_item import ProjectItem
+from spinetoolbox.spine_db_editor.widgets.multi_spine_db_editor import open_db_editor
 from spinetoolbox.spine_db_editor.widgets.spine_db_editor import SpineDBEditor
 from spinetoolbox.widgets.notification import Notification
 from .commands import PinOrUnpinDBValuesCommand
@@ -89,19 +90,19 @@ def save_selections(self):
     def open_editor(self, checked=False):
         """Opens selected db in the Spine database editor."""
         indexes = self._selected_reference_indexes()
-        db_url_codenames = self._db_url_codenames(indexes)
-        if not db_url_codenames:
+        db_urls = self._db_urls(indexes)
+        if not db_urls:
             return
-        self._toolbox.db_mngr.open_db_editor(db_url_codenames, reuse_existing_editor=True)
+        open_db_editor(db_urls, self._toolbox.db_mngr, reuse_existing_editor=True)
 
     @Slot(bool)
     def pin_values(self, checked=False):
         """Opens selected db in the Spine database editor to pin values."""
         indexes = self._selected_reference_indexes()
-        db_url_codenames = self._db_url_codenames(indexes)
-        if not db_url_codenames:
+        db_urls = self._db_urls(indexes)
+        if not db_urls:
             return
-        db_editor = SpineDBEditor(self._toolbox.db_mngr, db_url_codenames)
+        db_editor = SpineDBEditor(self._toolbox.db_mngr, db_urls)
         dialog = _PinValuesDialog(self, db_editor)
         db_editor.pinned_values_updated.connect(dialog.update_pinned_values)
         dialog.data_committed.connect(self._pin_db_values)
@@ -115,7 +116,7 @@ def selected_pinned_values(self):
         return self._properties_ui.treeView_pinned_values.selectedIndexes()
 
     def reference_resource_label_from_url(self, url):
-        for label, (ref_url, _) in self._references.items():
+        for label, ref_url in self._references.items():
             if str(ref_url) == str(url):
                 return label
         self._logger.msg_error.emit(f"<b>{self.name}</b>: Can't find any resource having url {url}.")
@@ -203,7 +204,7 @@ def add_to_plot(self, fetch_id, db_map, parameter_record):
     def failed_to_plot(self, fetch_id, db_map, conditions):
         self._logger.msg_warning.emit(
             f"<b>{self.name}</b>: "
-            f"Couldn't find any values having {_pk_to_ul(conditions)} in <b>{db_map.codename}</b>"
+            f"Couldn't find any values having {_pk_to_ul(conditions)} in <b>{self._toolbox.db_mngr.name_registry.display_name(db_map.sa_url)}</b>"
         )
         if not self._fetched_parameter_values:
             self._fetched_parameter_values = len(self._data_fetchers) * [None]
@@ -218,7 +219,7 @@ def _plot_if_all_fetched(self):
         if any(v != "skip" for v in self._fetched_parameter_values):
             plot_db_maps, plot_items = zip(*existing_fetched_parameter_values)
             try:
-                plot_widget = plot_db_mngr_items(plot_items, plot_db_maps)
+                plot_widget = plot_db_mngr_items(plot_items, plot_db_maps, self._toolbox.db_mngr.name_registry)
             except PlottingError as error:
                 self._logger.msg_error.emit(str(error))
                 return
@@ -279,12 +280,12 @@ def upstream_resources_updated(self, resources):
         for resource in resources:
             if resource.type_ == "database":
                 url = make_url(resource.url)
-                self._references[resource.label] = (url, resource.provider_name)
+                self._references[resource.label] = url
             elif resource.type_ == "file":
                 filepath = resource.path
                 if os.path.splitext(filepath)[1] == ".sqlite":
                     url = URL("sqlite", database=filepath)
-                    self._references[resource.label] = (url, resource.provider_name)
+                    self._references[resource.label] = url
         self.populate_reference_list()
 
     def replace_resources_from_upstream(self, old, new):
@@ -321,9 +322,9 @@ def _selected_pinned_value_indexes(self):
             self._properties_ui.treeView_pinned_values.selectAll()
         return self._properties_ui.treeView_pinned_values.selectionModel().selectedRows()
 
-    def _db_url_codenames(self, indexes):
-        """Returns a dict mapping url to provider's name for given indexes in the reference model."""
-        return dict(self._references[index.data(Qt.ItemDataRole.DisplayRole)] for index in indexes)
+    def _db_urls(self, indexes):
+        """Returns a list of database URLs for given indexes in the reference model."""
+        return [self._references[index.data(Qt.ItemDataRole.DisplayRole)] for index in indexes]
 
     def notify_destination(self, source_item):
         """See base class."""

tests/data_store/test_DataStoreExecutable.py

@@ -12,6 +12,7 @@
 
 """Unit tests for DataStoreExecutable."""
 from multiprocessing import Lock
+import os.path
 from pathlib import Path
 from tempfile import TemporaryDirectory
 import unittest
@@ -82,7 +83,7 @@ def test_output_resources_backward(self):
         self.assertEqual(len(resources), 1)
         resource = resources[0]
         self.assertEqual(resource.type_, "database")
-        self.assertEqual(resource.url, "sqlite:///" + str(db_file_path))
+        self.assertEqual(resource.url, "sqlite:///" + os.path.normcase(str(db_file_path)))
         self.assertEqual(resource.label, "db_url@name")
 
     def test_output_resources_forward(self):
@@ -95,7 +96,7 @@ def test_output_resources_forward(self):
         self.assertEqual(len(resources), 1)
         resource = resources[0]
         self.assertEqual(resource.type_, "database")
-        self.assertEqual(resource.url, "sqlite:///" + str(db_file_path))
+        self.assertEqual(resource.url, "sqlite:///" + os.path.normcase(str(db_file_path)))
         self.assertEqual(resource.label, "db_url@name")
 
 

tests/data_store/test_dataStore.py

@@ -27,7 +27,7 @@
 from spine_items.utils import convert_to_sqlalchemy_url, database_label
 from spinedb_api import create_new_spine_database
 from spinetoolbox.helpers import signal_waiter
-from ..mock_helpers import (
+from tests.mock_helpers import (
     create_mock_project,
     create_mock_toolbox,
     create_toolboxui_with_project,
@@ -104,21 +104,21 @@ def test_rename(self):
         # Check that DS is connected to an existing DS.sqlite file that is in data_dir
         url = self.ds_properties_ui.url_selector_widget.url_dict()
         self.assertEqual(url["dialect"], "sqlite")
-        self.assertEqual(url["database"], os.path.join(self.ds.data_dir, "temp_db.sqlite"))  # data_dir before rename
+        self.assertEqual(url["database"], os.path.normcase(os.path.join(self.ds.data_dir, "temp_db.sqlite")))
         self.assertTrue(os.path.exists(url["database"]))
         expected_name = "ABC"
         expected_short_name = "abc"
         expected_data_dir = os.path.join(self._project.items_dir, expected_short_name)
-        self.ds.rename(expected_name, "")  # Do rename
+        self.assertTrue(self.ds.rename(expected_name, ""))  # Do rename
         # Check name
         self.assertEqual(expected_name, self.ds.name)  # item name
         self.assertEqual(expected_name, self.ds.get_icon().name())  # name item on Design View
         # Check data_dir and logs_dir
         self.assertEqual(expected_data_dir, self.ds.data_dir)  # Check data dir
         # Check that the database path in properties has been updated
-        expected_db_path = os.path.join(expected_data_dir, "temp_db.sqlite")
+        expected_db_path = os.path.normcase(os.path.join(expected_data_dir, "temp_db.sqlite"))
         url = self.ds_properties_ui.url_selector_widget.url_dict()
-        self.assertEqual(url["database"], expected_db_path)
+        self.assertEqual(os.path.normcase(url["database"]), expected_db_path)
         # Check that the db file has actually been moved
         self.assertTrue(os.path.exists(url["database"]))
 
@@ -192,7 +192,7 @@ def tearDown(self):
 
     def create_temp_db(self):
         """Let's create a real db to more easily test complicated stuff (such as opening a tree view)."""
-        temp_db_path = os.path.join(self.ds.data_dir, "temp_db.sqlite")
+        temp_db_path = os.path.normcase(os.path.join(self.ds.data_dir, "temp_db.sqlite"))
         sqlite_url = "sqlite:///" + temp_db_path
         create_new_spine_database(sqlite_url)
         return temp_db_path
@@ -209,7 +209,7 @@ def test_item_dict(self):
                     self.assertTrue(url_key in d[k], f"Key '{url_key}' not in dict {d[k]}")
 
     def test_create_new_empty_spine_database(self):
-        """Test that a new Spine database is not created when clicking on 'New Spine db tool button'
+        """Test that an open file dialog is shown when clicking on 'New Spine db tool button'
         with an empty Data Store.
         """
         self.ds.activate()
@@ -218,7 +218,10 @@ def test_create_new_empty_spine_database(self):
         self.assertEqual(url["database"], "")
         # Click New Spine db button
         self.toolbox.db_mngr = mock.MagicMock()
-        self.ds_properties_ui.pushButton_create_new_spine_db.click()
+        with mock.patch("spine_items.data_store.data_store.QFileDialog") as file_dialog:
+            file_dialog.getSaveFileName.return_value = ("", "")
+            self.ds_properties_ui.pushButton_create_new_spine_db.click()
+            file_dialog.getSaveFileName.assert_called_once()
         url = self.ds_properties_ui.url_selector_widget.url_dict()
         self.assertEqual(url["dialect"], "sqlite")
         self.assertEqual(url["database"], "")
@@ -333,7 +336,7 @@ def test_copy_db_url_to_clipboard(self):
         self.ds_properties_ui.toolButton_copy_url.click()
         # noinspection PyArgumentList
         clipboard_text = QApplication.clipboard().text()
-        expected_url = "sqlite:///" + os.path.join(self.ds.data_dir, "temp_db.sqlite")
+        expected_url = "sqlite:///" + os.path.normcase(os.path.join(self.ds.data_dir, "temp_db.sqlite"))
         self.assertEqual(expected_url, clipboard_text.strip())
 
     def test_open_db_editor1(self):
@@ -345,14 +348,14 @@ def test_open_db_editor1(self):
         self.ds_properties_ui.url_selector_widget.set_url({"dialect": "sqlite", "database": temp_db_path})
         self.ds._update_url_from_properties()
         self.toolbox.db_mngr = mock.MagicMock()
-        self.toolbox.db_mngr.get_all_multi_spine_db_editors = lambda: iter([])
         while not self.ds.is_url_validated():
             QApplication.processEvents()
         self.assertTrue(self.ds_properties_ui.pushButton_ds_open_editor.isEnabled())
-        self.ds_properties_ui.pushButton_ds_open_editor.click()
-        sa_url = convert_to_sqlalchemy_url(self.ds.url(), "DS", logger=None)
-        self.assertIsNotNone(sa_url)
-        self.toolbox.db_mngr.open_db_editor.assert_called_with({sa_url: "DS"}, True)
+        with mock.patch("spine_items.data_store.data_store.open_db_editor") as open_db_editor:
+            self.ds_properties_ui.pushButton_ds_open_editor.click()
+            sa_url = convert_to_sqlalchemy_url(self.ds.url(), "DS", logger=None)
+            self.assertIsNotNone(sa_url)
+            open_db_editor.assert_called_with([sa_url], self.toolbox.db_mngr, True)
 
     def test_open_db_editor2(self):
         """Test that selecting the 'sqlite' dialect, typing the path to an existing db file,
@@ -363,14 +366,14 @@ def test_open_db_editor2(self):
         self.ds_properties_ui.url_selector_widget.set_url({"dialect": "sqlite", "database": temp_db_path})
         self.ds._update_url_from_properties()
         self.toolbox.db_mngr = mock.MagicMock()
-        self.toolbox.db_mngr.get_all_multi_spine_db_editors = lambda: iter([])
         while not self.ds.is_url_validated():
             QApplication.processEvents()
         self.assertTrue(self.ds_properties_ui.pushButton_ds_open_editor.isEnabled())
-        self.ds_properties_ui.pushButton_ds_open_editor.click()
-        sa_url = convert_to_sqlalchemy_url(self.ds.url(), "DS", logger=None)
-        self.assertIsNotNone(sa_url)
-        self.toolbox.db_mngr.open_db_editor.assert_called_with({sa_url: "DS"}, True)
+        with mock.patch("spine_items.data_store.data_store.open_db_editor") as open_db_editor:
+            self.ds_properties_ui.pushButton_ds_open_editor.click()
+            sa_url = convert_to_sqlalchemy_url(self.ds.url(), "DS", logger=None)
+            self.assertIsNotNone(sa_url)
+            open_db_editor.assert_called_with([sa_url], self.toolbox.db_mngr, True)
 
     def test_notify_destination(self):
         self.ds.logger.msg = mock.MagicMock()
@@ -400,14 +403,14 @@ def test_notify_destination(self):
         )
 
     def test_do_update_url_uses_filterable_resources_when_replacing_them(self):
-        database_1 = os.path.join(self._temp_dir.name, "db1.sqlite")
+        database_1 = os.path.normcase(os.path.join(self._temp_dir.name, "db1.sqlite"))
         Path(database_1).write_bytes(b"empty fake db")
         self.ds.do_update_url(dialect="sqlite", database=database_1)
         self.project.notify_resource_changes_to_predecessors.assert_called_once_with(self.ds)
         self.project.notify_resource_changes_to_successors.assert_called_once_with(self.ds)
         while not self.ds.is_url_validated():
             QApplication.processEvents()
-        database_2 = os.path.join(self._temp_dir.name, "db2.sqlite")
+        database_2 = os.path.normcase(os.path.join(self._temp_dir.name, "db2.sqlite"))
         Path(database_2).write_bytes(b"empty fake db")
         self.ds.do_update_url(dialect="sqlite", database=database_2)
         while not self.ds.is_url_validated():

tests/exporter/test_utils.py

@@ -28,7 +28,7 @@ def test_creates_url_resources_for_channels_with_urls(self):
             ),
         ]
         resources = output_database_resources(item_name, channels)
-        expected_url = "sqlite:///" + os.path.abspath("/path/to/db.sqlite")
+        expected_url = "sqlite:///" + os.path.normcase(os.path.abspath("/path/to/db.sqlite"))
         self.assertEqual(resources, [url_resource(item_name, expected_url, "out database")])