def test_broken_metadata_with_backup(self):
        """test that each time a metadata file is updated a .old is kept"""
        self.shelf['bad_file'] = {'value': 'old'}
        path = self.shelf.key_file('bad_file')
        self.assertFalse(path_exists(path+'.old'))
        self.assertEqual({'value': 'old'}, self.shelf['bad_file'])
        # force the creation of the .old file
        self.shelf['bad_file'] = {'value': 'new'}
        self.assertTrue(path_exists(path+'.old'))
        # check that the new value is there
        self.assertEqual({'value': 'new'}, self.shelf['bad_file'])
        # write the current md file fwith 0 bytes
        open_file(path, 'w').close()
        # test that the old value is retrieved
        self.assertEqual({'value': 'old'}, self.shelf['bad_file'])

        self.shelf['broken_pickle'] = {'value': 'old'}
        path = self.shelf.key_file('broken_pickle')
        # check that .old don't exist
        self.assertFalse(path_exists(path+'.old'))
        # force the creation of the .old file
        self.shelf['broken_pickle'] = {'value': 'new'}
        # check that .old exists
        self.assertTrue(path_exists(path+'.old'))
        # check that the new value is there
        self.assertEqual({'value': 'new'}, self.shelf['broken_pickle'])
        # write random bytes to the md file
        with open_file(path, 'w') as f:
            f.write(BROKEN_PICKLE)
        # check that the old value is retrieved
        self.assertEqual({'value': 'old'}, self.shelf['broken_pickle'])

 def test_remove_dir(self):
     """Test the remove dir."""
     testdir = os.path.join(self.basedir, 'foodir')
     os.mkdir(testdir)
     assert path_exists(testdir)
     remove_dir(testdir)
     self.assertFalse(path_exists(testdir))

