Skip to content

GitLab

Commit 88cdb453 authored by Samuel GAIST's avatar Samuel GAIST
Browse files

[backend] Moved AssetType to asset.py 

Also fix the name of AssetModel test file and
improve the storage property implementation.
parent 60413c85
Hide whitespace changes
Inline Side-by-side
beat/editor/backend/asset.py
View file @ 88cdb453
......@@ -23,7 +23,155 @@
# #
###############################################################################
from .assetmodel import AssetType
import importlib
import os
import beat.core
from enum import Enum, unique
from beat.backend.python import utils
from beat.core.schema import validate
from beat.cmdline import common
@unique
class AssetType(Enum):
"""All possible assets available on the BEAT platform"""
UNKNOWN = ("unknown", None)
ALGORITHM = ("algorithms", beat.core.algorithm.Algorithm)
DATABASE = ("databases", beat.core.database.Database)
DATAFORMAT = ("dataformats", beat.core.dataformat.DataFormat)
EXPERIMENT = ("experiments", beat.core.experiment.Experiment)
LIBRARY = ("libraries", beat.core.library.Library)
PLOTTER = ("plotters", beat.core.plotter.Plotter)
PLOTTERPARAMETER = (
"plotterparameters",
beat.core.plotterparameter.Plotterparameter,
)
PROTOCOLTEMPLATE = (
"protocoltemplates",
beat.core.protocoltemplate.ProtocolTemplate,
)
TOOLCHAIN = ("toolchains", beat.core.toolchain.Toolchain)
def __init__(self, path, klass):
self.path = path
self.klass = klass
@property
def storage(self):
mod = importlib.import_module(self.klass.__module__)
return getattr(mod, "Storage")
@staticmethod
def from_path(path):
for asset_type in AssetType:
if asset_type.path == path:
return asset_type
raise RuntimeError("Unknown asset path {}".format(path))
def can_create(self):
"""Returns whether a new asset can be created from scratch"""
return self not in [self.UNKNOWN, self.EXPERIMENT]
def can_fork(self):
"""Returns whether a new asset can be forked"""
return self not in [self.UNKNOWN, self.DATABASE, self.PROTOCOLTEMPLATE]
def split_count(self):
"""Returns the number of "/" that should be part of its name"""
if self == self.UNKNOWN:
return 0
elif self == self.EXPERIMENT:
return 5
elif self not in [self.DATABASE, self.PROTOCOLTEMPLATE]:
return 2
else:
return 1
def validate(self, data):
"""Runs the schema validation and returns whether an asset is valid
:param data str: asset content
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to validate unknown type")
return validate(self.name.lower(), data)
def create_new(self, prefix, name):
"""Create a new asset from a prototype
:param prefix str: Path to the prefix
:param name str: name of the asset
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to create an asset of unknown type")
success = common.create(prefix, self.name.lower(), [name])
return success == 0
def create_new_version(self, prefix, name):
"""Create a new version of the asset
:param prefix str: Path to the prefix
:param name str: name of the asset
"""
if self == self.UNKNOWN:
raise RuntimeError(
"Trying to create a new version of an asset of unknown type"
)
success = common.new_version(prefix, self.name.lower(), name)
return success == 0
def fork(self, prefix, source, destination):
"""Fork an asset
:param prefix str: Path to the prefix
:param source str: name of the original asset
:param destination str: name of the new asset
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to fork an asset of unknown type")
success = common.fork(prefix, self.name.lower(), source, destination)
return success == 0
def delete(self, prefix, name):
"""Delete an asset
:param prefix str: Path to the prefix
:param name str: name of the asset to delete
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to delete an asset of unknown type")
success = common.delete_local(prefix, self.name.lower(), [name])
return success == 0
def declaration_path(self, prefix, name):
"""Returns the full path to the declaration file
:param prefix str: Path to the prefix
:param name str: name of the asset to delete
"""
asset_folder = os.path.join(prefix, self.path)
return os.path.join(asset_folder, "{}.json".format(name))
def has_code(self):
"""Returns whether this asset type contains code"""
if self.klass is None:
return False
return issubclass(self.storage, utils.CodeStorage)
class Asset:
......
beat/editor/backend/assetmodel.py
View file @ 88cdb453
......@@ -25,159 +25,14 @@
import os
import beat.core
from enum import Enum, unique
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtCore import pyqtSlot
from PyQt5.QtCore import pyqtProperty
from PyQt5.QtCore import QStringListModel
from beat.backend.python import utils
from beat.core.schema import validate
from beat.cmdline import common
from ..utils import dataformat_basetypes
@unique
class AssetType(Enum):
"""All possible assets available on the BEAT platform"""
UNKNOWN = ("unknown", None)
ALGORITHM = ("algorithms", beat.core.algorithm.Algorithm)
DATABASE = ("databases", beat.core.database.Database)
DATAFORMAT = ("dataformats", beat.core.dataformat.DataFormat)
EXPERIMENT = ("experiments", beat.core.experiment.Experiment)
LIBRARY = ("libraries", beat.core.library.Library)
PLOTTER = ("plotters", beat.core.plotter.Plotter)
PLOTTERPARAMETER = (
"plotterparameters",
beat.core.plotterparameter.Plotterparameter,
)
PROTOCOLTEMPLATE = (
"protocoltemplates",
beat.core.protocoltemplate.ProtocolTemplate,
)
TOOLCHAIN = ("toolchains", beat.core.toolchain.Toolchain)
def __init__(self, path, klass):
self.path = path
self.klass = klass
@property
def storage(self):
mod = __import__(self.klass.__module__, fromlist=["Storage"])
return getattr(mod, "Storage")
@staticmethod
def from_path(path):
for asset_type in AssetType:
if asset_type.path == path:
return asset_type
raise RuntimeError("Unknown asset path {}".format(path))
def can_create(self):
"""Returns whether a new asset can be created from scratch"""
return self not in [self.UNKNOWN, self.EXPERIMENT]
def can_fork(self):
"""Returns whether a new asset can be forked"""
return self not in [self.UNKNOWN, self.DATABASE, self.PROTOCOLTEMPLATE]
def split_count(self):
"""Returns the number of "/" that should be part of its name"""
if self == self.UNKNOWN:
return 0
elif self == self.EXPERIMENT:
return 5
elif self not in [self.DATABASE, self.PROTOCOLTEMPLATE]:
return 2
else:
return 1
def validate(self, data):
"""Runs the schema validation and returns whether an asset is valid
:param data str: asset content
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to validate unknown type")
return validate(self.name.lower(), data)
def create_new(self, prefix, name):
"""Create a new asset from a prototype
:param prefix str: Path to the prefix
:param name str: name of the asset
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to create an asset of unknown type")
success = common.create(prefix, self.name.lower(), [name])
return success == 0
def create_new_version(self, prefix, name):
"""Create a new version of the asset
:param prefix str: Path to the prefix
:param name str: name of the asset
"""
if self == self.UNKNOWN:
raise RuntimeError(
"Trying to create a new version of an asset of unknown type"
)
success = common.new_version(prefix, self.name.lower(), name)
return success == 0
def fork(self, prefix, source, destination):
"""Fork an asset
:param prefix str: Path to the prefix
:param source str: name of the original asset
:param destination str: name of the new asset
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to fork an asset of unknown type")
success = common.fork(prefix, self.name.lower(), source, destination)
return success == 0
def delete(self, prefix, name):
"""Delete an asset
:param prefix str: Path to the prefix
:param name str: name of the asset to delete
"""
if self == self.UNKNOWN:
raise RuntimeError("Trying to delete an asset of unknown type")
success = common.delete_local(prefix, self.name.lower(), [name])
return success == 0
def declaration_path(self, prefix, name):
"""Returns the full path to the declaration file
:param prefix str: Path to the prefix
:param name str: name of the asset to delete
"""
asset_folder = os.path.join(prefix, self.path)
return os.path.join(asset_folder, "{}.json".format(name))
def has_code(self):
"""Returns whether this asset type contains code"""
if self.klass is None:
return False
return issubclass(self.storage, utils.CodeStorage)
from .asset import AssetType
class AssetModel(QStringListModel):
......
beat/editor/test/test_asset.py
View file @ 88cdb453
......@@ -24,13 +24,131 @@
###############################################################################
import os
import tempfile
import pytest
import shutil
import tempfile
from ..backend.asset import Asset
from ..backend.assetmodel import AssetType
@pytest.fixture(params=[item for item in AssetType if item.can_create()])
def creatable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if not item.can_create()])
def uncreatable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if item.can_fork()])
def forkable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if not item.can_fork()])
def unforkable_asset_type(request):
return request.param
class TestAssetType:
"""Test the asset type enum"""
asset_name_map = {
AssetType.ALGORITHM: "user/foo/1",
AssetType.DATABASE: "foo/1",
AssetType.DATAFORMAT: "user/foo/1",
AssetType.EXPERIMENT: "user/foo/bar/1/baz",
AssetType.LIBRARY: "user/foo/1",
AssetType.PLOTTER: "user/foo/1",
AssetType.PLOTTERPARAMETER: "user/foo/1",
AssetType.PROTOCOLTEMPLATE: "foo/1",
AssetType.TOOLCHAIN: "user/foo/1",
AssetType.UNKNOWN: None,
}
existing_asset_name_map = {
AssetType.ALGORITHM: "autonomous/add/1",
AssetType.DATABASE: "simple/1",
AssetType.DATAFORMAT: "user/complexes/1",
AssetType.EXPERIMENT: "user/user/double/1/double",
AssetType.LIBRARY: "user/sum/1",
AssetType.PLOTTER: "user/scatter/1",
AssetType.PLOTTERPARAMETER: "plot/config/1",
AssetType.PROTOCOLTEMPLATE: "double/1",
AssetType.TOOLCHAIN: "user/double/1",
AssetType.UNKNOWN: None,
}
def test_unknown_validation(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.validate({})
def test_unknown_creation(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.create_new("foo", "bar")
def test_unknown_new_version(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.create_new_version("foo", "bar")
def test_unknown_fork(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.fork("foo", "bar", "baz")
def test_unknown_delete(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.delete("foo", "bar")
def test_creation(self, test_prefix, creatable_asset_type):
asset_name = self.asset_name_map[creatable_asset_type]
result = creatable_asset_type.create_new(test_prefix, asset_name)
assert result
def test_creation_failure(self, test_prefix, uncreatable_asset_type):
asset_name = self.asset_name_map[uncreatable_asset_type]
with pytest.raises(RuntimeError):
uncreatable_asset_type.create_new(test_prefix, asset_name)
def test_create_new_version(self, test_prefix, creatable_asset_type):
asset_name = self.existing_asset_name_map[creatable_asset_type]
result = creatable_asset_type.create_new_version(test_prefix, asset_name)
assert result
def test_create_new_version_failure(self, test_prefix, uncreatable_asset_type):
asset_name = self.existing_asset_name_map[uncreatable_asset_type]
with pytest.raises(RuntimeError):
uncreatable_asset_type.create_new_version(test_prefix, asset_name)
def test_fork(self, test_prefix, forkable_asset_type):
source_asset_name = self.existing_asset_name_map[forkable_asset_type]
dest_asset_name = self.asset_name_map[forkable_asset_type]
result = forkable_asset_type.fork(
test_prefix, source_asset_name, dest_asset_name
)
assert result
def test_fork_failure(self, test_prefix, unforkable_asset_type):
source_asset_name = self.existing_asset_name_map[unforkable_asset_type]
dest_asset_name = self.asset_name_map[unforkable_asset_type]
with pytest.raises(RuntimeError):
unforkable_asset_type.fork(test_prefix, source_asset_name, dest_asset_name)
def test_delete(self, test_prefix, asset_type):
with tempfile.TemporaryDirectory(suffix=".prefix") as prefix:
tmp_prefix = os.path.join(prefix, "prefix")
shutil.copytree(test_prefix, tmp_prefix)
asset_name = self.existing_asset_name_map[asset_type]
result = asset_type.delete(tmp_prefix, asset_name)
assert result
def test_declaration_path(self, test_prefix, asset_type):
asset_name = self.existing_asset_name_map[asset_type]
declaration_path = asset_type.declaration_path(test_prefix, asset_name)
assert os.path.exists(declaration_path)
class TestAsset:
"""Test that the asset works correctly"""
......
beat/editor/test/test_models.py → beat/editor/test/test_assetmodel.py
View file @ 88cdb453
......@@ -23,35 +23,13 @@
# #
###############################################################################
import os
import pytest
import tempfile
import shutil
from ..backend.assetmodel import AssetModel, AssetType
from ..backend.assetmodel import AssetModel
from ..backend.asset import AssetType
from ..utils import dataformat_basetypes
@pytest.fixture(params=[item for item in AssetType if item.can_create()])
def creatable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if not item.can_create()])
def uncreatable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if item.can_fork()])
def forkable_asset_type(request):
return request.param
@pytest.fixture(params=[item for item in AssetType if not item.can_fork()])
def unforkable_asset_type(request):
return request.param
def create_model(qtbot, test_prefix, asset_type):
model = AssetModel()
......@@ -118,100 +96,3 @@ class TestAssetModel:
else:
assert len(new_asset_list) > len(asset_list)
assert new_asset_list != asset_list
class TestAssetType:
"""Test the asset type enum"""
asset_name_map = {
AssetType.ALGORITHM: "user/foo/1",
AssetType.DATABASE: "foo/1",
AssetType.DATAFORMAT: "user/foo/1",
AssetType.EXPERIMENT: "user/foo/bar/1/baz",
AssetType.LIBRARY: "user/foo/1",
AssetType.PLOTTER: "user/foo/1",
AssetType.PLOTTERPARAMETER: "user/foo/1",
AssetType.PROTOCOLTEMPLATE: "foo/1",
AssetType.TOOLCHAIN: "user/foo/1",
AssetType.UNKNOWN: None,
}
existing_asset_name_map = {
AssetType.ALGORITHM: "autonomous/add/1",
AssetType.DATABASE: "simple/1",
AssetType.DATAFORMAT: "user/complexes/1",
AssetType.EXPERIMENT: "user/user/double/1/double",
AssetType.LIBRARY: "user/sum/1",
AssetType.PLOTTER: "user/scatter/1",
AssetType.PLOTTERPARAMETER: "plot/config/1",
AssetType.PROTOCOLTEMPLATE: "double/1",
AssetType.TOOLCHAIN: "user/double/1",
AssetType.UNKNOWN: None,
}
def test_unknown_validation(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.validate({})
def test_unknown_creation(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.create_new("foo", "bar")
def test_unknown_new_version(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.create_new_version("foo", "bar")
def test_unknown_fork(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.fork("foo", "bar", "baz")
def test_unknown_delete(self):
with pytest.raises(RuntimeError):
AssetType.UNKNOWN.delete("foo", "bar")
def test_creation(self, test_prefix, creatable_asset_type):
asset_name = self.asset_name_map[creatable_asset_type]
result = creatable_asset_type.create_new(test_prefix, asset_name)
assert result
def test_creation_failure(self, test_prefix, uncreatable_asset_type):
asset_name = self.asset_name_map[uncreatable_asset_type]
with pytest.raises(RuntimeError):
uncreatable_asset_type.create_new(test_prefix, asset_name)
def test_create_new_version(self, test_prefix, creatable_asset_type):
asset_name = self.existing_asset_name_map[creatable_asset_type]
result = creatable_asset_type.create_new_version(test_prefix, asset_name)
assert result
def test_create_new_version_failure(self, test_prefix, uncreatable_asset_type):
asset_name = self.existing_asset_name_map[uncreatable_asset_type]
with pytest.raises(RuntimeError):
uncreatable_asset_type.create_new_version(test_prefix, asset_name)
def test_fork(self, test_prefix, forkable_asset_type):
source_asset_name = self.existing_asset_name_map[forkable_asset_type]
dest_asset_name = self.asset_name_map[forkable_asset_type]
result = forkable_asset_type.fork(
test_prefix, source_asset_name, dest_asset_name
)
assert result
def test_fork_failure(self, test_prefix, unforkable_asset_type):
source_asset_name = self.existing_asset_name_map[unforkable_asset_type]
dest_asset_name = self.asset_name_map[unforkable_asset_type]
with pytest.raises(RuntimeError):
unforkable_asset_type.fork(test_prefix, source_asset_name, dest_asset_name)
def test_delete(self, test_prefix, asset_type):
with tempfile.TemporaryDirectory(suffix=".prefix") as prefix:
tmp_prefix = os.path.join(prefix, "prefix")
shutil.copytree(test_prefix, tmp_prefix)
asset_name = self.existing_asset_name_map[asset_type]
result = asset_type.delete(tmp_prefix, asset_name)
assert result
def test_declaration_path(self, test_prefix, asset_type):
asset_name = self.existing_asset_name_map[asset_type]
declaration_path = asset_type.declaration_path(test_prefix, asset_name)
assert os.path.exists(declaration_path)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment