QSfera/Desktop/test/gui/steps/file_context.py

import os
import re
import builtins
import shutil
import zipfile
from os.path import isfile, join, isdir, exists
from behave import when as When, then as Then
from sure import ensure

from helpers.SetupClientHelper import get_resource_path, get_temp_resource_path
from helpers.SyncHelper import (
    wait_for_client_to_be_ready,
    listen_sync_status_for_item,
    wait_for,
)
from helpers.ConfigHelper import get_config, is_windows
from helpers.FilesHelper import (
    build_conflicted_regex,
    sanitize_path,
    can_read,
    can_write,
    read_file_content,
    get_size_in_bytes,
    prefix_path_namespace,
    remember_path,
    convert_path_separators_for_os,
    get_file_for_upload,
)


def folder_exists(folder_path, timeout=1000):
    return wait_for(
        lambda: isdir(sanitize_path(folder_path)),
        timeout,
    )


def file_exists(file_path, timeout=1000):
    return wait_for(
        lambda: isfile(sanitize_path(file_path)),
        timeout,
    )


# To create folders in a temporary directory, we set is_temp_folder True
# And if is_temp_folder is True, the create_folder function create folders in tempFolderPath
def create_folder(foldername, username=None, is_temp_folder=False):
    if is_temp_folder:
        folder_path = join(get_config('tempFolderPath'), foldername)
    else:
        folder_path = get_resource_path(foldername, username)
    os.makedirs(prefix_path_namespace(convert_path_separators_for_os(folder_path)))


def rename_file_folder(source, destination):
    source = get_resource_path(source)
    destination = get_resource_path(destination)
    os.rename(source, destination)


def create_file_with_size(filename, filesize, is_temp_folder=False):
    if is_temp_folder:
        file = join(get_config('tempFolderPath'), filename)
    else:
        file = get_resource_path(filename)
    with open(prefix_path_namespace(file), 'wb') as f:
        f.seek(get_size_in_bytes(filesize) - 1)
        f.write(b'\0')


def write_file(resource, content):
    with open(prefix_path_namespace(resource), 'w', encoding='utf-8') as f:
        f.write(content)


def wait_and_write_file(path, content):
    wait_for_client_to_be_ready()
    listen_sync_status_for_item(get_resource_path(path), 'FILE')
    write_file(path, content)


def wait_and_try_to_write_file(resource, content):
    wait_for_client_to_be_ready()
    listen_sync_status_for_item(get_resource_path(resource), 'FILE')
    try:
        write_file(resource, content)
    except:
        pass


def create_zip(resources, zip_file_name, cwd=''):
    os.chdir(cwd)
    with zipfile.ZipFile(zip_file_name, 'w') as zipped_file:
        for resource in resources:
            zipped_file.write(resource)


def extract_zip(zip_file_path, destination_dir):
    with zipfile.ZipFile(zip_file_path, 'r') as zip_file:
        zip_file.extractall(destination_dir)


def add_copy_suffix(resource_path, resource_type):
    suffix = ' (Copy)'
    if resource_type == 'file':
        source_dir = resource_path.rsplit('.', 1)
        return source_dir[0] + suffix + '.' + source_dir[-1]
    return resource_path + suffix


def copy_resource(resource_type, source, destination, from_files_for_upload=False):
    if from_files_for_upload:
        source = get_file_for_upload(source)
    else:
        source = get_resource_path(source)
    destination = get_resource_path(destination)
    if source == destination and destination != '/':
        destination = add_copy_suffix(source, resource_type)

    wait_for_client_to_be_ready()
    listen_sync_status_for_item(destination, resource_type)
    if resource_type == 'folder':
        return shutil.copytree(source, destination)
    return shutil.copy2(source, destination)


def move_resource(username, resource_type, source, destination, is_temp_folder=False):
    if not is_temp_folder:
        source = get_resource_path(source, username)
    if destination == '/':
        destination = ''
    destination = get_resource_path(destination, username)

    wait_for_client_to_be_ready()
    listen_sync_status_for_item(destination, resource_type)
    shutil.move(source, destination)


def deleteResource(resource, resource_type):
    wait_for_client_to_be_ready()
    listen_sync_status_for_item(resource, resource_type)
    resource_path = sanitize_path(get_resource_path(resource))
    if resource_type == 'file':
        os.remove(resource_path)
    else:
        shutil.rmtree(resource_path)


@When(
    'user "{username}" creates a file "{filename}" with the following content inside the sync folder'
)
def step(context, username, filename):
    file = get_resource_path(filename, username)
    wait_and_write_file(convert_path_separators_for_os(file), context.text)


