def test_clean_non_root(
            self,
            listdir,
            access,
            is_shared,
            unlink_shared,
            delete,
            config,
            rmdir,
            lexists):
        """
        Ensure that empty directories more than 1 level up from root are removed.
        """
        listdir.return_value = None
        access.return_value = True
        path = '/storage/pulp/content/test/lvl1/lvl2/thing.remove'
        storage_dir = '/storage/pulp'
        is_shared.return_value = False
        config.get.return_value = storage_dir
        lexists.return_value = True

        OrphanManager.delete_orphaned_file(path)
        rmdir.assert_has_calls(
            [
                call('/storage/pulp/content/test/lvl1/lvl2'),
                call('/storage/pulp/content/test/lvl1')
            ])

    def test_delete_exception(self, is_dir, unlink, log_error):
        path = 'path-1'
        is_dir.return_value = False
        unlink.side_effect = OSError

        # test
        OrphanManager.delete(path)

        # validation
        self.assertTrue(log_error.called)

    def test_delete_file(self, is_file, unlink, rmtree):
        path = 'path-1'
        is_file.return_value = True

        # test
        OrphanManager.delete(path)

        # validation
        is_file.assert_called_with(path)
        unlink.assert_called_with(path)
        self.assertFalse(rmtree.called)

    def test_path_not_exists(self, is_shared, unlink_shared, delete, config, rmdir, lexists):
        lexists.return_value = False
        path = '/tmp/path-1'

        # test
        OrphanManager.delete_orphaned_file(path)

        # validation
        lexists.assert_called_once_with(path)
        self.assertFalse(is_shared.called)
        self.assertFalse(delete.called)
        self.assertFalse(rmdir.called)
        self.assertFalse(unlink_shared.called)

    def test_delete_target_not_sibling(self, delete, read_link, listdir):
        path = '/parent/links/path-1'
        content = '/NOT-SAME-PARENT/content'
        read_link.return_value = content
        listdir.return_value = []

        # test
        OrphanManager.unlink_shared(path)

        # validation
        read_link.assert_called_once_with(path)
        listdir.assert_called_once_with(os.path.dirname(path))
        delete.assert_called_once_with(path)

    def test_delete_links_not_empty(self, delete, read_link, listdir):
        path = '/parent/links/path-1'
        content = '/parent/content'
        read_link.return_value = content
        listdir.return_value = ['link-2']

        # test
        OrphanManager.unlink_shared(path)

        # validation
        read_link.assert_called_once_with(path)
        listdir.assert_called_once_with(os.path.dirname(path))
        delete.assert_called_once_with(path)

    def test_delete_dir(self, is_file, is_link, unlink, rmtree):
        path = 'path-1'
        is_file.return_value = False
        is_link.return_value = False

        # test
        OrphanManager.delete(path)

        # validation
        is_file.assert_called_with(path)
        is_link.assert_called_with(path)
        rmtree.assert_called_with(path)
        self.assertFalse(unlink.called)

    def test_not_shared(self, is_shared, unlink_shared, delete, config):
        path = '/path-1'
        storage_dir = '/storage/pulp/dir'
        is_shared.return_value = False
        config.get.return_value = storage_dir

        # test
        OrphanManager.delete_orphaned_file(path)

        # validation
        is_shared.assert_called_once_with(storage_dir, path)
        delete.assert_called_once_with(path)
        self.assertFalse(unlink_shared.called)

    def test_clean_no_access(self, mls, maccess, is_shared, unlink_shared, delete, config, m_rmdir):
        """
        Ensure that rmdir is not called when user does not have access to dir.
        """
        mls.return_value = None
        maccess.return_value = False
        path = '/storage/pulp/content/test/lvl1/lvl2/thing.remove'
        storage_dir = '/storage/pulp'
        is_shared.return_value = False
        config.get.return_value = storage_dir

        OrphanManager.delete_orphaned_file(path)
        self.assertFalse(m_rmdir.called)

    def test_clean_nonempty(self, mls, maccess, is_shared, unlink_shared, delete, config, m_rmdir):
        """
        Ensure that nonempty directories are not removed.
        """
        mls.return_value = ['some', 'files']
        maccess.return_value = True
        path = '/storage/pulp/content/test/lvl1/lvl2/thing.remove'
        storage_dir = '/storage/pulp'
        is_shared.return_value = False
        config.get.return_value = storage_dir

        OrphanManager.delete_orphaned_file(path)
        self.assertFalse(m_rmdir.called)

 def test_generate_orphans_by_type_with_unit_keys_invalid_type(self):
     """
     Assert that when an invalid content type is passed to
     generate_orphans_by_type_with_unit_keys a MissingResource exception is raised.
     """
     self.assertRaises(
         pulp_exceptions.MissingResource,
         OrphanManager.generate_orphans_by_type_with_unit_keys('Not a type').next)

    def test_delete_all(self, delete, read_link, listdir):
        path = '/parent/links/path-1'
        content = '/parent/content'
        read_link.return_value = content
        listdir.return_value = []

        # test
        OrphanManager.unlink_shared(path)

        # validation
        read_link.assert_called_once_with(path)
        listdir.assert_called_once_with(os.path.dirname(path))
        self.assertEqual(
            delete.call_args_list,
            [
                ((path,), {}),
                ((content,), {}),
            ])

    def test_not_links_dir(self, is_link):
        storage_dir = '/var/lib/pulp'
        path = os.path.join(storage_dir, 'content/shared/repo-id/OTHER/link-1')
        is_link.return_value = True

        # test
        shared = OrphanManager.is_shared(storage_dir, path)

        # validation
        self.assertFalse(shared)

    def test_generate_orphans_by_type_with_unit_keys_invalid_type(self, mock_get_unit_key_fields):
        """
        Assert that when an invalid content type is passed to
        generate_orphans_by_type_with_unit_keys a MissingResource exception is raised.
        """
        # simulate the type not being found
        mock_get_unit_key_fields.side_effect = ValueError

        self.assertRaises(
            pulp_exceptions.MissingResource,
            OrphanManager.generate_orphans_by_type_with_unit_keys('Not a type').next)

 def setUp(self):
     super(OrphanManagerTests, self).setUp()
     content_type_db.update_database([PHONY_TYPE_1, PHONY_TYPE_2])
     self.content_root = tempfile.mkdtemp(prefix='content_orphan_manager_unittests-')
     self.orphan_manager = OrphanManager()

