<commit_before>
from tests.testcase import TestCase
import edmunds.support.helpers as helpers
import os


class TestFile(TestCase):
    """
    Test the File
    """

    def set_up(self):
        """
        Set up the test case
        """

        super(TestFile, self).set_up()

        self.prefix = helpers.random_str(20) + '.'
        self.storage_directory = os.sep + 'storage' + os.sep
        self.clear_paths = []

    def tear_down(self):
        """
        Tear down the test case
        """

        super(TestFile, self).tear_down()

        # Remove all profiler files
        for directory in self.clear_paths:
            for root, subdirs, files in os.walk(directory):
                for file in files:
                    if file.startswith(self.prefix):
                        os.remove(os.path.join(root, file))

    def test_file(self):
        """
        Test the file
        """

        string = helpers.random_str(20)

        # Write config
        self.write_config([
            "from edmunds.storage.drivers.file import File \n",
            "from logging import WARNING \n",
            "APP = { \n",
            "   'storage': { \n",
            "       'instances': [ \n",
            "           { \n",
            "               'name': 'file',\n",
            "               'driver': File,\n",
            "               'directory': '%s',\n" % self.storage_directory,
            "               'prefix': '%s',\n" % self.prefix,
            "           }, \n",
            "       ], \n",
            "   }, \n",
            "   'log': { \n",
            "       'instances': [ \n",
            "       ], \n",
            "   }, \n",
            "} \n",
        ])

        # Create app
        app = self.create_application()
        directory = app.fs()._get_processed_path(None)
        self.clear_paths.append(directory)
        self.assert_equal(self.storage_directory, app.config('app.storage.instances')[0]['directory'])
        self.assert_equal(self.prefix, app.config('app.storage.instances')[0]['prefix'])

        # Write
        stream = app.fs().write_stream('nice.txt')
        try:
            stream.write(string)
        finally:
            stream.close()

        # Read
        stream = app.fs().read_stream('nice.txt')
        try:
            self.assert_in(string, stream.read())
        finally:
            stream.close()

        # Copy
        self.assert_true(app.fs().copy('nice.txt', 'nice2.txt'))
        stream = app.fs().read_stream('nice2.txt')
        try:
            self.assert_in(string, stream.read())
        finally:
            stream.close()

        # Exists and delete
        self.assert_true(app.fs().exists('nice2.txt'))
        self.assert_true(app.fs().delete('nice2.txt'))
        self.assert_false(app.fs().exists('nice2.txt'))<commit_msg>#21: Added coverage for copy and error fails scenarios.<commit_after>
from tests.testcase import TestCase
import edmunds.support.helpers as helpers
import os
import mock


