# Copyright 2018 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import logging
import os
import sys

from dbus.mainloop.glib import DBusGMainLoop

from autotest_lib.client.bin import test
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import utils
from autotest_lib.client.common_lib.cros import smbprovider

class enterprise_SmbProviderDaemon(test.test):
    """
    Test for SmbProvider Daemon.

    """

    version = 1

    WORKGROUP = ''
    USERNAME = ''
    PASSWORD = ''

    def setup(self):
        """
        Compiles protobufs for error type and input/output parameters.

        """

        os.chdir(self.srcdir)
        utils.make('OUT_DIR=.')

    def initialize(self):
        """
        Initializes the D-Bus loop and creates Python wrapper.

        """

        bus_loop = DBusGMainLoop(set_as_default=True)
        self._smbprovider = smbprovider.SmbProvider(bus_loop, self.srcdir)

        # Append path for directory_entry_pb2 imports.
        sys.path.append(self.srcdir)

    def run_once(self, mount_path):
        """
        Runs smbproviderd D-Bus commands.

        @param mount_path: Address of the SMB share.
        """

        self.sanity_test(mount_path)

    def _generate_random_id(self, size):
        """
        Generates a random string of size N.

        @param size: Size of the generated string.

        @return: Returns a random alphanumeric string of size N.

        """

        import string
        import random

        return ''.join(random.choice(string.ascii_uppercase +
                                string.digits) for i in range(size))

    def sanity_test(self, mount_path):
        """
        Sanity test that runs through all filesystem operations
        on the SmbProvider Daemon.

        @param mount_path: Address of the SMB share.

        """

        from directory_entry_pb2 import ERROR_EXISTS

        # Mount the SMB share.
        mount_id = self._check_mount(mount_path)

        # Generate random directory.
        rand_dir_id = self._generate_random_id(10)
        test_dir = '/autotest_' + rand_dir_id + '/'
        self._check_create_directory(mount_id, test_dir, False)

        # Get metadata of a directory.
        metadata = self._check_get_metadata(mount_id, test_dir)

        # Check that GetMetadata has correct values of a directory.
        self._check_metadata(test_dir[1:-1], 0, True, metadata)

        # Create file inside directory.
        test_file = test_dir + '1.txt'
        self._check_create_file(mount_id, test_file)

        # Open file with Read-Only privileges.
        file_id = self._check_open_file(mount_id, test_file, False)
        self._check_close_file(mount_id, file_id)

        # Open + Close file with Read-Only privileges.
        file_id = self._check_open_file(mount_id, test_file, False)
        self._check_close_file(mount_id, file_id)

        # Open file for writing.
        file_id = self._check_open_file(mount_id, test_file, True)

        # Write data to file.
        data = 'Hello World!'
        self._check_write_file(mount_id, file_id, 0, data)

        # Read data from file.
        read_data = self._check_read_file(mount_id, file_id, 0, len(data))

        # Close file.
        self._check_close_file(mount_id, file_id)

        # Verify data is written to file correctly.
        self._check_contents(data, read_data)

        # Get the metadata of the file.
        metadata = self._check_get_metadata(mount_id, test_file)

        # Check that GetMetadeta has correct values of a file.
        # TODO(jimmyxgong): len() only works properly for UTF-8. Find way to
        # get size universally.
        self._check_metadata('1.txt', len(data), False, metadata)

        # Delete file.
        self._check_delete_entry(mount_id, test_file, False)

        # Create recursive directories.
        recursive_dir = test_dir + 'test1/test2/'
        self._check_create_directory(mount_id, recursive_dir, True)

        # Create file within the new directory.
        test_file2 = recursive_dir + '2.txt'
        self._check_create_file(mount_id, test_file2)

        # Check moving to existing entry is handled.
        self._check_move_entry(mount_id, test_file2, test_dir, ERROR_EXISTS)

        # Move file up to root test directory.
        self._check_move_entry(mount_id, test_file2, test_dir + 'moved.txt')

        # Move back down to original location.
        self._check_move_entry(mount_id, test_dir + 'moved.txt', test_file2)

        # TODO(jimmyxgong): Delete contents of autotest directory recursively.
        self._check_delete_entry(mount_id, test_file2, False)
        self._check_delete_entry(mount_id, test_dir + 'test1/test2/', False)
        self._check_delete_entry(mount_id, test_dir + 'test1/', False)

        # Delete autotest directory.
        self._check_delete_entry(mount_id, test_dir, False)

        # Unmount the SMB share.
        self._check_unmount(mount_id)

    def _check_mount(self, mount_path):
        """
        Checks that mount is working.

        @param mount_path: Address of the SMB share.

        @return mount_id: Unique identifier of the mount.

        """

        from directory_entry_pb2 import ERROR_OK

        error, mount_id = self._smbprovider.mount(mount_path,
                                                  self.WORKGROUP,
                                                  self.USERNAME,
                                                  self.PASSWORD)

        if mount_id < 0 :
            error.TestFail('Unexpected failure with mount id.')

        self._check_result('Mount', error)
        return mount_id

    def _check_unmount(self, mount_id):
        """
        Checks that unmount is working.

        @param mount_id: Unique identifier of the mount.

        """

        error = self._smbprovider.unmount(mount_id)

        self._check_result('Unmount', error)

    def _check_get_metadata(self, mount_id, entry_path):
        """
        Checks that get metadata is working.

        @param mount_id: Unique identifier of the mount.
        @param entry_path: Path of the entry.

        @return: GetMetaDataEntryOptionsProto blob string returned by the D-Bus
                 call.

        """

        error, metadata_blob = self._smbprovider.get_metadata(mount_id,
                                                              entry_path)

        self._check_result('Get Metadata', error)

        return metadata_blob

    def _check_metadata(self, entry_path, size, is_dir, metadata_blob):
        """
        Checks that metadata_blob has the correct values.

        @param entry_path: File path of the entry we are checking.
        @param size: Size of the entry in bytes.
        @param is_dir: Boolean that indicates whether the entry is a directory.
        @param metadata_blob: Blob that contains metadata of the entry.

        """

        if entry_path != metadata_blob.name or \
                size != metadata_blob.size or \
                is_dir != metadata_blob.is_directory:
            logging.error('Failed: Metadata is incorrect')
            raise error.TestFail('Unexpected error with metadata')

    def _check_create_file(self, mount_id, file_path):
        """
        Checks that create file is working.

        @param mount_id: Unique identifier of the mount.
        @param file_path: Path of where the new file will be created.

        """

        error = self._smbprovider.create_file(mount_id, file_path)

        self._check_result('Create File', error)

    def _check_open_file(self, mount_id, file_path, writeable):
        """
        Checks that open file is working.

        @param mount_id: Unique identifier of the mount.
        @param file_path: Path of where the file is located.
        @param writeable: Boolean to indicated whether the file should
                be opened with write access.

        """

        error, file_id = self._smbprovider.open_file(mount_id,
                                                     file_path,
                                                     writeable)

        if not file_id:
            error.TestFail('Unexpected file id failure.')

        self._check_result('Open File', error)

        return file_id

    def _check_close_file(self, mount_id, file_id):
        """
        Checks that close file is working.

        @param mount_id: Unique identifier of the mount.
        @param file_id: Unique identifier of the file.

        """

        error = self._smbprovider.close_file(mount_id, file_id)

        self._check_result('Close File', error)

    def _check_write_file(self, mount_id, file_id, offset, data):
        """
        Checks that write file is working.

        @param mount_id: Unique identifier of the mount.
        @param file_id: Unique identifier of the file.
        @param offset: Offset of the file to start writing to.
        @param data: Data to be written.

        """

        error = self._smbprovider.write_file(mount_id, file_id, offset, data)

        self._check_result('Write File', error)

    def _check_read_file(self, mount_id, file_id, offset, length):
        """
        Checks that read file is working.

        @param mount_id: Unique identifier of the mount.
        @param file_id: Unique identifier of the file.
        @param offset: Offset of the file to start reading from.
        @param length: Length of data to read in bytes.

        @return A buffer containing the data read.

        """

        error, fd = self._smbprovider.read_file(mount_id, file_id, offset,
                                                                   length)

        self._check_result('Read File', error)

        return fd

    def _check_contents(self, data, read_data):
        """
        Checks that read_data is equal to data.

        @param data: Original data to be compared to.
        @param read_data: Data to be compared to the original data.

        """

        if data != read_data:
            logging.error('Failed: Written data does not match Read data')
            raise error.TestFail(
                    'Unexpected mismatch of written data and read data.\
                    Expected: %s , but got: %s' % (data, read_data))

    def _check_create_directory(self, mount_id,
                                      directory_path,
                                      recursive):
        """
        Checks that create directory is working.

        @param mount_id: Unique identifier of the mount.
        @param directory_path: Path for the test directory.
        @param recursive: Boolean to indicate whether directories should be
                created recursively.

        """

        error = self._smbprovider.create_directory(mount_id,
                                                   directory_path,
                                                   recursive)

        self._check_result('Create Directory', error)

    def _check_delete_entry(self, mount_id, entry_path, recursive):
        """
        Checks that delete an entry works.

        @param mount_id: Unique identifier of the mount.
        @param entry_path: Path to the file/directory to delete.
        @param recursive: Boolean to indicate recursive deletes.

        """

        error = self._smbprovider.delete_entry(mount_id,
                                               entry_path,
                                               recursive)

        self._check_result('Delete Entry', error)

    def _check_move_entry(self, mount_id, source_path, target_path,
                                                       expected=None):
        """
        Checks that move entry is working.

        @param mount_id: Unique identifier of the mount.
        @param source_path: Path of the entry to be moved.
        @param target_path: Path of the destination for the entry.
        @param expected: Expected ErrorType. Default: None (ERROR_OK)

        """

        error = self._smbprovider.move_entry(mount_id,
                                             source_path,
                                             target_path)

        self._check_result('Move Entry', error, expected)

    def _check_result(self, method_name, result, expected=None):
        """
        Helper to check error codes and throw on mismatch.

        Checks whether the returned ErrorType from a D-Bus call to smbproviderd
        matches the expected ErrorType. In case of a mismatch, throws a
        TestError.

        @param method_name: Name of the D-Bus method that was called.
        @param result: ErrorType returned from the D-Bus call.
        @param expected: Expected ErrorType. Default: ErrorType.ERROR_OK.

        """

        from directory_entry_pb2 import ErrorType
        from directory_entry_pb2 import ERROR_OK

        if not expected:
            expected = ERROR_OK

        if result != expected:
            logging.error('Failed to run %s', method_name)
            raise error.TestFail(
                    '%s failed with error %s (%s), expected %s (%s)' % (
                    method_name, result, ErrorType.Name(result), expected,
                    ErrorType.Name(expected)))