<commit_before># coding=utf-8
import codecs
import unittest
import os, shutil, re

# Load a custom configuration for unit testing
os.environ['BDKD_DATASTORE_CONFIG'] = os.path.join(os.path.dirname(__file__), 
    '..', '..', 'conf', 'test.conf')
import bdkd.datastore

FIXTURES = os.path.join(os.path.dirname(__file__), 
    '..', '..', '..', 'fixtures')
TEST_PATH='/var/tmp/test'


class UtilitiesTest(unittest.TestCase):

    def test_common_directory_single(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b']),
                'a/b')

    def test_common_directory_common(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b', 'a/c']),
                'a')

    def test_common_directory_different(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b', 'c/d']),
                '')

    def test_common_directory_blanks(self):
        self.assertEquals(bdkd.datastore.common_directory(['', '']),
                '')


class ConfigurationTest(unittest.TestCase):
    def test_config_settings(self):
        settings = bdkd.datastore.settings()
        self.assertTrue(isinstance(settings, dict))
        self.assertTrue(settings['cache_root'] == 
                os.path.join(TEST_PATH, 'bdkd/cache'))
        self.assertTrue(settings['working_root'] == 
                os.path.join(TEST_PATH, 'bdkd/working'))

    def test_config_hosts(self):
        hosts = bdkd.datastore.hosts()
        self.assertTrue(isinstance(hosts, dict))
        self.assertTrue('test-host' in hosts)
        
    def test_config_repositories(self):
        self.assertTrue(isinstance(bdkd.datastore.repositories(), dict))

    def test_config_repository(self):
        self.assertTrue(bdkd.datastore.repository('test-repository'))
        self.assertFalse(bdkd.datastore.repository('does-not-exist'))


class HostTest(unittest.TestCase):

    def test_configured_host(self):
        hosts = bdkd.datastore.hosts()
        self.assertTrue(isinstance(hosts['test-host'], bdkd.datastore.Host))

    def test_host_init(self):
        host = bdkd.datastore.Host('access-key', 'secret-key', host='hostname', port=80, secure=True, calling_format=None)
        self.assertTrue(host)
        self.assertTrue(host.connection)
        self.assertTrue(host.netloc)


class RepositoryTest(unittest.TestCase):

    @classmethod
    def _clear_local(cls, repository):
        for tmp_path in [ repository.local_cache, repository.working ]:
            if tmp_path and tmp_path.startswith(TEST_PATH):
                if os.path.exists(tmp_path):
                    shutil.rmtree(tmp_path)

    @classmethod
    def fixture(cls):
        return bdkd.datastore.repository('test-repository')

    def setUp(self):
        self.repository = bdkd.datastore.repository('test-repository')
        self.resource = ResourceTest.fixture()
        self.resource_name = self.resource.name
        RepositoryTest._clear_local(self.repository)

    def test_configured_repository(self):
        repositories = bdkd.datastore.repositories()
        self.assertTrue(isinstance(repositories['test-repository'], bdkd.datastore.Repository))

    def test_repository_init_defaults(self):
        repository = bdkd.datastore.Repository(None, 'defaults')
        self.assertEquals(repository.host, None)
        self.assertEquals(repository.name, 'defaults')
        self.assertEquals(repository.local_cache, 
                os.path.join(bdkd.datastore.settings()['cache_root'], 
                    str(os.getuid()), 
                    repository.name))
        self.assertEquals(repository.working, 
                os.path.join(bdkd.datastore.settings()['working_root'], 
                    str(os.getuid()), 
                    str(os.getpid()), 
                    repository.name))
        self.assertEquals(repository.bucket, None)

    def test_list(self):
        self.assertEquals(len(self.repository.list()), 0)
        self.repository.save(self.resource)
        self.assertEquals(len(self.repository.list()), 1)

    def test_get(self):
        self.assertEquals(self.repository.get(self.resource_name), None)
        self.repository.save(self.resource)
        self.assertTrue(self.repository.get(self.resource_name))

    def test_save(self):
        self.assertEquals(self.repository.get(self.resource_name), None)
        self.repository.save(self.resource)
        self.assertEquals(self.resource.repository, self.repository)
        self.assertFalse(self.resource.is_edit)

    def test_edit_resource(self):
        # This is the default starting state of a Resource (i.e. unsaved)
        self.assertTrue(self.resource.is_edit)
        self.repository.save(self.resource)
        # After saving the resource should no longer be in edit mode
        self.assertFalse(self.resource.is_edit)
        self.assertTrue(self.resource.path.startswith(self.repository.local_cache))
        self.assertFalse(os.access(self.resource.path, os.W_OK))
        # Change to edit; path should be changed to repository's working
        self.repository.edit_resource(self.resource)
        self.assertTrue(self.resource.is_edit)
        self.assertTrue(self.resource.path.startswith(self.repository.working))
        self.assertTrue(os.access(self.resource.path, os.W_OK))

    def test_refresh_resource(self):
        self.repository.save(self.resource)
        self.repository.refresh_resource(self.resource)
        self.assertTrue(self.resource)

    def test_delete_resource(self):
        self.assertEquals(self.repository.get(self.resource_name), None)<commit_msg>BDKD-198: Added unit tests for Resource.save() and .set_edit()<commit_after># coding=utf-8
