[backend][assetmodel] Refactor AssetModel

Make a base class in order to keep our current
AssetModel as is and add one specialised for
DataFormat which can contain:
- only the DataFormat created
- the DataFormat created as well as the basetypes
- a subset of both

The last one is usually done through a QSortFilterProxyModel

beat/editor/backend/assetmodel.py

@@ -30,15 +30,107 @@ from PyQt5.QtCore import pyqtProperty
 from PyQt5.QtCore import pyqtSignal
 from PyQt5.QtCore import pyqtSlot
 
+from ..decorators import frozen
 from ..utils import dataformat_basetypes
 from .asset import Asset
 from .asset import AssetType
 
 
-class AssetModel(QStringListModel):
-    """The asset model present a list of available asset from a given type"""
+def enumerate_assets(prefix_path, asset_type, latest_only=False):
+    """Enumerate the assets available in the given prefix for the given type
+    """
+
+    def _find_json_files(path):
+        """Return all json files from folder sorted"""
+
+        asset_items = os.scandir(path)
+        json_files = sorted(
+            [
+                item.name
+                for item in asset_items
+                if item.is_file() and item.name.endswith("json")
+            ]
+        )
+        return json_files
+
+    assets_list = []
+    asset_folder = os.path.join(prefix_path, asset_type.path)
+
+    if asset_type in [AssetType.DATABASE, AssetType.PROTOCOLTEMPLATE]:
+        # These assets have no user associated with them
+        asset_folders = [entry for entry in os.scandir(asset_folder) if entry.is_dir()]
+        for asset_folder in asset_folders:
+            json_files = _find_json_files(asset_folder)
+            if json_files:
+                if latest_only:
+                    json_files = json_files[-1:]
+
+                for json_file in json_files:
+                    assets_list.append(
+                        "{name}/{version}".format(
+                            name=asset_folder.name, version=json_file.split(".")[0]
+                        )
+                    )
+    else:
+        # Assets belonging to a user
+        asset_users = [entry for entry in os.scandir(asset_folder) if entry.is_dir()]
+
+        for asset_user in asset_users:
+            asset_folders = [
+                entry for entry in os.scandir(asset_user) if entry.is_dir()
+            ]
+            for asset_folder in asset_folders:
+                if asset_type == AssetType.EXPERIMENT:
+                    for root, dirs, files in os.walk(asset_folder, topdown=False):
+                        if dirs:
+                            continue
+                        anchor = "experiments/"
+                        position = root.index(anchor) + len(anchor)
+                        experiment_path = root[position:]
+                        for json_file in [
+                            file for file in files if file.endswith("json")
+                        ]:
+                            assets_list.append(
+                                "{experiment_path}/{name}".format(
+                                    experiment_path=experiment_path,
+                                    name=json_file.split(".")[0],
+                                )
+                            )
+                else:
+                    json_files = _find_json_files(asset_folder)
+                    if json_files:
+                        if latest_only:
+                            json_files = json_files[-1:]
+
+                        for json_file in json_files:
+                            assets_list.append(
+                                "{user}/{name}/{version}".format(
+                                    user=asset_user.name,
+                                    name=asset_folder.name,
+                                    version=json_file.split(".")[0],
+                                )
+                            )
+
+    return sorted(assets_list)
+
+
+def json_path(prefix_path, asset_type, asset_name):
+    """Returns the full path to the json file matching the asset given
+
+    :param prefix_path str: path of the prefix
+    :param asset_type AssetType: asset type to search in
+    :param asset_name str: fully qualified asset name
+    """
+
+    asset = Asset(prefix_path, asset_type, asset_name)
+    if not os.path.exists(asset.declaration_path):
+        raise ValueError("Invalid asset {}".format(asset_name))
+    return asset.declaration_path
+
+
+class AbstractAssetModel(QStringListModel):
+    """Base class for asset related models"""
 
-    assetTypeChanged = pyqtSignal(AssetType)
     prefixPathChanged = pyqtSignal(str)
 
     def __init__(self, parent=None):
@@ -46,102 +138,87 @@ class AssetModel(QStringListModel):
 
         super().__init__(parent)
 
