Unit tests for main window actions

dtool_lookup_gui/main.py

@@ -1,6 +1,5 @@
 #
-# Copyright 2021-2023 Johannes Laurin Hörmann
-#           2023 Ashwin Vazhappilly
+# Copyright 2021-2022 Johannes Laurin Hörmann
 #           2021 Lars Pastewka
 #
 # ### MIT license
@@ -34,7 +33,6 @@
 
 import dtoolcore
 import dtool_lookup_api.core.config
-
 from dtool_lookup_api.core.LookupClient import authenticate
 
 import gi
@@ -87,6 +85,13 @@ def __init__(self, *args, loop=None, **kwargs):
                          flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE, **kwargs)
         self.loop = loop
         self.args = None
+        self._progress_revealer = None
+        self._progress_popover = None
+        self._main_window = None
+
+    @property
+    def main_window(self):
+        return self._main_window
 
     def do_activate(self):
         logger.debug("do_activate")
@@ -105,6 +110,9 @@ def do_activate(self):
             logger.debug("Build GUI.")
 
             win = MainWindow(application=self)
+            self._main_window = win
+            logger.debug("New main window ID: %s", win.get_id())
+
             glob_pattern = os.path.join(os.path.dirname(__file__), os.pardir, 'data','icons','*','dtool_logo.xpm')
             icon_file_list = glob.glob(glob_pattern)
             if len(icon_file_list) > 0:
@@ -115,7 +123,7 @@ def do_activate(self):
             else:
                 logger.warning("Could not load app icons.")
             win.connect('destroy', lambda _: self.loop.stop())
-            self.loop.call_soon(win.refresh)  # Populate widgets after event loop starts
+            #self.loop.call_soon(win.refresh)  # Populate widgets after event loop starts
 
         logger.debug("Present main window.")
         win.present()
@@ -234,6 +242,11 @@ def do_startup(self):
         renew_token_action.connect("activate", self.do_renew_token)
         self.add_action(renew_token_action)
 
+        # copy-manager action
+        copy_action = Gio.SimpleAction.new("start-copy", GLib.VariantType.new("(ss)"))
+        copy_action.connect("activate", self.do_copy)
+        self.add_action(copy_action)
+
         Gtk.Application.do_startup(self)
 
     # custom application-scoped actions
@@ -338,6 +351,43 @@ async def retrieve_token(auth_url, username, password):
             username,
             password))
 
+    async def do_copy(self, action, value):
+        dataset, destination = value.unpack()
+
+        # Ensure that _progress_revealer and _progress_popover are initialized
+        if not self._progress_revealer or not self._progress_popover:
+            logger.error("Progress revealer or popover not initialized")
+            return
+
+        self._progress_revealer.set_reveal_child(True)
+        tracker = self._progress_popover.add_status_box(
+            self.progress_update, f'Copying dataset "{dataset}" to "{destination}"')
+
+        try:
+            await dataset.copy(destination, progressbar=tracker)
+        except ChildProcessError as exc:
+            logger.error(str(exc))
+            # Handle specific exceptions here
+        except Exception as e:
+            logger.error(f"Unexpected error: {e}")
+            # Handle other unexpected errors
+
+        tracker.set_done()
+        self.check_all_operations_done()
+
+    def progress_update(self):
+        if not self._progress_popover or not self._progress_chart:
+            logger.error("Progress popover or chart not initialized")
+            return
+
+        total_length = sum([len(tracker) for tracker in self._progress_popover.status_boxes])
+        total_step = sum([tracker.step for tracker in self._progress_popover.status_boxes])
+
+        if total_length > 0:
+            self._progress_chart.set_fraction(total_step / total_length)
+        else:
+            self._progress_chart.set_fraction(1.0)
+
 
 def run_gui():
     GObject.type_register(GtkSource.View)

dtool_lookup_gui/utils/copy_manager.py

@@ -36,11 +36,16 @@ class CopyManager:
 
     def __init__(self, progress_revealer, progress_popover):
         # Note: This is not particularly abstract, as it interacts directly with the Gtk widgets
+
+        # TODO: instead of a single progress_revealer and progress_popover assigned at instantiation,
+        # the copy manager would have an arbitrary amount of those that can be dynamically attached and detached
+
         self._progress_revealer = progress_revealer
         self._progress_chart = progress_revealer.get_child().get_child()
         self._progress_popover = progress_popover
 
     async def copy(self, dataset, destination):
+        # TODO: here the copy manager would loop over all attached status widgets
         self._progress_revealer.set_reveal_child(True)
         tracker = self._progress_popover.add_status_box(
             self.progress_update, f'Copying dataset »{dataset}« to »{destination}«')
@@ -61,6 +66,7 @@ async def copy(self, dataset, destination):
             self._progress_revealer.set_reveal_child(False)
 
     def progress_update(self):
+        # TODO: here the copy manager would loop over all attached status widgets
         # Refresh pie chart
         total_length = sum([len(tracker) for tracker in self._progress_popover.status_boxes])
         total_step = sum([tracker.step for tracker in self._progress_popover.status_boxes])