def get_config_files():
    """ return the path to the config files or and empty list.
    The search path is based on the paths returned by load_config_paths
    but it's returned in reverse order (e.g: /etc/xdg first).
    """
    config_files = []
    for xdg_config_dir in load_config_paths('ubuntuone'):
        config_file = os.path.join(xdg_config_dir, CONFIG_FILE)
        if path_exists(config_file):
            config_files.append(config_file)

        config_logs = os.path.join(xdg_config_dir, CONFIG_LOGS)
        if path_exists(config_logs):
            config_files.append(config_logs)

    # reverse the list as load_config_paths returns the user dir first
    config_files.reverse()
    # if we are running from a branch, get the config files from it too
    config_file = os.path.join(os.path.dirname(__file__), os.path.pardir,
                               os.path.pardir, 'data', CONFIG_FILE)
    if path_exists(config_file):
        config_files.append(config_file)

    config_logs = os.path.join('data', CONFIG_LOGS)
    if path_exists(config_logs):
        config_files.append(config_logs)

    return config_files

 def test_create_dirs_already_exists_symlink_too(self):
     """test that creating a Main instance works as expected."""
     link = os.path.join(self.root, 'Shared With Me')
     make_link(self.shares, link)
     self.assertTrue(is_link(link))
     self.assertTrue(path_exists(self.shares))
     self.assertTrue(path_exists(self.root))
     main = self.build_main()
     # check that the shares link is actually a link
     self.assertTrue(is_link(main.shares_dir_link))

 def test_create_dirs_none_exists(self):
     """test that creating a Main instance works as expected."""
     # remove the existing dirs
     remove_dir(self.root)
     remove_dir(self.shares)
     main = self.build_main()
     # check that the shares link is actually a link
     self.assertTrue(is_link(main.shares_dir_link))
     self.assertTrue(path_exists(self.shares))
     self.assertTrue(path_exists(self.root))

    def test_rename_file(self, target=None):
        """Rename a file."""
        if target is None:
            target = os.path.join(self.basedir, 'target')

        assert path_exists(self.testfile)
        rename(self.testfile, target)

        self.assertFalse(path_exists(self.testfile),
            'Path %r should not exist after rename.' % self.testfile)
        self.assertTrue(path_exists(target),
            'Path %r should exist after rename.' % target)

 def test_delete_backups_too(self):
     """test that delitem also deletes the .old/.new files left around"""
     self.shelf["foo"] = "bar"
     # create the .old backup
     self.shelf["foo"] = "bar1"
     path = self.shelf.key_file('foo')
     # create a .new file (a hard reboot during the rename dance)
     open_file(path+'.new', 'w').close()
     # write 0 bytes to both
     del self.shelf['foo']
     self.assertFalse(path_exists(path))
     self.assertFalse(path_exists(path+'.old'), 'there is a .old file!')
     self.assertFalse(path_exists(path+'.new'), 'there is a .new file!')

    def test_rename_dir(self, source=None, target=None):
        """Rename a dir."""
        if source is None:
            source = os.path.join(self.basedir, 'source')
            os.mkdir(source)
        if target is None:
            target = os.path.join(self.basedir, 'target')

        rename(source, target)

        self.assertFalse(path_exists(source),
            'Path %r should not exist after rename.' % source)
        self.assertTrue(path_exists(target),
            'Path %r should exist after rename.' % target)

    def is_ignored(self, path):
        """should we ignore this path?"""
        # check first if the platform code knows hat to do with it
        if not self.platform_is_ignored(path):
            # check if we can read
            if path_exists(path) and not access(path):
                self.log.warning("Ignoring path as we don't have enough "
                                 "permissions to track it: %r", path)
                return True

            is_conflict = self.conflict_RE.search
            dirname, filename = os.path.split(path)
            # ignore conflicts
            if is_conflict(filename):
                return True
            # ignore partial downloads
            if filename == '.u1partial' or filename.startswith('.u1partial.'):
                return True

            # and ignore paths that are inside conflicts (why are we even
            # getting the event?)
            if any(part.endswith('.u1partial') or is_conflict(part)
                   for part in dirname.split(os.path.sep)):
                return True

            if self.ignore_RE is not None and self.ignore_RE.match(filename):
                return True

            return False
        return True

 def test_allow_writes(self):
     """Test for allow_writes."""
     set_dir_readonly(self.basedir)
     with allow_writes(self.basedir):
         foo_dir = os.path.join(self.basedir, 'foo')
         os.mkdir(foo_dir)
         self.assertTrue(path_exists(foo_dir))

 def test_write_extra(self):
     """Writing the throttling back to the file, with extra sections."""
     conf_file = os.path.join(
         self.test_root, 'test_write_extra_config.conf')
     # write some throttling values to the config file
     with open_file(conf_file, 'w') as fp:
         fp.write('[__main__]\n')
         fp.write('log_level = INFO\n')
         fp.write('disable_ssl_verify = True\n')
         fp.write('\n')
         fp.write('[bandwidth_throttling]\n')
         fp.write('on = False\n')
         fp.write('read_limit = 2000\n')
         fp.write('write_limit = 200\n')
     self.assertTrue(path_exists(conf_file))
     conf = config._Config(conf_file)
     conf.set_throttling(True)
     conf.set_throttling_read_limit(3000)
     conf.set_throttling_write_limit(300)
     conf.save()
     # load the config in a barebone ConfigParser and check
     conf_1 = ConfigParser()
     conf_1.read(conf_file)
     self.assertThrottlingSection(conf_1, conf, True, 3000, 300)
     self.assertEqual(conf_1.get('__main__', 'log_level'),
                      conf.get('__main__', 'log_level'))
     self.assertEqual(conf_1.getboolean('__main__', 'disable_ssl_verify'),
                      conf.getboolean('__main__', 'disable_ssl_verify'))

    def start(self):
        """Start the comparison."""
        log_info("start scan all volumes")
        to_scan = self._get_volumes(all_volumes=False)
        for vol in to_scan:
            # check that the path exists in disk
            if not path_exists(vol.path):
                log_warning('Volume dissapeared: %r - %r', vol.volume_id, vol.path)
                if isinstance(vol, volume_manager.Share):
                    log_debug('Removing %r metadata', vol.volume_id)
                    self.vm.share_deleted(vol.volume_id)
                elif isinstance(vol, volume_manager.UDF):
                    log_debug('Unsubscribing %r', vol.volume_id)
                    self.vm.unsubscribe_udf(vol.volume_id)
                # if root is missing, we should crash and burn as it's
                # created on each startup!
                continue
            try:
                mdobj = self.fsm.get_by_path(vol.path)
            except KeyError:
                # this could happen in a strange corruption situation where FSM
                # lost the share information, so we remove it, because VM will
                # download it again
                self.vm.share_deleted(vol.volume_id)
            else:
                self._queue.appendleft((vol, vol.path, mdobj.mdid, False))

        # first of all, remove old partials and clean trash
        self._remove_partials()
        self._process_limbo()
        self._process_ro_shares()

        yield self._queue_scan()
        self._show_broken_nodes()

    def __init__(self, path, auto_merge=True, dead_bytes_threshold=0.5, max_immutable_files=20):
        """Initialize the instance.

        @param auto_merge: disable auto merge/compaction.
        @param dead_bytes_threshold: the limit factor of dead vs live bytes to
            trigger a merge and/or live file rotation.
        @param max_immutable_files: the max number of inactive files to use,
            once this value is reached a merge is triggered.
        """
        logger.info("Initializing Tritcask on: %s", path)
        self._keydir = Keydir()
        self.base_path = path
        self.dead_bytes_threshold = dead_bytes_threshold
        self.max_immutable_files = max_immutable_files
        self.auto_merge = auto_merge
        if not path_exists(self.base_path):
            make_dir(self.base_path, recursive=True)
        elif not is_dir(self.base_path):
            raise ValueError("path must be a directory.")
        self.live_file = None
        self._immutable = {}
        self._find_data_files()
        self._build_keydir()
        # now check if we should rotate the live file
        # and merge immutable ones
        self._rotate_and_merge()
        # check if we found a live data file
        # if not, define one (it will be created later)
        if self.live_file is None:
            # it's a clean start, let's create the first file
            self.live_file = DataFile(self.base_path)

    def test_broken_metadata_items(self):
        """Test that broken metadata is ignored during iteritems."""
        self.shelf['ok_key'] = {'status': 'this is valid metadata'}
        self.shelf['bad_file'] = {}
        path = self.shelf.key_file('bad_file')
        open_file(path, 'w').close()
        self.assertRaises(KeyError, self.shelf.__getitem__, 'bad_file')
        self.assertEqual(1, len(list(self.shelf.items())))
        self.assertFalse(path_exists(path))

        self.shelf['broken_pickle'] = {}
        path = self.shelf.key_file('broken_pickle')
        with open_file(path, 'w') as f:
            f.write(BROKEN_PICKLE)
        self.assertRaises(KeyError, self.shelf.__getitem__, 'broken_pickle')
        self.assertEqual(1, len(list(self.shelf.items())))
        self.assertFalse(path_exists(path))

 def _check_move_file(self, src, dst, real_dst):
     """Check that a file was indeed moved."""
     with open_file(src, "rb") as f:
         contents = f.read()
     recursive_move(src, dst)
     with open_file(real_dst, "rb") as f:
         self.assertEqual(contents, f.read())
     self.assertFalse(path_exists(src))

    def test_movetotrash_file_ok(self):
        """Move a file to trash ok.

        Just check it was removed because can't monkeypatch the trash.
        to see that that was actually called.
        """
        move_to_trash(self.testfile)
        self.assertFalse(path_exists(self.testfile))

 def mktemp(self, name="temp"):
     """ Customized mktemp that accepts an optional name argument. """
     tempdir = os.path.join(self.tmpdir, name)
     if path_exists(tempdir):
         self.rmtree(tempdir)
     self.makedirs(tempdir)
     self.addCleanup(self.rmtree, tempdir)
     assert isinstance(tempdir, str)
     return tempdir

 def add_share(self, share):
     """Add share to the shares dict."""
     self.shares[share.id] = share
     # if the share don't exists, create it
     if not path_exists(share.path):
         make_dir(share.path)
     # if it's a ro share, change the perms
     if not share.can_write():
         set_dir_readonly(share.path)

 def test_ignore_one(self):
     """Test ignore files config, one regex."""
     conf_file = os.path.join(self.test_root, 'test_new_config.conf')
     with open_file(conf_file, 'w') as fp:
         fp.write('[__main__]\n')
         fp.write('ignore = .*\\.pyc\n')  # all .pyc files
     self.assertTrue(path_exists(conf_file))
     self.cp.read([conf_file])
     self.cp.parse_all()
     self.assertEqual(self.cp.get('__main__', 'ignore').value, [r'.*\.pyc'])

    def test_movetotrash_dir_ok(self):
        """Move a dir to trash ok.

        Just check it was removed because can't monkeypatch the trash
        to see that that was actually called.
        """
        path = os.path.join(self.basedir, 'foo')
        make_dir(path)
        move_to_trash(path)
        self.assertFalse(path_exists(path))