import codecs
import unittest
from mock import patch
import os, shutil, re

# Load a custom configuration for unit testing
os.environ['BDKD_DATASTORE_CONFIG'] = os.path.join(os.path.dirname(__file__), 
    '..', '..', 'conf', 'test.conf')
import bdkd.datastore

FIXTURES = os.path.join(os.path.dirname(__file__), 
    '..', '..', '..', 'fixtures')
TEST_PATH='/var/tmp/test'


class UtilitiesTest(unittest.TestCase):

    def test_common_directory_single(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b']),
                'a/b')

    def test_common_directory_common(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b', 'a/c']),
                'a')

    def test_common_directory_different(self):
        self.assertEquals(bdkd.datastore.common_directory(['a/b', 'c/d']),
                '')

    def test_common_directory_blanks(self):
        self.assertEquals(bdkd.datastore.common_directory(['', '']),
                '')


class ConfigurationTest(unittest.TestCase):
    def test_config_settings(self):
        settings = bdkd.datastore.settings()
        self.assertTrue(isinstance(settings, dict))
        self.assertTrue(settings['cache_root'] == 
                os.path.join(TEST_PATH, 'bdkd/cache'))
        self.assertTrue(settings['working_root'] == 
                os.path.join(TEST_PATH, 'bdkd/working'))

    def test_config_hosts(self):
        hosts = bdkd.datastore.hosts()
        self.assertTrue(isinstance(hosts, dict))
        self.assertTrue('test-host' in hosts)
        
    def test_config_repositories(self):
        self.assertTrue(isinstance(bdkd.datastore.repositories(), dict))

    def test_config_repository(self):
        self.assertTrue(bdkd.datastore.repository('test-repository'))
        self.assertFalse(bdkd.datastore.repository('does-not-exist'))


class HostTest(unittest.TestCase):

    def test_configured_host(self):
        hosts = bdkd.datastore.hosts()
        self.assertTrue(isinstance(hosts['test-host'], bdkd.datastore.Host))

    def test_host_init(self):
        host = bdkd.datastore.Host('access-key', 'secret-key', host='hostname', port=80, secure=True, calling_format=None)
        self.assertTrue(host)
        self.assertTrue(host.connection)
        self.assertTrue(host.netloc)