dtool_lookup_gui/views/main_window.py

@@ -24,6 +24,11 @@
 # SOFTWARE.
 #
 
+# TODO: any atomic operations that reflect core dtool behavior as documented
+# on https://dtoolcore.readthedocs.io/en/latest/api/dtoolcore.html#dtoolcore.DataSetCreator
+# or https://dtoolcore.readthedocs.io/en/latest/api/dtoolcore.html#dtoolcore.ProtoDataSet
+# should move to atomic actions on the main app level
+
 import asyncio
 import logging
 import os
@@ -178,8 +183,6 @@ class MainWindow(Gtk.ApplicationWindow):
     contents_per_page = Gtk.Template.Child()
     linting_errors_button = Gtk.Template.Child()
 
-
-
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
@@ -198,8 +201,7 @@ def __init__(self, *args, **kwargs):
 
         self.readme_buffer.connect("changed", self.on_readme_buffer_changed)
 
-        self.error_bar.set_revealed(False)
-        self.progress_revealer.set_reveal_child(False)
+
 
         # connect log handler to error bar
         root_logger = logging.getLogger()
@@ -218,6 +220,13 @@ def __init__(self, *args, **kwargs):
         self.server_versions_dialog = ServerVersionsDialog(application=self.application)
         self.error_linting_dialog = LintingErrorsDialog(application=self.application)
 
+        # Initialize the progress revealer and popover
+        self.progress_revealer = None  # Replace this with actual initialization
+        self.progress_popover = None  # Replace this with actual initialization
+
+        # Initialize the CopyManager
+        # self._copy_manager = CopyManager(self.progress_revealer, self.progress_popover)
+
         # window-scoped actions
 
         # search action
@@ -269,7 +278,14 @@ def __init__(self, *args, **kwargs):
 
         self.dependency_graph_widget.search_by_uuid = self._search_by_uuid
 
-        self._copy_manager = CopyManager(self.progress_revealer, self.progress_popover)
+        # TODO: is it possible to attach the CopyManager to the main app instead of this main window?
+        # One possibility would be to refactor the CopyManager in a way that it can have any
+        # number of arbitrary progress_revealer and progress_popover components attached.
+        # The copy manager lives at the Application scope.
+        # The main window here attaches its own progress_revealer and progress_popover to the
+        # CopyManager of the application instead of instantiating the copy manager here.
+        self.get_action_group("app").activate_action("start-copy-manager", None)
+
 
         _logger.debug(f"Constructed main window for app '{self.application.get_application_id()}'")
 
@@ -336,15 +352,23 @@ async def _refresh_base_uri_list_box(self):
     def _update_search_summary(self, datasets):
         row = self.base_uri_list_box.search_results_row
         total_size = sum([0 if dataset.size_int is None else dataset.size_int for dataset in datasets])
-        total_value = self.pagination['total']
+        if hasattr(self.pagination, 'total'):
+            total_value = self.pagination['total']
+        else:
+            _logger.warning("No pagination information available.")
+            total_value = len(datasets)
+
         row.info_label.set_text(f'{total_value} datasets, {sizeof_fmt(total_size).strip()}')
 
     def _update_main_statusbar(self):
-        total_number = self.pagination['total']
-        current_page = self.pagination['page']
-        last_page = self.pagination['last_page']
-        self.main_statusbar.push(0,
-                                 f"Total Number of Datasets: {total_number}, Current Page: {current_page} of {last_page}")
+        if hasattr(self.pagination, 'total') and hasattr(self.pagination, 'page') and hasattr(self.pagination, 'last_page'):
+            total_number = self.pagination['total']
+            current_page = self.pagination['page']
+            last_page = self.pagination['last_page']
+            self.main_statusbar.push(0,
+                                     f"Total Number of Datasets: {total_number}, Current Page: {current_page} of {last_page}")
+        else:
+            _logger.warning("No pagination information available to display in status bar.")
 
     def contents_per_page_changed(self, widget):
         self.contents_per_page_value = widget.get_active_text()
@@ -420,7 +444,6 @@ async def retry():
         # If a widget was passed in, re-enable it
         self.enable_pagination_buttons()
 
-
     def _search_by_uuid(self, uuid):
         search_text = dump_single_line_query_text({"uuid": uuid})
         self._search_by_search_text(search_text)
@@ -775,6 +798,7 @@ def on_add_items_clicked(self, widget):
             fpaths = dialog.get_filenames()
             for fpath in fpaths:
                 # uri = urllib.parse.unquote(uri, encoding='utf-8', errors='replace')
+                # TODO: _add_item should turn into an action on the app level as well
                 self._add_item(fpath)
         elif response == Gtk.ResponseType.CANCEL:
             pass
@@ -844,6 +868,7 @@ def on_linting_errors_button_clicked(self, widget):
         else:
             pass
 
+    # TODO: try to refactor into a do_freeze action in the main app as well
     @Gtk.Template.Callback()
     def on_freeze_clicked(self, widget):
         row = self.dataset_list_box.get_selected_row()