class TestFile(TestCase):
    """
    Test the File
    """

    def set_up(self):
        """
        Set up the test case
        """

        super(TestFile, self).set_up()

        self.prefix = helpers.random_str(20) + '.'
        self.storage_directory = os.sep + 'storage' + os.sep
        self.clear_paths = []

    def tear_down(self):
        """
        Tear down the test case
        """

        super(TestFile, self).tear_down()

        # Remove all profiler files
        for directory in self.clear_paths:
            for root, subdirs, files in os.walk(directory):
                for file in files:
                    if file.startswith(self.prefix):
                        os.remove(os.path.join(root, file))

    def test_file(self):
        """
        Test the file
        """

        string = helpers.random_str(20)

        # Write config
        self.write_config([
            "from edmunds.storage.drivers.file import File \n",
            "from logging import WARNING \n",
            "APP = { \n",
            "   'storage': { \n",
            "       'instances': [ \n",
            "           { \n",
            "               'name': 'file',\n",
            "               'driver': File,\n",
            "               'directory': '%s',\n" % self.storage_directory,
            "               'prefix': '%s',\n" % self.prefix,
            "           }, \n",
            "       ], \n",
            "   }, \n",
            "   'log': { \n",
            "       'instances': [ \n",
            "       ], \n",
            "   }, \n",
            "} \n",
        ])

        # Create app
        app = self.create_application()
        directory = app.fs()._get_processed_path(None)
        self.clear_paths.append(directory)
        self.assert_equal(self.storage_directory, app.config('app.storage.instances')[0]['directory'])
        self.assert_equal(self.prefix, app.config('app.storage.instances')[0]['prefix'])

        # Write
        stream = app.fs().write_stream('nice.txt')
        try:
            stream.write(string)
        finally:
            stream.close()

        # Read
        stream = app.fs().read_stream('nice.txt')
        try:
            self.assert_in(string, stream.read())
        finally:
            stream.close()

        # Copy
        self.assert_true(app.fs().copy('nice.txt', 'nice2.txt'))
        stream = app.fs().read_stream('nice2.txt')
        try:
            self.assert_in(string, stream.read())
        finally:
            stream.close()

        # Exists and delete
        self.assert_true(app.fs().exists('nice2.txt'))
        self.assert_true(app.fs().delete('nice2.txt'))
        self.assert_false(app.fs().exists('nice2.txt'))

    def test_copy_error(self):
        """
        Test error scenario for copy
        :return:    void
        """

        string = helpers.random_str(20)

        # Write config
        self.write_config([
            "from edmunds.storage.drivers.file import File \n",
            "from logging import WARNING \n",
            "APP = { \n",
            "   'storage': { \n",
            "       'instances': [ \n",
            "           { \n",
            "               'name': 'file',\n",
            "               'driver': File,\n",
            "               'directory': '%s',\n" % self.storage_directory,
            "               'prefix': '%s',\n" % self.prefix,
            "           }, \n",
            "       ], \n",
            "   }, \n",
            "   'log': { \n",
            "       'instances': [ \n",
            "       ], \n",
            "   }, \n",
            "} \n",
            ])

        # Create app
        app = self.create_application()
        directory = app.fs()._get_processed_path(None)
        self.clear_paths.append(directory)

        # Write
        stream = app.fs().write_stream('nice.txt')
        try:
            stream.write(string)
        finally:
            stream.close()

        # Copy
        with mock.patch('shutil.copy2', side_effect=IOError()):
            self.assert_false(app.fs().copy('nice.txt', 'nice2.txt'))
            self.assert_false(app.fs().copy('nice.txt', 'nice2.txt', raise_errors=False))

            with self.assert_raises(IOError):
                app.fs().copy('nice.txt', 'nice2.txt', raise_errors=True)

    def test_delete_error(self):
        """
        Test delete scenario for error
        :return:    void
        """

        string = helpers.random_str(20)

        # Write config
        self.write_config([
            "from edmunds.storage.drivers.file import File \n",
            "from logging import WARNING \n",
            "APP = { \n",
            "   'storage': { \n",
            "       'instances': [ \n",
            "           { \n",
            "               'name': 'file',\n",
            "               'driver': File,\n",
            "               'directory': '%s',\n" % self.storage_directory,
            "               'prefix': '%s',\n" % self.prefix,
            "           }, \n",
            "       ], \n",
            "   }, \n",
            "   'log': { \n",
            "       'instances': [ \n",
            "       ], \n",
            "   }, \n",
            "} \n",
            ])

        # Create app
        app = self.create_application()
        directory = app.fs()._get_processed_path(None)
        self.clear_paths.append(directory)

        # Write
        stream = app.fs().write_stream('nice.txt')
        try:
            stream.write(string)
        finally:
            stream.close()

        # Copy
        with mock.patch('os.remove', side_effect=IOError()):
            self.assert_false(app.fs().delete('nice.txt'))
            self.assert_false(app.fs().delete('nice.txt', raise_errors=False))

            with self.assert_raises(IOError):
                app.fs().delete('nice.txt', raise_errors=True)
{'fraction_non_alphanumeric': 0.14809417040358744, 'fraction_numerical': 0.003587443946188341, 'mean_word_length': 1.3588048651507139, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6867696', 'n_tokens_mistral': 2632, 'n_tokens_neox': 2561, 'n_words': 468}