[ENH] CSV File Import: Add support for explicit workflow relative paths

Orange/widgets/data/tests/test_owcsvimport.py

@@ -1,32 +1,60 @@
-# pylint: disable=no-self-use,protected-access
+# pylint: disable=no-self-use,protected-access,invalid-name,arguments-differ
 import unittest
 from unittest import mock
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
 
 import os
 import io
 import csv
 import json
+from typing import Type, TypeVar, Optional
 
 import numpy as np
 from numpy.testing import assert_array_equal
 
-from AnyQt.QtCore import QSettings
+from AnyQt.QtCore import QSettings, Qt
+from AnyQt.QtGui import QIcon
+from AnyQt.QtWidgets import QFileDialog
+from AnyQt.QtTest import QSignalSpy
+
+from orangewidget.tests.utils import simulate
+from orangewidget.widget import OWBaseWidget
 
 from Orange.data import DiscreteVariable, TimeVariable, ContinuousVariable, \
  StringVariable
 from Orange.tests import named_file
 from Orange.widgets.tests.base import WidgetTest, GuiTest
 from Orange.widgets.data import owcsvimport
 from Orange.widgets.data.owcsvimport import (
- pandas_to_table, ColumnType, RowSpec
+ OWCSVFileImport, pandas_to_table, ColumnType, RowSpec,
 )
+from Orange.widgets.utils.pathutils import PathItem, samepath
 from Orange.widgets.utils.settings import QSettings_writeArray
 from Orange.widgets.utils.state_summary import format_summary_details
 
+W = TypeVar("W", bound=OWBaseWidget)
+
 
 class TestOWCSVFileImport(WidgetTest):
+ def create_widget(
+ self, cls: Type[W], stored_settings: Optional[dict] = None,
+ reset_default_settings=True, **kwargs) -> W:
+ if reset_default_settings:
+ self.reset_default_settings(cls)
+ widget = cls.__new__(cls, signal_manager=self.signal_manager,
+ stored_settings=stored_settings, **kwargs)
+ widget.__init__()
+
+ def delete():
+ widget.onDeleteWidget()
+ widget.close()
+ widget.deleteLater()
+
+ self._stack.callback(delete)
+ return widget
+
  def setUp(self):
+ super().setUp()
  self._stack = ExitStack().__enter__()
  # patch `_local_settings` to avoid side effects, across tests
  fname = self._stack.enter_context(named_file(""))
@@ -37,10 +65,9 @@ def setUp(self):
  self.widget = self.create_widget(owcsvimport.OWCSVFileImport)
 
  def tearDown(self):
- self.widgets.remove(self.widget)
- self.widget.onDeleteWidget()
- self.widget = None
+ del self.widget
  self._stack.close()
+ super().tearDown()
 
  def test_basic(self):
  w = self.widget
@@ -58,6 +85,8 @@ def test_basic(self):
  (range(1, 3), RowSpec.Skipped),
  ],
  )
+ data_regions_path = os.path.join(
+ os.path.dirname(__file__), "data-regions.tab")
 
  def _check_data_regions(self, table):
  self.assertEqual(len(table), 3)
@@ -82,7 +111,7 @@ def test_restore(self):
  }
  )
  item = w.current_item()
- self.assertEqual(item.path(), path)
+ self.assertTrue(samepath(item.path(), path))
  self.assertEqual(item.options(), self.data_regions_options)
  out = self.get_output("Data", w)
  self._check_data_regions(out)
@@ -102,12 +131,19 @@ def test_restore_from_local(self):
  owcsvimport.OWCSVFileImport,
  )
  item = w.current_item()
- self.assertEqual(item.path(), path)
+ self.assertIsNone(item)
+ simulate.combobox_activate_index(w.recent_combo, 0)
+ item = w.current_item()
+ self.assertTrue(samepath(item.path(), path))
  self.assertEqual(item.options(), self.data_regions_options)
+ data = w.settingsHandler.pack_data(w)
  self.assertEqual(
- w._session_items, [(path, self.data_regions_options.as_dict())],
- "local settings item must be recorded in _session_items when "
- "activated in __init__",
+ data['_session_items_v2'], [
+ (PathItem.AbsPath(path).as_dict(),
+ self.data_regions_options.as_dict())
+ ],
+ "local settings item must be recorded in _session_items_v2 when "
+ "activated",
  )
  self._check_data_regions(self.get_output("Data", w))
 
@@ -189,6 +225,134 @@ def test_backward_compatibility(self):
  self.assertIsInstance(domain["numeric2"], ContinuousVariable)
  self.assertIsInstance(domain["string"], StringVariable)
 