@@ -984,7 +1009,9 @@ def enable_pagination_buttons(self):
         self.page_advancer_button.set_sensitive(True)
         self.next_page_advancer_button.set_sensitive(True)
 
-    # @Gtk.Template.Callback(), not in .ui
+    # TODO: this should be an action do_copy
+    # if it is possible to hand to strings, e.g. source and destination to an action, then this action should
+    # go to the main app.
     def on_copy_clicked(self, widget):
         async def _copy():
             try:

test/conftest.py

@@ -1,20 +1,31 @@
+import json
 import logging
+import os
 import pytest
 
+from unittest.mock import patch
+
 import gi
 gi.require_version('Gtk', '3.0')
 gi.require_version('GtkSource', '4')
 from gi.repository import GLib, GObject, Gio, Gtk, GtkSource, GdkPixbuf
 
+import asyncio
 import gbulb
 gbulb.install(gtk=True)
 
+import dtool_lookup_api.core.LookupClient
+
 from dtool_lookup_gui.main import Application
+from dtool_lookup_gui.views.main_window import MainWindow
 
 
 logger = logging.getLogger(__name__)
 
 
+_HERE = os.path.dirname(os.path.abspath(__file__))
+# _ROOT = os.path.join(_HERE, "..")
+
 # ==========================================================================
 # fixtures from https://github.com/beeware/gbulb/blob/main/tests/conftest.py
 # ==========================================================================
@@ -76,6 +87,62 @@ def app(gtk_loop):
     app = Application(loop=gtk_loop)
     yield app
 
+@pytest.fixture(scope="function")
+def mock_token(scope="function"):
+    """Provide a mock authentication token"""
+    yield "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJmcmVzaCI6ZmFsc2UsImlhdCI6MTcwMTY4OTU5MywianRpIjoiYTdkN2Y5ZWItZGI3MS00YjExLWFhMzktZGQ2YzgzOTJmOWE4IiwidHlwZSI6ImFjY2VzcyIsInN1YiI6InRlc3R1c2VyIiwibmJmIjoxNzAxNjg5NTkzLCJleHAiOjE3MDE2OTMxOTN9.t9SQ00ecZRc-pspz-Du7321xfWIzgTTFKobkNed1CuQYHvtNrc3vdHYbqCWaYCqZpEVF8RlltldT4Lookx6vNgnW4olpiS2KTZ-X2asMhn7SShDtUJuU54CGeViWzYX_V_Pzckoe_cgjFkOutRvnwy_072Whnmc0TwYojwNqUScAIJRu0pzym84JngloXfdI7r25GcRVNtzsGUl7DDfrIz4aSOeVDAVEXhPjgEatKsvNdVZl1DIJsTZpuI7Jh7ZW1WsyjonqHR0J0kIVQn9imQyLyS9_CtmURBQ3kabx6cxhpx5LADrzLutSu24eA4FyECOdzjJ3SPGb9nIVTEDxQg"
+
+
+@pytest.fixture(scope="function")
+def mock_dataset_list():
+    """Provide a mock dataset list"""
+    with open(os.path.join(_HERE, 'data', 'mock_dataset_search_response.json'), 'r') as f:
+        dataset_list = json.load(f)
+
+    yield dataset_list
+
+
+@pytest.fixture(scope="function")
+def mock_config_info():
+    """Provide a mock server config info"""
+    with open(os.path.join(_HERE, 'data', 'mock_config_info_response.json'), 'r') as f:
+        config_info = json.load(f)
+
+    yield config_info
+
+
+@pytest.fixture(scope="function")
+def app_without_authentication(app, mock_token):
+    """Removes authentication from app."""
+    with patch("dtool_lookup_api.core.LookupClient.authenticate", return_value=mock_token):
+        yield app
+
+
+@pytest.fixture(scope="function")
+def app_with_mock_dtool_lookup_api_calls(
+        app_without_authentication, mock_dataset_list, mock_config_info):
+    """Replaces lookup api calls with mock methods that return fake lists of datasets."""
+
+    # TODO: figure out whether mocked methods work as they should
+    with (
+            patch("dtool_lookup_api.core.LookupClient.TokenBasedLookupClient.all", return_value=mock_dataset_list),
+            patch("dtool_lookup_api.core.LookupClient.TokenBasedLookupClient.search", return_value=mock_dataset_list),
+            patch("dtool_lookup_api.core.LookupClient.TokenBasedLookupClient.config", return_value=mock_config_info),
+            patch("dtool_lookup_api.core.LookupClient.ConfigurationBasedLookupClient.has_valid_token", return_value=True)
+        ):
+
+        yield app_without_authentication
+
+
+@pytest.fixture(scope="function")
+def populated_app(app_with_mock_dtool_lookup_api_calls):
+    """Provides app populated with mock datasets."""
+
+    # TODO: figure out how to populate and refresh app with data here before app launched
+    # app.main_window.activate_action('refresh-view')
+
+    yield app_with_mock_dtool_lookup_api_calls
+
 
 # ================================================================
 # fixtures related to https://github.com/pytest-dev/pytest-asyncio