Commit e37e8026 authored by Samuel GAIST's avatar Samuel GAIST

[core] Add fork push test

parent 3183d9e7
......@@ -240,8 +240,16 @@ class AssetRemoteTest(AssetBaseTest):
def _modify_asset(self, asset_name):
"""Modify an asset"""
raise NotImplementedError
def _prepare_fork_dependencies(self, asset_name):
"""Prepare prefix content with fork dependencies"""
src_storage = self.storage_cls(prefix, asset_name)
dst_storage = self.storage_cls(tmp_prefix, asset_name)
dst_storage.save(*src_storage.load())
@slow
@skip_disconnected
def test_remote_list(self):
......@@ -321,6 +329,56 @@ class AssetRemoteTest(AssetBaseTest):
]
nose.tools.assert_equal(len(aoi_list), number_of_versions + 1)
@slow
@skip_disconnected
def test_push_different_forks(self):
with common.Selector(tmp_prefix) as selector:
if not selector.can_fork(self.asset_type):
raise nose.SkipTest("{} does not support fork".format(self.asset_type))
asset_name = self.object_map["fork"]
if "fork_from" in self.object_map:
original = self.object_map["fork_from"]
else:
original = self.object_map["pull"]
self._prepare_fork_dependencies(original)
number_of_forks = 1
if asset_name.count("/") != 4:
version_pos = asset_name.rindex("/") + 1
original_name = asset_name[: version_pos - 1]
else:
original_name = asset_name
for i in range(number_of_forks):
fork_name = "{}_{}".format(original_name, i)
if asset_name.count("/") != 4:
fork_name += "/1"
exit_code, outputs = self.call(
"fork", original, fork_name, prefix=tmp_prefix
)
nose.tools.eq_(exit_code, 0, outputs)
asset_name = fork_name
asset_name = "{}_{}".format(original_name, number_of_forks - 1)
if asset_name.count("/") != 4:
asset_name += "/1"
exit_code, output = self.call("push", asset_name, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, output)
config = MockConfig(self.live_server_url, user, token)
with common.make_webapi(config) as webapi:
asset_list = common.retrieve_remote_list(webapi, self.asset_type, ["name"])
aoi_list = [
asset for asset in asset_list if asset["name"].startswith(original_name)
]
nose.tools.assert_equal(len(aoi_list), number_of_forks)
@slow
@skip_disconnected
def test_push_and_delete(self):
......
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