-        self.__latest_only = True
-        self.__prefix_path = None
-        self.__asset_type = AssetType.UNKNOWN
+        self._latest_only = True
+        self._prefix_path = None
+
+    @property
+    def asset_folder(self):
+        """Returns the folder matching this model asset type"""
+
+        raise NotImplementedError
+
+    def lastestOnlyEnabled(self):
+        return self._latest_only
 
     def setLatestOnlyEnabled(self, enabled):
-        if self.__latest_only == enabled:
+        if self._latest_only == enabled:
             return
-        self.__latest_only = enabled
+
+        self._latest_only = enabled
         self.reload()
 
+    latest_only_enabled = pyqtProperty(
+        str, fget=lastestOnlyEnabled, fset=setLatestOnlyEnabled
+    )
+
+    def prefixPath(self):
+        """Returns the prefix path use by this model
+
+        :return: the prefix path used
+        """
+
+        return self._prefix_path
+
+    def setPrefixPath(self, path):
+        """Set the prefix path use by this model
+
+        :param str path: Path to prefix
+        """
+
+        if self._prefix_path == path:
+            return
+
+        self._prefix_path = path
+        self.reload()
+        self.prefixPathChanged.emit(path)
+
+    prefix_path = pyqtProperty(
+        str, fget=prefixPath, fset=setPrefixPath, notify=prefixPathChanged
+    )
+
+    def json_path(self, asset_name):
+        """Returns the full path to the json file matching the asset given
+
+        :param asset_name str: fully qualified asset name
+        """
+
+        raise NotImplementedError
+
     @pyqtSlot()
     def reload(self):
         """Loads the content regarding the asset property from the prefix"""
 
-        if not self.__prefix_path or self.__asset_type == AssetType.UNKNOWN:
-            return
+        raise NotImplementedError
 
-        def _find_json_files(path):
-            """Return all json files from folder sorted"""
 
-            asset_items = os.scandir(path)
-            json_files = sorted(
-                [
-                    item.name
-                    for item in asset_items
-                    if item.is_file() and item.name.endswith("json")
-                ]
-            )
-            return json_files
+@frozen
+class AssetModel(AbstractAssetModel):
+    """The asset model present a list of available asset from a given type"""
 
-        assets_list = []
+    assetTypeChanged = pyqtSignal(AssetType)
 
-        if self.asset_type in [AssetType.DATABASE, AssetType.PROTOCOLTEMPLATE]:
-            # These assets have no user associated with them
-            asset_folders = [
-                entry for entry in os.scandir(self.asset_folder) if entry.is_dir()
-            ]
-            for asset_folder in asset_folders:
-                json_files = _find_json_files(asset_folder)
-                if json_files:
-                    if self.__latest_only:
-                        json_files = json_files[-1:]
-
-                    for json_file in json_files:
-                        assets_list.append(
-                            "{name}/{version}".format(
-                                name=asset_folder.name, version=json_file.split(".")[0]
-                            )
-                        )
-        else:
-            # Assets belonging to a user
-            asset_users = [
-                entry for entry in os.scandir(self.asset_folder) if entry.is_dir()
-            ]
+    def __init__(self, parent=None):
+        """Constructor"""
 
-            for asset_user in asset_users:
-                asset_folders = [
-                    entry for entry in os.scandir(asset_user) if entry.is_dir()
-                ]
-                for asset_folder in asset_folders:
-                    if self.asset_type == AssetType.EXPERIMENT:
-                        for root, dirs, files in os.walk(asset_folder, topdown=False):
-                            if dirs:
-                                continue
-                            anchor = "experiments/"
-                            position = root.index(anchor) + len(anchor)
-                            experiment_path = root[position:]
-                            for json_file in [
-                                file for file in files if file.endswith("json")
-                            ]:
-                                assets_list.append(
-                                    "{experiment_path}/{name}".format(
-                                        experiment_path=experiment_path,
-                                        name=json_file.split(".")[0],
-                                    )
-                                )
-                    else:
-                        json_files = _find_json_files(asset_folder)
-                        if json_files:
-                            if self.__latest_only:
-                                json_files = json_files[-1:]
-
-                            for json_file in json_files:
-                                assets_list.append(
-                                    "{user}/{name}/{version}".format(
-                                        user=asset_user.name,
-                                        name=asset_folder.name,
-                                        version=json_file.split(".")[0],
-                                    )
-                                )
+        super().__init__(parent)
+
+        self._asset_type = AssetType.UNKNOWN
 
-        assets_list = sorted(assets_list)
-        if self.asset_type == AssetType.DATAFORMAT:
-            assets_list = dataformat_basetypes() + assets_list
+    @property
+    def asset_folder(self):
+        """Reimpl"""
 
-        self.setStringList(assets_list)
+        return os.path.join(self.prefix_path, self.asset_type.path)
 
     def assetType(self):
         """Returns the asset type of this model
@@ -149,7 +226,7 @@ class AssetModel(QStringListModel):
         :return: Asset type of this model
         """
 
-        return self.__asset_type
+        return self._asset_type
 
     def setAssetType(self, type_):
         """Set the asset type of this model
@@ -157,10 +234,10 @@ class AssetModel(QStringListModel):
         :param AssetType type_: Asset type this model should show
         """
 
-        if self.__asset_type == type_:
+        if self._asset_type == type_:
             return
 
-        self.__asset_type = type_
+        self._asset_type = type_
         self.reload()
         self.assetTypeChanged.emit(type_)
 
@@ -168,44 +245,79 @@ class AssetModel(QStringListModel):
         AssetType, fget=assetType, fset=setAssetType, notify=assetTypeChanged
     )
 
-    def prefixPath(self):
-        """Returns the prefix path use by this model
+    def json_path(self, asset_name):
+        """Reimpl"""
 
-        :return: the prefix path used
+        return json_path(self.prefix_path, self.asset_type, asset_name)
+
+    @pyqtSlot()
+    def reload(self):
+        """Reimpl"""
+
+        if not self.prefix_path or self._asset_type == AssetType.UNKNOWN:
+            return
+
+        self.setStringList(
+            enumerate_assets(
+                self.prefix_path, self.asset_type, self.latest_only_enabled
+            )
+        )
+
+
+@frozen
+class DataFormatModel(AbstractAssetModel):
+    """DataFormat specific asset model"""
+
+    def __init__(self, parent=None):
+        super().__init__(parent)
+
+        self._full_list_enabled = False
+        self.__asset_type = AssetType.DATAFORMAT
+
+    @property
+    def asset_folder(self):
+        """Reimpl"""
+
+        return os.path.join(self.prefix_path, self.__asset_type.path)
+
+    def fullListEnabled(self):
+        """Returns whether the full list mode is enabled.
+
+        :return: if full list is enabled
         """
 
-        return self.__prefix_path
+        return self._full_list_enabled
 
-    def setPrefixPath(self, path):
-        """Set the prefix path use by this model
+    def setFullListEnabled(self, enabled):
+        """Enable/disable the full list mode
 
-        :param str path: Path to prefix
+        :param bool enabled: enable the full list mode
         """
 
-        if self.__prefix_path == path:
+        if self._full_list_enabled == enabled:
             return
 
-        self.__prefix_path = path
+        self._full_list_enabled = enabled
         self.reload()
-        self.prefixPathChanged.emit(path)
 
-    prefix_path = pyqtProperty(
-        str, fget=prefixPath, fset=setPrefixPath, notify=prefixPathChanged
-    )
+    full_list_enabled = pyqtProperty(str, fget=fullListEnabled, fset=setFullListEnabled)
 
-    @property
-    def asset_folder(self):
-        """Returns the folder matching this model asset type"""
+    def json_path(self, asset_name):
+        """Reimpl"""
 
-        return os.path.join(self.prefix_path, self.asset_type.path)
+        return json_path(self.prefix_path, self.__asset_type, asset_name)
 
-    def json_path(self, asset_name):
-        """Returns the full path to the json file matching the asset given
+    @pyqtSlot()
+    def reload(self):
+        """Reimpl"""
 
-        :param asset_name str: fully qualified asset name
-        """
+        if not self.prefix_path:
+            return
+
+        assets = enumerate_assets(
+            self.prefix_path, self.__asset_type, self.latest_only_enabled
+        )
+        if self._full_list_enabled:
+            assets = dataformat_basetypes() + assets
 
-        asset = Asset(self.prefix_path, self.asset_type, asset_name)
-        if not os.path.exists(asset.declaration_path):
-            raise ValueError("Invalid asset {}".format(asset_name))
-        return asset.declaration_path
+        self.setStringList(assets)

beat/editor/test/test_assetmodel.py

@@ -29,6 +29,7 @@ import pytest
 
 from ..backend.asset import AssetType
 from ..backend.assetmodel import AssetModel