class OrphanManagerTests(base.PulpServerTests):

    def setUp(self):
        super(OrphanManagerTests, self).setUp()
        content_type_db.update_database([PHONY_TYPE_1, PHONY_TYPE_2])
        self.content_root = tempfile.mkdtemp(prefix='content_orphan_manager_unittests-')
        self.orphan_manager = OrphanManager()

    def tearDown(self):
        super(OrphanManagerTests, self).tearDown()
        RepoContentUnit.get_collection().remove(safe=True)
        content_type_db.clean()
        if os.path.exists(self.content_root): # can be removed by delete operations
            shutil.rmtree(self.content_root)

    def number_of_files_in_content_root(self):
        if not os.path.exists(self.content_root): # can be removed by delete operations
            return 0
        contents = os.listdir(self.content_root)
        return len(contents)

    # utilities test methods ---------------------------------------------------

    def test_content_creation(self):
        self.assertTrue(self.number_of_files_in_content_root() == 0)
        content_unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        self.assertTrue(os.path.exists(content_unit['_storage_path']))
        self.assertTrue(self.number_of_files_in_content_root() == 1)

    # list test methods --------------------------------------------------------

    def test_list_one_orphan(self):
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 0)
        content_unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 1)
        self.assertTrue(orphans[0]['_id'] == content_unit['_id'])

    def test_list_two_orphans(self):
        unit_1 = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        unit_2 = gen_content_unit(PHONY_TYPE_2.id, self.content_root)
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 2)

    def test_list_orphans_by_type(self):
        unit_1 = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        unit_2 = gen_content_unit(PHONY_TYPE_2.id, self.content_root)
        orphans_1 = self.orphan_manager.list_orphans_by_type(PHONY_TYPE_1.id)
        self.assertTrue(len(orphans_1) == 1)
        orphans_2 = self.orphan_manager.list_orphans_by_type(PHONY_TYPE_2.id)
        self.assertTrue(len(orphans_2) == 1)

    def test_get_orphan(self):
        unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        orphan = self.orphan_manager.get_orphan(PHONY_TYPE_1.id, unit['_id'])
        self.assertTrue(orphan['_id'] == unit['_id'])

    def test_get_missing_orphan(self):
        self.assertRaises(pulp_exceptions.MissingResource,
                          self.orphan_manager.get_orphan,
                          PHONY_TYPE_1.id, 'non-existent')

    def test_associated_unit(self):
        unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        associate_content_unit_with_repo(unit)
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 0)
        unassociate_content_unit_from_repo(unit)
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 1)

    # delete test methods ------------------------------------------------------

    def test_delete_one_orphan(self):
        unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        self.orphan_manager.delete_all_orphans()
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 0)
        self.assertTrue(self.number_of_files_in_content_root() == 0)

    def test_delete_one_orphan_with_directory(self):
        unit = gen_content_unit_with_directory(PHONY_TYPE_1.id, self.content_root)
        self.orphan_manager.delete_all_orphans()
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 0)
        self.assertTrue(self.number_of_files_in_content_root() == 0)

    def test_delete_by_type(self):
        unit_1 = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
        unit_2 = gen_content_unit(PHONY_TYPE_2.id, self.content_root)
        self.assertTrue(self.number_of_files_in_content_root() == 2)
        self.orphan_manager.delete_orphans_by_type(PHONY_TYPE_1.id)
        orphans = self.orphan_manager.list_all_orphans()
        self.assertTrue(len(orphans) == 1)
        self.assertFalse(os.path.exists(unit_1['_storage_path']))
        self.assertTrue(os.path.exists(unit_2['_storage_path']))

    def test_delete_by_id(self):
        unit = gen_content_unit(PHONY_TYPE_1.id, self.content_root)
#.........这里部分代码省略.........