+ @staticmethod
+ @contextmanager
+ def _browse_setup(widget: OWCSVFileImport, path: str):
+ browse_dialog = widget._browse_dialog
+ with mock.patch.object(widget, "_browse_dialog") as r:
+ dlg = browse_dialog()
+ dlg.setOption(QFileDialog.DontUseNativeDialog)
+ dlg.selectFile(path)
+ dlg.exec_ = dlg.exec = lambda: QFileDialog.Accepted
+ r.return_value = dlg
+ with mock.patch.object(owcsvimport.CSVImportDialog, "exec_",
+ lambda _: QFileDialog.Accepted):
+ yield
+
+ def test_browse(self):
+ widget = self.widget
+ path = self.data_regions_path
+ with self._browse_setup(widget, path):
+ widget.browse()
+ cur = widget.current_item()
+ self.assertIsNotNone(cur)
+ self.assertTrue(samepath(cur.path(), path))
+
+ def test_browse_prefix(self):
+ widget = self.widget
+ path = self.data_regions_path
+ with self._browse_setup(widget, path):
+ basedir = os.path.dirname(__file__)
+ widget.workflowEnv = lambda: {"basedir": basedir}
+ widget.workflowEnvChanged("basedir", basedir, "")
+ widget.browse_relative(prefixname="basedir")
+
+ cur = widget.current_item()
+ self.assertIsNotNone(cur)
+ self.assertTrue(samepath(cur.path(), path))
+ self.assertIsInstance(cur.varPath(), PathItem.VarPath)
+
+ def test_browse_prefix_parent(self):
+ widget = self.widget
+ path = self.data_regions_path
+
+ with self._browse_setup(widget, path):
+ basedir = os.path.join(os.path.dirname(__file__), "bs")
+ widget.workflowEnv = lambda: {"basedir": basedir}
+ widget.workflowEnvChanged("basedir", basedir, "")
+ mb = widget._path_must_be_relative_mb = mock.Mock()
+ widget.browse_relative(prefixname="basedir")
+ mb.assert_called()
+ self.assertIsNone(widget.current_item())
+
+ def test_browse_for_missing(self):
+ missing = os.path.dirname(__file__) + "/this file does not exist.csv"
+ widget = self.create_widget(
+ owcsvimport.OWCSVFileImport, stored_settings={
+ "_session_items": [
+ (missing, self.data_regions_options.as_dict())
+ ]
+ }
+ )
+ widget.activate_recent(0)
+ dlg = widget.findChild(QFileDialog)
+ assert dlg is not None
+ # calling selectFile when using native (macOS) dialog does not have
+ # an effect - at least not immediately;
+ dlg.setOption(QFileDialog.DontUseNativeDialog)
+ dlg.selectFile(self.data_regions_path)
+ dlg.accept()
+ cur = widget.current_item()
+ self.assertTrue(samepath(self.data_regions_path, cur.path()))
+ self.assertEqual(
+ self.data_regions_options.as_dict(), cur.options().as_dict()
+ )
+
+ def test_browse_for_missing_prefixed(self):
+ path = self.data_regions_path
+ basedir = os.path.dirname(path)
+ widget = self.create_widget(
+ owcsvimport.OWCSVFileImport, stored_settings={
+ "__version__": 3,
+ "_session_items_v2": [
+ (PathItem.VarPath("basedir", "this file does not exist.csv").as_dict(),
+ self.data_regions_options.as_dict())]
+ },
+ env={"basedir": basedir}
+ )
+ widget.activate_recent(0)
+ dlg = widget.findChild(QFileDialog)
+ assert dlg is not None
+ # calling selectFile when using native (macOS) dialog does not have
+ # an effect - at least not immediately;
+ dlg.setOption(QFileDialog.DontUseNativeDialog)
+ dlg.selectFile(path)
+ dlg.accept()
+ cur = widget.current_item()
+ self.assertTrue(samepath(path, cur.path()))
+ self.assertEqual(
+ cur.varPath(), PathItem.VarPath("basedir", "data-regions.tab"))
+ self.assertEqual(
+ self.data_regions_options.as_dict(), cur.options().as_dict()
+ )
+
+ def test_browse_for_missing_prefixed_parent(self):
+ path = self.data_regions_path
+ basedir = os.path.join(os.path.dirname(path), "origin1")
+ item = (PathItem.VarPath("basedir",
+ "this file does not exist.csv"),
+ self.data_regions_options)
+ widget = self.create_widget(
+ owcsvimport.OWCSVFileImport, stored_settings={
+ "__version__": 3,
+ "_session_items_v2": [(item[0].as_dict(), item[1].as_dict())]
+ },
+ env={"basedir": basedir}
+ )
+ mb = widget._path_must_be_relative_mb = mock.Mock()
+ widget.activate_recent(0)
+ dlg = widget.findChild(QFileDialog)
+ assert dlg is not None
+ # calling selectFile when using native (macOS) dialog does not have
+ # an effect - at least not immediately;
+ dlg.setOption(QFileDialog.DontUseNativeDialog)
+ dlg.selectFile(path)
+ dlg.accept()
+ mb.assert_called()
+ cur = widget.current_item()
+ self.assertEqual(item[0], cur.varPath())
+ self.assertEqual(item[1].as_dict(), cur.options().as_dict())
+
 
 class TestImportDialog(GuiTest):
  @staticmethod
