mirror of
https://github.com/haiwen/seafile-server.git
synced 2025-09-01 15:36:37 +00:00
add unit test for file operation api
This commit is contained in:
111
tests/test_file_operation/test_file_operation.py
Normal file
111
tests/test_file_operation/test_file_operation.py
Normal file
@@ -0,0 +1,111 @@
|
||||
import pytest
|
||||
import os
|
||||
import time
|
||||
from tests.config import USER
|
||||
from seaserv import seafile_api as api
|
||||
|
||||
file_name = 'test.txt'
|
||||
new_file_name = 'new_test.txt'
|
||||
empty_file_name = 'empty_test.txt'
|
||||
new_empty_file_name = 'new_empty_test.txt'
|
||||
file_content = 'test file content'
|
||||
file_path = os.getcwd() + '/' + file_name
|
||||
dir_name = "test_dir"
|
||||
|
||||
def create_the_file ():
|
||||
with open(file_path, 'w') as fp:
|
||||
fp.write(file_content)
|
||||
|
||||
def test_file_operation():
|
||||
t_repo_version = 1
|
||||
t_repo_id1 = api.create_repo('test_file_operation1', '', USER, passwd = None)
|
||||
|
||||
create_the_file()
|
||||
|
||||
# test post_file
|
||||
assert api.post_file(t_repo_id1, file_path, '/', file_name, USER) == 0
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id1, '/' + file_name)
|
||||
t_file_size = len(file_content)
|
||||
assert t_file_size == api.get_file_size(t_repo_id1, t_repo_version, t_file_id)
|
||||
|
||||
# test post_dir
|
||||
assert api.post_dir(t_repo_id1, '/', dir_name, USER) == 0
|
||||
|
||||
# test copy_file (synchronize)
|
||||
t_copy_file_result1 = api.copy_file(t_repo_id1, '/', file_name, t_repo_id1, '/', new_file_name, USER, 0, 1)
|
||||
assert t_copy_file_result1
|
||||
assert t_copy_file_result1.task_id is None
|
||||
assert not t_copy_file_result1.background
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id1, '/' + new_file_name)
|
||||
assert t_file_size == api.get_file_size(t_repo_id1, t_repo_version, t_file_id)
|
||||
|
||||
# test copy_file (asynchronous)
|
||||
t_repo_id2 = api.create_repo('test_file_operation2', '', USER, passwd = None)
|
||||
t_copy_file_result2 = api.copy_file(t_repo_id1, '/', file_name, t_repo_id2, '/', file_name, USER, 1, 0)
|
||||
assert t_copy_file_result2
|
||||
assert t_copy_file_result2.task_id
|
||||
assert t_copy_file_result2.background
|
||||
while True:
|
||||
time.sleep(0.1)
|
||||
t_copy_task = api.get_copy_task(t_copy_file_result2.task_id)
|
||||
if t_copy_task.successful:
|
||||
break;
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id2, '/' + file_name)
|
||||
assert t_file_size == api.get_file_size(t_repo_id2, t_repo_version, t_file_id)
|
||||
|
||||
# test move_file (synchronize)
|
||||
t_move_file_info1 = api.get_dirent_by_path(t_repo_id1, '/' + new_file_name)
|
||||
t_move_file_result1 = api.move_file(t_repo_id1, '/', new_file_name, t_repo_id1, '/' + dir_name, new_file_name, 1, USER, 0, 1)
|
||||
assert t_move_file_result1
|
||||
t_move_file_info2 = api.get_dirent_by_path(t_repo_id1, '/' + dir_name + '/' + new_file_name)
|
||||
assert t_move_file_info1.mtime == t_move_file_info2.mtime
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id1, '/' + new_file_name)
|
||||
assert t_file_id is None
|
||||
|
||||
# test move_file (asynchronous)
|
||||
t_move_file_result2 = api.move_file(t_repo_id1, '/', file_name, t_repo_id2, '/' , new_file_name, 1, USER, 1, 0)
|
||||
assert t_move_file_result2
|
||||
assert t_move_file_result2.task_id
|
||||
assert t_move_file_result2.background
|
||||
while True:
|
||||
time.sleep(0.1)
|
||||
t_move_task = api.get_copy_task(t_move_file_result2.task_id)
|
||||
if t_move_task.successful:
|
||||
break
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id2, '/' + new_file_name)
|
||||
assert t_file_size == api.get_file_size(t_repo_id2, t_repo_version, t_file_id)
|
||||
|
||||
# test post_empty_file
|
||||
assert api.post_empty_file(t_repo_id1, '/' + dir_name, empty_file_name, USER) == 0
|
||||
t_file_id = api.get_file_id_by_path(t_repo_id1, '/' + dir_name + '/' + empty_file_name)
|
||||
assert api.get_file_size(t_repo_id1, t_repo_version, t_file_id) == 0
|
||||
|
||||
# test rename_file
|
||||
assert api.rename_file(t_repo_id1, '/' + dir_name, empty_file_name, new_empty_file_name, USER) == 0
|
||||
|
||||
#test put_file
|
||||
t_new_file_id = api.put_file(t_repo_id1, file_path, '/' + dir_name, new_empty_file_name, USER, None)
|
||||
assert t_new_file_id
|
||||
|
||||
# test get_file_revisions
|
||||
t_commit_list = api.get_file_revisions(t_repo_id2, None, '/' + file_name, 2)
|
||||
assert t_commit_list
|
||||
assert len(t_commit_list) == 2
|
||||
assert t_commit_list[0].creator_name == USER
|
||||
|
||||
# test del_file
|
||||
assert api.del_file(t_repo_id2, '/', file_name, USER) == 0
|
||||
|
||||
# test get_deleted
|
||||
t_deleted_file_list = api.get_deleted(t_repo_id2, 1)
|
||||
assert t_deleted_file_list
|
||||
assert len(t_deleted_file_list) == 2
|
||||
assert t_deleted_file_list[0].obj_name == file_name
|
||||
assert t_deleted_file_list[0].basedir == '/'
|
||||
|
||||
assert api.del_file(t_repo_id1, '/' + dir_name, new_empty_file_name, USER) == 0
|
||||
assert api.del_file(t_repo_id1, '/' + dir_name, new_file_name, USER) == 0
|
||||
assert api.del_file(t_repo_id2, '/', new_file_name, USER) == 0
|
||||
|
||||
time.sleep(1)
|
||||
api.remove_repo(t_repo_id1)
|
Reference in New Issue
Block a user