+from ..backend.assetmodel import DataFormatModel
 from ..utils import dataformat_basetypes
 
 
@@ -55,19 +56,14 @@ class TestAssetModel:
         assert len(asset_list) > 0
         asset_type = asset_model.asset_type
 
-        if asset_type == AssetType.DATAFORMAT:
-            basetypes = dataformat_basetypes()
-            for item in asset_list:
-                assert len(item.split("/")) == 3 or item in basetypes
+        if asset_type == AssetType.EXPERIMENT:
+            split_size = 5
+        elif asset_type in [AssetType.DATABASE, AssetType.PROTOCOLTEMPLATE]:
+            split_size = 2
         else:
-            if asset_type == AssetType.EXPERIMENT:
-                split_size = 5
-            elif asset_type in [AssetType.DATABASE, AssetType.PROTOCOLTEMPLATE]:
-                split_size = 2
-            else:
-                split_size = 3
-            for item in asset_list:
-                assert len(item.split("/")) == split_size
+            split_size = 3
+        for item in asset_list:
+            assert len(item.split("/")) == split_size
 
     def test_json_path(self, asset_model):
         asset_list = asset_model.stringList()
@@ -150,3 +146,91 @@ class TestAssetModel:
         asset_model.reload()
 
         assert asset_model.stringList() == asset_list
+
+
+class TestDataFormatModel:
+    """Test that the mock editor works correctly"""
+
+    @pytest.fixture(params=[False, True], ids=["Normal list", "Include base types"])
+    def df_model(self, request, test_prefix):
+        df_model = DataFormatModel()
+        df_model.prefix_path = test_prefix
+        df_model.full_list_enabled = request.param
+        return df_model
+
+    def test_model_load(self, df_model):
+        asset_list = df_model.stringList()
+        assert len(asset_list) > 0
+
+        basetypes = dataformat_basetypes()
+        for item in asset_list:
+            assert len(item.split("/")) == 3 or item in basetypes
+
+    def test_json_path(self, df_model):
+        asset_list = df_model.stringList()
+        assert len(asset_list) > 0
+        base_types = dataformat_basetypes()
+        for item in asset_list:
+
+            if item in base_types:
+                with pytest.raises(RuntimeError):
+                    _ = df_model.json_path(item)
+            else:
+                path = df_model.json_path(item)
+                assert path
+
+    def test_json_path_invalid_asset(self, df_model):
+        with pytest.raises(ValueError):
+            path = "invalid/invalid_name/1"
+            df_model.json_path(path)
+
+    def test_latest_only(self, df_model):
+        asset_list = df_model.stringList()
+        assert len(asset_list) > 0
+        df_model.setLatestOnlyEnabled(False)
+        new_asset_list = df_model.stringList()
+
+        assert len(new_asset_list) > len(asset_list)
+        assert new_asset_list != asset_list
+
+    def test_latest_only_no_reload(self, df_model):
+        asset_list = df_model.stringList()
+        assert len(asset_list) > 0
+        df_model.setLatestOnlyEnabled(True)
+        new_asset_list = df_model.stringList()
+        assert asset_list == new_asset_list
+
+    def test_setting_same_path_no_reload(self, qtbot, df_model):
+        asset_list = df_model.stringList()
+        assert len(asset_list) > 0
+        with qtbot.assertNotEmitted(df_model.prefixPathChanged):
+            df_model.setPrefixPath(df_model.prefix_path)
+        new_asset_list = df_model.stringList()
+        assert asset_list == new_asset_list
+
+    def test_unexpected_files(self, df_model):
+        """This test ensures that unexpected files in the prefix don't break
+           the AssetModel class loading code.
+        """
+
+        def __create_file(path):
+            with open(os.path.join(path, "unexpected.txt"), "wt") as unexpected_file:
+                unexpected_file.write("nothing")
+
+        asset_list = df_model.stringList()
+
+        items = os.scandir(df_model.asset_folder)
+        folders = []
+        for item in items:
+            path = os.path.join(df_model.asset_folder, item)
+            if os.path.isdir(path):
+                folders.append(path)
+
+        for folder in folders:
+            __create_file(os.path.join(df_model.asset_folder, folder))
+
+        __create_file(df_model.asset_folder)
+
+        df_model.reload()
+
+        assert df_model.stringList() == asset_list