@@ -219,6 +383,42 @@ def test_dialog():
  opts1 = d.options()
 
 
+class TestModel(GuiTest):
+ def test_model(self):
+ path = TestOWCSVFileImport.data_regions_path
+ model = owcsvimport.VarPathItemModel()
+ model.setItemPrototype(owcsvimport.ImportItem())
+ it1 = owcsvimport.ImportItem()
+ it1.setVarPath(PathItem.VarPath("prefix", "data-regions.tab"))
+ it2 = owcsvimport.ImportItem()
+ it2.setVarPath(PathItem.AbsPath(path))
+ model.appendRow([it1])
+ model.appendRow([it2])
+
+ def data(row, role):
+ return model.data(model.index(row, 0), role)
+
+ self.assertIsInstance(data(0, Qt.DecorationRole), QIcon)
+ self.assertIsInstance(data(1, Qt.DecorationRole), QIcon)
+
+ self.assertEqual(data(0, Qt.DisplayRole), "data-regions.tab")
+ self.assertEqual(data(1, Qt.DisplayRole), "data-regions.tab")
+
+ self.assertEqual(data(0, Qt.ToolTipRole), "${prefix}/data-regions.tab (missing)")
+ self.assertTrue(samepath(data(1, Qt.ToolTipRole), path))
+
+ self.assertIsNotNone(data(0, Qt.ForegroundRole))
+ self.assertIsNone(data(1, Qt.ForegroundRole))
+ spy = QSignalSpy(model.dataChanged)
+ model.setReplacementEnv({"prefix": os.path.dirname(path)})
+ self.assertSequenceEqual(
+ [[model.index(0, 0), model.index(1, 0), []]],
+ list(spy)
+ )
+ self.assertEqual(data(0, Qt.ToolTipRole), "${prefix}/data-regions.tab")
+ self.assertIsNone(data(0, Qt.ForegroundRole))
+
+
 class TestUtils(unittest.TestCase):
  def test_load_csv(self):
  contents = (
@@ -347,6 +547,14 @@ def test_open_compressed(self):
  with owcsvimport._open(fname, "rt", encoding="ascii") as f:
  self.assertEqual(content, f.read())
 
+ def test_sniff_csv(self):
+ f = io.StringIO("A|B|C\n1|2|3\n1|2|3")
+ dialect, header = owcsvimport.sniff_csv(f)
+ self.assertEqual(dialect.delimiter, "|")
+ self.assertTrue(header)
+ with self.assertRaises(csv.Error):
+ owcsvimport.sniff_csv(f, delimiters=["."])
+
 
 def _open_write(path, mode, encoding=None):
  # pylint: disable=import-outside-toplevel

Orange/widgets/utils/__init__.py

@@ -1,7 +1,10 @@
+import enum
 import inspect
 import sys
 from collections import deque
-from typing import TypeVar, Callable, Any, Iterable, Optional, Hashable
+from typing import (
+ TypeVar, Callable, Any, Iterable, Optional, Hashable, Type, Union
+)
 
 from AnyQt.QtCore import QObject
 
@@ -81,7 +84,13 @@ def mypredicate(x):
  return inspect.getmembers(obj, mypredicate)
 
 
-_T1 = TypeVar("_T1")
+def qname(type_: type) -> str:
+ """Return the fully qualified name for a `type_`."""
+ return "{0.__module__}.{0.__qualname__}".format(type_)
+
+
+_T1 = TypeVar("_T1") # pylint: disable=invalid-name
+_E = TypeVar("_E", bound=enum.Enum) # pylint: disable=invalid-name
 
 
 def apply_all(seq, op):
@@ -116,3 +125,14 @@ def unique_everseen(iterable, key=None):
  if el_k not in seen:
  seen.add(el_k)
  yield el
+
+
+def enum_get(etype: Type[_E], name: str, default: _T1) -> Union[_E, _T1]:
+ """
+ Return an Enum member by `name`. If no such member exists in `etype`
+ return `default`.
+ """
+ try:
+ return etype[name]
+ except LookupError:
+ return default