@When('user "{username}" creates a folder "{foldername}" inside the sync folder')
def step(context, username, foldername):
    wait_for_client_to_be_ready()
    create_folder(foldername, username)


@Given('user "{username}" has created a folder "{foldername}" inside the sync folder')
def step(context, username, foldername):
    create_folder(foldername, username)


@When(
    'user "{user}" creates a file "{filename}" with size "{filesize}" inside the sync folder'
)
def step(context, user, filename, filesize):
    create_file_with_size(filename, filesize)


@When(r'the user copies (file|folder) "([^"]*)" into folder "([^"]*)"', regexp=True)
def step(context, resource_type, resource_name, destination_dir):
    copy_resource(resource_type, resource_name, destination_dir, False)


@When(
    'the user copies {resource_type:ResourceType} "{resource_name}" into the same directory'
)
def step(context, resource_type, resource_name):
    copy_resource(resource_type, resource_name, resource_name, False)


@When('the user renames a file "{source}" to "{destination}"')
@When('the user renames a folder "{source}" to "{destination}"')
def step(context, source, destination):
    wait_for_client_to_be_ready()
    rename_file_folder(source, destination)


@Then(
    'the file "{file_path}" should exist on the file system with the following content'
)
def step(context, file_path):
    expected = context.text
    file_path = get_resource_path(file_path)
    with open(file_path, 'r', encoding='utf-8') as f:
        contents = f.read()
    with ensure(
        '{0} expected to exist with content "{1}" but has content "{2}"',
        file_path,
        expected,
        contents,
    ):
        contents.should.equal(expected)


@Then('the {resource_type:ResourceType} "{resource}" should exist on the file system')
def step(context, resource_type, resource):
    resource_path = get_resource_path(resource)
    resource_exists = False
    timeout = get_config('maxSyncTimeout') * 1000
    if resource_type == 'file':
        resource_exists = file_exists(resource_path, timeout)
    else:
        resource_exists = folder_exists(resource_path, timeout)

    with ensure(
        '{0} "{1}" should exist, but it does not',
        resource_type.capitalize(),
        resource,
    ):
        resource_exists.should.be.true


@Then(
    'the {resource_type:ResourceType} "{resource}" should not exist on the file system'
)
def step(context, resource_type, resource):
    resource_path = get_resource_path(resource)
    with ensure(
        '{0} "{1}" should not exist, but it does',
        resource_type.capitalize(),
        resource,
    ):
        exists(resource_path).should.be.false


@Given('the user has changed the content of local file "|any|" to:')
def step(context, filename):
    file_content = '\n'.join(context.multiLineText)
    wait_and_write_file(get_resource_path(filename), file_content)


@Then(
    'a conflict file for "|any|" should exist on the file system with the following content'
)
def step(context, filename):
    expected = '\n'.join(context.multiLineText)

    onlyfiles = [
        f for f in os.listdir(get_resource_path()) if isfile(get_resource_path(f))
    ]
    found = False
    pattern = re.compile(build_conflicted_regex(filename))
    for file in onlyfiles:
        if pattern.match(file):
            with open(get_resource_path(file), 'r', encoding='utf-8') as f:
                if f.read() == expected:
                    found = True
                    break

    if not found:
        raise AssertionError('Conflict file not found with given name')


@When('the user overwrites the file "{resource}" with content "{content}"')
def step(context, resource, content):
    resource = get_resource_path(resource)
    wait_and_write_file(resource, content)


@When('the user tries to overwrite the file "|any|" with content "|any|"')
def step(context, resource, content):
    resource = get_resource_path(resource)
    wait_and_try_to_write_file(resource, content)


@When('user "|any|" tries to overwrite the file "|any|" with content "|any|"')
def step(context, user, resource, content):
    resource = get_resource_path(resource, user)
    wait_and_try_to_write_file(resource, content)


@When('the user deletes the {resource_type:ResourceType} "{resource_name}"')
def step(context, resource_type, resource_name):
    deleteResource(resource_name, resource_type)


@When('user "|any|" creates the following files inside the sync folder:')
def step(context, username):
    for row in context.table[1:]:
        file = get_resource_path(row[0], username)
        wait_and_write_file(file, '')


@Given('the user has created a folder "|any|" in temp folder')
def step(context, folder_name):
    create_folder(folder_name, is_temp_folder=True)


