tests: simple update example
This commit is contained in:
@@ -272,6 +272,19 @@ class bam_session:
|
|||||||
with open(os.path.join(session_rootdir, ".bam_paths_uuid.json")) as f:
|
with open(os.path.join(session_rootdir, ".bam_paths_uuid.json")) as f:
|
||||||
return json.load(f)
|
return json.load(f)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def is_dirty(session_rootdir):
|
||||||
|
paths_add = {}
|
||||||
|
paths_modified = {}
|
||||||
|
paths_remove = {}
|
||||||
|
|
||||||
|
bam_session.status(
|
||||||
|
session_rootdir,
|
||||||
|
paths_add, paths_remove, paths_modified
|
||||||
|
)
|
||||||
|
|
||||||
|
return any((paths_add, paths_modified, paths_remove))
|
||||||
|
|
||||||
|
|
||||||
class bam_commands:
|
class bam_commands:
|
||||||
"""
|
"""
|
||||||
@@ -439,34 +452,49 @@ class bam_commands:
|
|||||||
|
|
||||||
# TODO(cam) multiple paths
|
# TODO(cam) multiple paths
|
||||||
session_rootdir = bam_config.find_sessiondir(paths[0], abort=True)
|
session_rootdir = bam_config.find_sessiondir(paths[0], abort=True)
|
||||||
|
# so as to avoid off-by-one errors string mangling
|
||||||
|
session_rootdir = session_rootdir.rstrip(os.sep)
|
||||||
|
|
||||||
paths_uuid = bam_session.load_paths_uuid(session_rootdir)
|
paths_uuid = bam_session.load_paths_uuid(session_rootdir)
|
||||||
|
|
||||||
if not paths_uuid:
|
if not paths_uuid:
|
||||||
print("Nothing to update!")
|
print("Nothing to update!")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
if bam_session.is_dirty(session_rootdir):
|
||||||
|
fatal("Local changes detected, commit before checking out!")
|
||||||
|
|
||||||
# -------------------------------------------------------------------------------
|
# -------------------------------------------------------------------------------
|
||||||
# TODO(cam) don't guess this important info
|
# TODO(cam) don't guess this important info
|
||||||
files = os.listdir(session_rootdir)
|
files = [f for f in os.listdir(session_rootdir) if not f.startswith(".")]
|
||||||
files_blend = [f for f in files if f.endswith(".blend")]
|
files_blend = [f for f in files if f.endswith(".blend")]
|
||||||
if files_blend:
|
if files_blend:
|
||||||
file = files_blend[0]
|
f = files_blend[0]
|
||||||
else:
|
else:
|
||||||
file = files[0]
|
f = files[0]
|
||||||
with open(os.path.join(session_rootdir, ".bam_paths_remap.json")) as f:
|
with open(os.path.join(session_rootdir, ".bam_paths_remap.json")) as fp:
|
||||||
paths_remap = json.load(f)
|
paths_remap = json.load(fp)
|
||||||
paths_remap_relbase = paths_remap.get(".", "")
|
paths_remap_relbase = paths_remap.get(".", "")
|
||||||
path = os.path.join(paths_remap_relbase, file)
|
path = os.path.join(paths_remap_relbase, f)
|
||||||
# -------------------------------------------------------------------------------
|
# -------------------------------------------------------------------------------
|
||||||
|
|
||||||
# Send to sever sha PUT
|
|
||||||
# retrieve zip GET
|
|
||||||
# merge sessions
|
# merge sessions
|
||||||
|
session_tmp = session_rootdir + ".tmp"
|
||||||
bam_commands.checkout(
|
bam_commands.checkout(
|
||||||
path,
|
path,
|
||||||
output_dir=session_rootdir.rstrip(os.sep) + ".tmp",
|
output_dir=session_tmp,
|
||||||
session_rootdir_partial=session_rootdir,
|
session_rootdir_partial=session_rootdir,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
for dirpath, dirnames, filenames in os.walk(session_tmp):
|
||||||
|
for filename in filenames:
|
||||||
|
filepath = os.path.join(dirpath, filename)
|
||||||
|
f_src = filepath
|
||||||
|
f_dst = session_rootdir + filepath[len(session_tmp):]
|
||||||
|
os.rename(f_src, f_dst)
|
||||||
|
import shutil
|
||||||
|
shutil.rmtree(session_tmp)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def commit(paths, message):
|
def commit(paths, message):
|
||||||
import requests
|
import requests
|
||||||
|
@@ -841,9 +841,41 @@ class BamCheckoutTest(BamSessionTestCase):
|
|||||||
self.assertEqual("", stderr)
|
self.assertEqual("", stderr)
|
||||||
self.assertEqual("Nothing to update!\n", stdout)
|
self.assertEqual("Nothing to update!\n", stdout)
|
||||||
|
|
||||||
#stdout, stderr = bam_run(["checkout"], session_path)
|
def test_update_simple(self):
|
||||||
|
session_name = "mysession"
|
||||||
|
file_name = "other_file.txt"
|
||||||
|
file_data = b"initial data!\n"
|
||||||
|
file_data_append = b"appended data!\n"
|
||||||
|
|
||||||
|
proj_path, session_path = self.init_session(session_name)
|
||||||
|
|
||||||
|
# now do a real commit
|
||||||
|
file_quick_write(session_path, file_name, file_data)
|
||||||
|
stdout, stderr = bam_run(["commit", "-m", "test message"], session_path)
|
||||||
|
self.assertEqual("", stderr)
|
||||||
|
|
||||||
|
# remove the path
|
||||||
|
shutil.rmtree(session_path)
|
||||||
|
|
||||||
|
# checkout the file again
|
||||||
|
session_path_a = session_path + "_a"
|
||||||
|
stdout, stderr = bam_run(["checkout", file_name, "--output", session_path_a], proj_path)
|
||||||
|
self.assertEqual("", stderr)
|
||||||
|
|
||||||
|
session_path_b = session_path + "_b"
|
||||||
|
stdout, stderr = bam_run(["checkout", file_name, "--output", session_path_b], proj_path)
|
||||||
|
self.assertEqual("", stderr)
|
||||||
|
|
||||||
|
file_quick_write(session_path_a, file_name, file_data_append, append=True)
|
||||||
|
|
||||||
|
stdout, stderr = bam_run(["commit", "-m", "commit appended data"], session_path_a)
|
||||||
|
self.assertEqual("", stderr)
|
||||||
|
|
||||||
|
stdout, stderr = bam_run(["update"], session_path_b)
|
||||||
|
self.assertEqual("", stderr)
|
||||||
|
|
||||||
|
with open(os.path.join(session_path_b, file_name), 'rb') as f:
|
||||||
|
self.assertEqual(f.read(), file_data + file_data_append)
|
||||||
|
|
||||||
|
|
||||||
class BamBlendTest(BamSimpleTestCase):
|
class BamBlendTest(BamSimpleTestCase):
|
||||||
|
Reference in New Issue
Block a user