class RepositoryTest(unittest.TestCase):

    @classmethod
    def _clear_local(cls, repository):
        for tmp_path in [ repository.local_cache, repository.working ]:
            if tmp_path and tmp_path.startswith(TEST_PATH):
                if os.path.exists(tmp_path):
                    shutil.rmtree(tmp_path)

    @classmethod
    def fixture(cls):
        return bdkd.datastore.repository('test-repository')

    def setUp(self):
        self.repository = bdkd.datastore.repository('test-repository')
        self.resource = ResourceTest.fixture()
        self.resource_name = self.resource.name
        RepositoryTest._clear_local(self.repository)

    def test_configured_repository(self):
        repositories = bdkd.datastore.repositories()
        self.assertTrue(isinstance(repositories['test-repository'], bdkd.datastore.Repository))

    def test_repository_init_defaults(self):
        repository = bdkd.datastore.Repository(None, 'defaults')
        self.assertEquals(repository.host, None)
        self.assertEquals(repository.name, 'defaults')
        self.assertEquals(repository.local_cache, 
                os.path.join(bdkd.datastore.settings()['cache_root'], 
                    str(os.getuid()), 
                    repository.name))
        self.assertEquals(repository.working, 
                os.path.join(bdkd.datastore.settings()['working_root'], 
                    str(os.getuid()), 
                    str(os.getpid()), 
                    repository.name))
        self.assertEquals(repository.bucket, None)

    def test_list(self):
        self.assertEquals(len(self.repository.list()), 0)
        self.repository.save(self.resource)
        self.assertEquals(len(self.repository.list()), 1)

    def test_get(self):
        self.assertEquals(self.repository.get(self.resource_name), None)
        self.repository.save(self.resource)
        self.assertTrue(self.repository.get(self.resource_name))

    def test_save(self):
        self.assertEquals(self.repository.get(self.resource_name), None)
        self.repository.save(self.resource)
        self.assertEquals(self.resource.repository, self.repository)
        self.assertFalse(self.resource.is_edit)

    def test_edit_from_resource_no_repo(self):
        self.resource.repository = None;
        # Calling set_edit() on a resource without a repository should raise a 
        # ValueError
        self.assertRaises(ValueError, self.resource.set_edit)

    @patch('bdkd.datastore.Repository')
    def test_edit_from_resource(self, MockRepository):
        mock_repo = MockRepository()
        self.resource.repository = mock_repo;
        # Calling set_edit() on a resource should call its repository's 
        # edit_resource() method
        self.resource.set_edit()
        mock_repo.edit_resource.assert_called_with(self.resource)

    def test_save_from_resource_no_repo(self):
        self.resource.repository = None;
        # Calling save() on a resource without a repository should raise a 
        # ValueError
        self.assertRaises(ValueError, self.resource.save)

    @patch('bdkd.datastore.Repository')
    def test_save_from_resource(self, MockRepository):
        mock_repo = MockRepository()
        self.resource.repository = mock_repo;
        # Calling save() on a resource should call its repository's save() 
        # method with overwrite=True
        self.resource.save()
        mock_repo.save.assert_called_with(self.resource, overwrite=True)

    def test_edit_resource(self):
        # This is the default starting state of a Resource (i.e. unsaved)
        self.assertTrue(self.resource.is_edit)
        self.repository.save(self.resource)
        # After saving the resource should no longer be in edit mode
        self.assertFalse(self.resource.is_edit)
        self.assertTrue(self.resource.path.startswith(self.repository.local_cache))
        self.assertFalse(os.access(self.resource.path, os.W_OK))
        # Change to edit; path should be changed to repository's working
        self.repository.edit_resource(self.resource)
        self.assertTrue(self.resource.is_edit)
        self.assertTrue(self.resource.path.startswith(self.repository.working))
        self.assertTrue(os.access(self.resource.path, os.W_OK))

    def test_refresh_resource(self):
        self.repository.save(self.resource)
        self.repository.refresh_resource(self.resource)
        self.assertTrue(self.resource)

    def test_delete_resource(self):
        self.assertEquals(self.repository.get(self.resource_name), None)
{'fraction_non_alphanumeric': 0.11886490807354116, 'fraction_numerical': 0.0010391686650679456, 'mean_word_length': 3.033204384268214, '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': '6197849', 'n_tokens_mistral': 3766, 'n_tokens_neox': 3662, 'n_words': 549}