@Given(
    'the user has created "|any|" files each of size "|any|" bytes inside folder "|any|" in temp folder'
)
def step(context, file_number, file_size, folder_name):
    current_sync_path = get_temp_resource_path(folder_name)
    if folder_exists(current_sync_path):
        file_size = builtins.int(file_size)
        for i in range(0, builtins.int(file_number)):
            file_name = f'file{i}.txt'
            create_file_with_size(join(current_sync_path, file_name), file_size, True)
    else:
        raise FileNotFoundError(
            f"Folder '{folder_name}' does not exist in the temp folder"
        )


@When(
    r'user "([^"]*)" reads the content of file "([^"]*)"',
    regexp=True,
)
def step(context, username, file):
    file_path = get_resource_path(file, username)
    with open(file_path, 'r') as f:
        f.read()


@When(
    r'user "([^"]*)" moves (folder|file) "([^"]*)" from the temp folder into the sync folder',
    regexp=True,
)
def step(context, username, resource_type, resource_name):
    source_dir = join(get_config('tempFolderPath'), resource_name)
    move_resource(username, resource_type, source_dir, '/', True)


@When(
    r'user "([^"]*)" moves (folder|file) "([^"]*)" to the temp folder',
    regexp=True,
)
def step(context, username, resource_type, resource_name):
    destination = join(get_config('tempFolderPath'), resource_name)
    move_resource(username, resource_type, resource_name, destination)


@When(
    'user "{username}" moves {resource_type:ResourceType} "{source}" to "{destination}" in the sync folder'
)
def step(context, username, resource_type, source, destination):
    move_resource(username, resource_type, source, destination)


@Then('user "{user}" should be able to open the file "{file_name}" on the file system')
def step(context, user, file_name):
    file_path = get_resource_path(file_name, user)
    with ensure(
        'File should be readable but user "{0}" cannot read file "{1}"',
        user,
        file_name,
    ):
        can_read(file_path).should.be.true


@Then(
    'as "{user}" the file "{file_name}" should have content "{content}" on the file system'
)
def step(context, user, file_name, content):
    file_path = get_resource_path(file_name, user)
    file_content = read_file_content(file_path)
    with ensure(
        'File "{0}" should have content "{1}" but got "{2}"',
        file_name,
        content,
        file_content,
    ):
        content.should.equal(file_content)


@Then('user "|any|" should not be able to edit the file "|any|" on the file system')
def step(context, user, file_name):
    file_path = get_resource_path(file_name, user)
    test.compare(not can_write(file_path), True, 'File should not be writable')


@Given(
    'the user has created a zip file "|any|" with the following resources in the temp folder'
)
def step(context, zip_file_name):
    resource_list = []

    for row in context.table[1:]:
        resource_list.append(row[0])
        resource = join(get_config('tempFolderPath'), row[0])
        if row[1] == 'folder':
            os.makedirs(resource)
        elif row[1] == 'file':
            content = ''
            if len(row) > 2 and row[2]:
                content = row[2]
            write_file(resource, content)
    create_zip(resource_list, zip_file_name, get_config('tempFolderPath'))


@When('user "|any|" unzips the zip file "|any|" inside the sync root')
def step(context, username, zip_file_name):
    destination_dir = get_resource_path('/', username)
    zip_file_path = join(destination_dir, zip_file_name)
    extract_zip(zip_file_path, destination_dir)


@When('user "|any|" copies file "|any|" to temp folder')
def step(context, username, source):
    wait_for_client_to_be_ready()
    source_dir = get_resource_path(source, username)
    destination_dir = get_temp_resource_path(source)
    shutil.copy2(source_dir, destination_dir)


@Given('the user has created folder "|any|" in the default home path')
def step(context, folder_name):
    folder_path = join(get_config('home_dir'), folder_name)
    os.makedirs(prefix_path_namespace(folder_path))
    remember_path(folder_path)
    # when account is added, folder with suffix will be created
    remember_path(f'{folder_path} (2)')


@Given(
    r'the user has copied file "([^"]*)" from outside the sync folder to "([^"]*)" in the sync folder',
    regexp=True,
)
def step(context, resource_name, destination):
    copy_resource('file', resource_name, destination, True)


@When(
    r'the user copies file "([^"]*)" from outside the sync folder to "([^"]*)" in the sync folder',
    regexp=True,
)
def step(context, resource_name, destination):
    copy_resource('file', resource_name, destination, True)


@When('the user deletes the following files')
def step(context):
    wait_for_client_to_be_ready()

    for row in context.table[1:]:
        filename = row[0]
        deleteResource(filename, 'file')


@Given('user "|any|" has created a file "|any|" with size "|any|" in the sync folder')
def step(context, _, filename, filesize):
    create_file_with_size(filename, filesize)