def __call__(self):
        """
        Invoke the callable object.

        All work is performed in the repository working directory and cleaned up after the call.

        :return: The final synchronization report.
        :rtype: SyncProgressReport
        """
        self.canceled = False
        self.report = SyncProgressReport(self.conduit)
        self.tmp_dir = mkdtemp(dir=self.repo.working_dir)
        try:
            manifest = self._fetch_manifest()
            if manifest is not None:
                module_paths = self._fetch_modules(manifest)
                self._import_modules(module_paths)
        finally:
            # Update the progress report one last time
            self.report.update_progress()

            shutil.rmtree(self.tmp_dir)
            self.tmp_dir = None

        return self.report

    def __call__(self, repository):
        """
        Invoke the callable object.
        All work is performed in the repository working directory and
        cleaned up after the call.

        :param repository: A Pulp repository object.
        :type repository: pulp.server.plugins.model.Repository
        :return: The final synchronization report.
        :rtype: SyncProgressReport
        """
        self.canceled = False
        self.report = SyncProgressReport(self.conduit)
        self.tmp_dir = mkdtemp(dir=repository.working_dir)
        try:
            inventory = Inventory(self.conduit)
            self._run(inventory)
        finally:
            # Update the progress report one last time
            self.report.update_progress()

            shutil.rmtree(self.tmp_dir)
            self.tmp_dir = None

        return self.report

    def __init__(self, repo, sync_conduit, config):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config

        self.progress_report = SyncProgressReport(sync_conduit)
        self.downloader = None

    def __init__(self, repo, sync_conduit, config, is_cancelled_call):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config
        self.is_cancelled_call = is_cancelled_call

        self.progress_report = SyncProgressReport(sync_conduit)

    def setUp(self):
        super(PuppetStatusRendererTests, self).setUp()
        self.renderer = PuppetStatusRenderer(self.context)

        self.config['logging'] = {'filename' : 'test-extension-status.log'}

        self.sync_report = SyncProgressReport.from_progress_dict(IMPORTER_REPORT)
        self.publish_report = PublishProgressReport.from_progress_dict(DISTRIBUTOR_REPORT)

    def test_directory_synchronization(self, forge_call, mock_call):
        conduit = Mock()
        repository = Mock()
        config = {constants.CONFIG_FEED: 'http://host/tmp/%s' % constants.MANIFEST_FILENAME}
        progress_report = SyncProgressReport(conduit)
        progress_report.metadata_state = constants.STATE_SUCCESS
        progress_report.modules_state = constants.STATE_SUCCESS
        mock_call.return_value = progress_report

        # test

        plugin = PuppetModuleImporter()
        report = plugin.sync_repo(repository, conduit, config)

        # validation

        mock_call.assert_called_with(repository)
        self.assertEquals(report, conduit.build_success_report.return_value)
        self.assertFalse(forge_call.called)

    def __init__(self, repo, sync_conduit, config):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config

        self.progress_report = SyncProgressReport(sync_conduit)
        self.downloader = None
        # Since SynchronizeWithPuppetForge creates a Nectar downloader for each unit, we cannot
        # rely on telling the current downloader to cancel. Therefore, we need another state
        # tracker to check in the download units loop.
        self._canceled = False

    def test_forge_synchronization(self, failed_call, mock_call):
        conduit = Mock()
        repository = Mock()
        config = {constants.CONFIG_FEED: 'http://host/tmp/forge'}

        # directory synchronization failure needed so the importer
        # will retry using the forge synchronization.
        failed_report = SyncProgressReport(conduit)
        failed_report.metadata_state = constants.STATE_FAILED
        failed_call.return_value = failed_report

        progress_report = SyncProgressReport(conduit)
        progress_report.metadata_state = constants.STATE_FAILED
        mock_call.return_value = progress_report

        # test

        plugin = PuppetModuleImporter()
        report = plugin.sync_repo(repository, conduit, config)

        # validation
        mock_call.assert_called_with()
        self.assertEquals(report, conduit.build_failure_report.return_value)

    def display_report(self, progress_report):

        # Sync Steps
        if constants.IMPORTER_ID in progress_report:
            sync_report = SyncProgressReport.from_progress_dict(progress_report[constants.IMPORTER_ID])
            self._display_sync_metadata_step(sync_report)
            self._display_sync_modules_step(sync_report)

        # Publish Steps
        if constants.DISTRIBUTOR_ID in progress_report:
            publish_report = PublishProgressReport.from_progress_dict(progress_report[constants.DISTRIBUTOR_ID])
            self._display_publish_modules_step(publish_report)
            self._display_publish_metadata_step(publish_report)
            self._display_publish_http_https_step(publish_report)

class SynchronizeWithDirectory(object):
    """
    A callable object used to synchronize with a directory of packaged puppet modules.
    The source of the import is a directory containing a PULP_MANIFEST and
    multiple puppet built puppet modules.

    :ivar repo: A Pulp repository object
    :type repo: pulp.plugins.model.Repository
    :ivar conduit: Provides access to relevant Pulp functionality.
    :type conduit: pulp.plugins.conduits.repo_sync.RepoSyncConduit
    :ivar config: Plugin configuration.
    :type config: pulp.plugins.config.PluginCallConfiguration
    :ivar report: An import report.
    :type report: SyncProgressReport
    :ivar canceled: The operation canceled flag.
    :type canceled: bool
    :ivar tmp_dir: The path to the temporary directory used to download files.
    :type tmp_dir: str
    """

    @staticmethod
    def _extract_metadata(module_path):
        """
        Extract the puppet module metadata from the tarball at the specified path.
        Search the tarball content for a file named: */metadata.json and extract
        it into temporary directory. Then read the file and return the json decoded content.

        :param module_path: The fully qualified path to the module.
        :type module_path: str

        :return: The puppet module metadata.
        :rtype: dict
        """
        tmp_dir = mkdtemp(dir=os.path.dirname(module_path))
        try:
            with closing(tarfile.open(module_path)) as tarball:
                for member in tarball.getmembers():
                    path = member.name.split('/')
                    if path[-1] == constants.MODULE_METADATA_FILENAME:
                        tarball.extract(member, tmp_dir)
                        with open(os.path.join(tmp_dir, member.name)) as fp:
                            return json.load(fp)
        finally:
            shutil.rmtree(tmp_dir)

    def __init__(self, repo, conduit, config):
        """
        :param repo: A Pulp repository object
        :type repo: pulp.plugins.model.Repository
        :param conduit: Provides access to relevant Pulp functionality.
        :type conduit: pulp.plugins.conduits.repo_sync.RepoSyncConduit
        :param config: Plugin configuration.
        :type config: pulp.plugins.config.PluginCallConfiguration
        """
        self.repo = repo
        self.conduit = conduit
        self.config = config
        self.report = None
        self.canceled = False
        self.tmp_dir = None

    def feed_url(self):
        """
        Get the feed URL from the configuration and ensure it has a trailing '/' so urljoin will
        work correctly.

        :return: The feed URL.
        :rtype: str
        """
        url = self.config.get(constants.CONFIG_FEED)
        if not url.endswith('/'):
            url += '/'
        return url

    def cancel(self):
        """
        Cancel puppet module import.
        """
        self.canceled = True

    def _download(self, urls):
        """
        Download files by URL.

        Encapsulates nectar details and provides a simplified method of downloading files.

        :param urls: A list of tuples: (url, destination).  The *url* and *destination* are both
                     strings.  The *destination* is the fully qualified path to where the file is
                     to be downloaded.
        :type urls: list

        :return: The nectar reports.  Tuple of: (succeeded_reports, failed_reports)
        :rtype: tuple
        """
        feed_url = self.feed_url()
        nectar_config = importer_config_to_nectar_config(self.config.flatten())
        nectar_class = URL_TO_DOWNLOADER[urlparse(feed_url).scheme]
        downloader = nectar_class(nectar_config)
        listener = DownloadListener(self, downloader)

#.........这里部分代码省略.........

class SynchronizeWithPuppetForge(object):
    """
    Used to perform a single sync of a puppet repository. This class will
    maintain state relevant to the run and should not be reused across runs.
    """

    def __init__(self, repo, sync_conduit, config):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config

        self.progress_report = SyncProgressReport(sync_conduit)
        self.downloader = None
        # Since SynchronizeWithPuppetForge creates a Nectar downloader for each unit, we cannot
        # rely on telling the current downloader to cancel. Therefore, we need another state
        # tracker to check in the download units loop.
        self._canceled = False

    def __call__(self):
        """
        Sync according to the configured state of the instance and return a report.

        This function will make update progress as appropriate.

        This function executes serially, and does not create any threads. It will not return until
        either a step fails or the entire sync is complete.

        :return: the report object to return to Pulp from the sync call
        :rtype: SyncProgressReport
        """
        msg = _('Beginning sync for repository <%(repo_id)s>')
        msg_dict = {'repo_id': self.repo.id}
        _logger.info(msg, msg_dict)

        # quit now if there is no feed URL defined
        if not self.config.get(constants.CONFIG_FEED):
            self.progress_report.metadata_state = STATE_FAILED
            msg = _('Cannot perform repository sync on a repository with no feed')
            self.progress_report.metadata_error_message = msg
            self.progress_report.update_progress()
            return self.progress_report.build_final_report()

        try:
            metadata = self._parse_metadata()
            if not metadata:
                report = self.progress_report.build_final_report()
                return report

            self._import_modules(metadata)
        finally:
            # One final progress update before finishing
            self.progress_report.update_progress()

            return self.progress_report

    def cancel(self):
        """
        Cancel an in-progress sync, if there is one.
        """
        self._canceled = True
        if self.downloader is None:
            return
        self.downloader.cancel()

    def _parse_metadata(self):
        """
        Takes the necessary actions (according to the run configuration) to
        retrieve and parse the repository's metadata. This call will return
        either the successfully parsed metadata or None if it could not
        be retrieved or parsed. The progress report will be updated with the
        appropriate description of what went wrong in the event of an error,
        so the caller should interpret a None return as an error occurring and
        not continue the sync.

        :return: object representation of the metadata
        :rtype:  RepositoryMetadata
        """
        msg = _('Beginning metadata retrieval for repository <%(repo_id)s>')
        msg_dict = {'repo_id': self.repo.id}
        _logger.info(msg, msg_dict)

        self.progress_report.metadata_state = STATE_RUNNING
        self.progress_report.update_progress()

        start_time = datetime.now()

        # Retrieve the metadata from the source
        try:
            downloader = self._create_downloader()
            self.downloader = downloader
            metadata_json_docs = downloader.retrieve_metadata(self.progress_report)

        except Exception as e:
            if self._canceled:
                msg = _('Exception occurred on canceled metadata download: %(exc)s')
                msg_dict = {'exc': e}
                _logger.warn(msg, msg_dict)
                self.progress_report.metadata_state = STATE_CANCELED
                return None
            msg = _('Exception while retrieving metadata for repository <%(repo_id)s>')
#.........这里部分代码省略.........

class SynchronizeWithPuppetForge(object):
    """
    Used to perform a single sync of a puppet repository. This class will
    maintain state relevant to the run and should not be reused across runs.
    """

    def __init__(self, repo, sync_conduit, config):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config

        self.progress_report = SyncProgressReport(sync_conduit)
        self.downloader = None
        # Since SynchronizeWithPuppetForge creats a Nectar downloader for each unit, we cannot
        # rely on telling the current downloader to cancel. Therefore, we need another state tracker
        # to check in the download units loop.
        self._canceled = False

    def __call__(self):
        """
        Performs the sync operation according to the configured state of the
        instance. The report to be sent back to Pulp is returned from this
        call. This call will make calls into the conduit's progress update
        as appropriate.

        This call executes serially. No threads are created by this call. It
        will not return until either a step fails or the entire sync is
        completed.

        :return: the report object to return to Pulp from the sync call
        :rtype:  SyncProgressReport
        """
        _logger.info('Beginning sync for repository <%s>' % self.repo.id)

        # quit now if there is no feed URL defined
        if not self.config.get(constants.CONFIG_FEED):
            self.progress_report.metadata_state = STATE_FAILED
            self.progress_report.metadata_error_message = _(
                'Cannot perform repository sync on a repository with no feed')
            self.progress_report.update_progress()
            return self.progress_report.build_final_report()

        try:
            metadata = self._parse_metadata()
            if not metadata:
                report = self.progress_report.build_final_report()
                return report

            self._import_modules(metadata)
        finally:
            # One final progress update before finishing
            self.progress_report.update_progress()

            return self.progress_report

    def cancel(self):
        """
        Cancel an in-progress sync, if there is one.
        """
        self._canceled = True
        if self.downloader is None:
            return
        self.downloader.cancel()

    def _parse_metadata(self):
        """
        Takes the necessary actions (according to the run configuration) to
        retrieve and parse the repository's metadata. This call will return
        either the successfully parsed metadata or None if it could not
        be retrieved or parsed. The progress report will be updated with the
        appropriate description of what went wrong in the event of an error,
        so the caller should interpet a None return as an error occuring and
        not continue the sync.

        :return: object representation of the metadata
        :rtype:  RepositoryMetadata
        """
        _logger.info('Beginning metadata retrieval for repository <%s>' % self.repo.id)

        self.progress_report.metadata_state = STATE_RUNNING
        self.progress_report.update_progress()

        start_time = datetime.now()

        # Retrieve the metadata from the source
        try:
            downloader = self._create_downloader()
            self.downloader = downloader
            metadata_json_docs = downloader.retrieve_metadata(self.progress_report)

        except Exception, e:
            if self._canceled:
                _logger.warn('Exception occurred on canceled metadata download: %s' % e)
                self.progress_report.metadata_state = STATE_CANCELED
                return None
            _logger.exception('Exception while retrieving metadata for repository <%s>' % self.repo.id)
            self.progress_report.metadata_state = STATE_FAILED
            self.progress_report.metadata_error_message = _('Error downloading metadata')
            self.progress_report.metadata_exception = e
            self.progress_report.metadata_traceback = sys.exc_info()[2]
#.........这里部分代码省略.........

class SynchronizeWithDirectory(object):
    """
    A callable object used to synchronize with a directory of packaged puppet modules.
    The source of the import is a directory containing a PULP_MANIFEST and
    multiple puppet built puppet modules.

    :ivar conduit: Provides access to relevant Pulp functionality.
    :type conduit: pulp.plugins.conduits.unit_import.ImportUnitConduit
    :ivar config: Plugin configuration.
    :type config: pulp.plugins.config.PluginCallConfiguration
    :ivar report: An import report.
    :type report: SyncProgressReport
    :ivar canceled: The operation canceled flag.
    :type canceled: bool
    :ivar tmp_dir: The path to the temporary directory used to download files.
    :type tmp_dir: str
    """

    @staticmethod
    def _extract_metadata(module_path):
        """
        Extract the puppet module metadata from the tarball at the specified path.
        Search the tarball content for a file named: */metadata.json and extract
        it into temporary directory.  Then read the file and return the json decoded content.

        :param module_path: The fully qualified path to the module.
        :type module_path: str
        :return: The puppet module metadata.
        :rtype: dict
        """
        tmp_dir = mkdtemp(dir=os.path.dirname(module_path))
        try:
            with closing(tarfile.open(module_path)) as tarball:
                for member in tarball.getmembers():
                    path = member.name.split('/')
                    if path[-1] == constants.MODULE_METADATA_FILENAME:
                        tarball.extract(member, tmp_dir)
                        with open(os.path.join(tmp_dir, member.name)) as fp:
                            return json.load(fp)
        finally:
            shutil.rmtree(tmp_dir)

    def __init__(self, conduit, config):
        """
        :param conduit: Provides access to relevant Pulp functionality.
        :type conduit: pulp.plugins.conduits.unit_import.ImportUnitConduit
        :param config: Plugin configuration.
        :type config: pulp.plugins.config.PluginCallConfiguration
        """
        self.conduit = conduit
        self.config = config
        self.report = None
        self.canceled = False
        self.tmp_dir = None

    def feed_url(self):
        """
        Get the feed URL from the configuration and ensure it has a
        trailing '/' so urljoin will work correctly.
        :return: The feed URL.
        :rtype: str
        """
        url = self.config.get(constants.CONFIG_FEED)
        if not url.endswith('/'):
            url += '/'
        return url

    def cancel(self):
        """
        Cancel puppet module import.
        """
        self.canceled = True

    def _run(self, inventory):
        """
        Perform the synchronization using the supplied inventory.

        :param inventory: An inventory object.
        :type inventory: Inventory
        """
        manifest = self._fetch_manifest()
        if manifest is None:
            # fetch manifest failed
            return
        module_paths = self._fetch_modules(manifest)
        imported_modules = self._import_modules(inventory, module_paths)
        self._purge_unwanted_modules(inventory, imported_modules)

    def _download(self, urls):
        """
        Download files by URL.
        Encapsulates nectar details and provides a simplified method
        of downloading files.

        :param urls: A list of tuples: (url, destination).  The *url* and
            *destination* are both strings.  The *destination* is the fully
            qualified path to where the file is to be downloaded.
        :type urls: list
        :return: The nectar reports.  Tuple of: (succeeded_reports, failed_reports)
        :rtype: tuple
#.........这里部分代码省略.........

    def setUp(self):
        super(PuppetStatusRendererTests, self).setUp()
        self.renderer = PuppetStatusRenderer(self.context)

        self.sync_report = SyncProgressReport.from_progress_dict(IMPORTER_REPORT)
        self.publish_report = PublishProgressReport.from_progress_dict(DISTRIBUTOR_REPORT)

class PuppetModuleSyncRun(object):
    """
    Used to perform a single sync of a puppet repository. This class will
    maintain state relevant to the run and should not be reused across runs.
    """

    def __init__(self, repo, sync_conduit, config, is_cancelled_call):
        self.repo = repo
        self.sync_conduit = sync_conduit
        self.config = config
        self.is_cancelled_call = is_cancelled_call

        self.progress_report = SyncProgressReport(sync_conduit)

    def perform_sync(self):
        """
        Performs the sync operation according to the configured state of the
        instance. The report to be sent back to Pulp is returned from this
        call. This call will make calls into the conduit's progress update
        as appropriate.

        This call executes serially. No threads are created by this call. It
        will not return until either a step fails or the entire sync is
        completed.

        :return: the report object to return to Pulp from the sync call
        :rtype:  pulp.plugins.model.SyncReport
        """
        _LOG.info('Beginning sync for repository <%s>' % self.repo.id)

        try:
            metadata = self._parse_metadata()
            if not metadata:
                report = self.progress_report.build_final_report()
                return report

            self._import_modules(metadata)
        finally:
            # One final progress update before finishing
            self.progress_report.update_progress()

            report = self.progress_report.build_final_report()
            return report

    def _parse_metadata(self):
        """
        Takes the necessary actions (according to the run configuration) to
        retrieve and parse the repository's metadata. This call will return
        either the successfully parsed metadata or None if it could not
        be retrieved or parsed. The progress report will be updated with the
        appropriate description of what went wrong in the event of an error,
        so the caller should interpet a None return as an error occuring and
        not continue the sync.

        :return: object representation of the metadata
        :rtype:  RepositoryMetadata
        """
        _LOG.info('Beginning metadata retrieval for repository <%s>' % self.repo.id)

        self.progress_report.metadata_state = STATE_RUNNING
        self.progress_report.update_progress()

        start_time = datetime.now()

        # Retrieve the metadata from the source
        try:
            downloader = self._create_downloader()
            metadata_json_docs = downloader.retrieve_metadata(self.progress_report)
        except Exception, e:
            _LOG.exception('Exception while retrieving metadata for repository <%s>' % self.repo.id)
            self.progress_report.metadata_state = STATE_FAILED
            self.progress_report.metadata_error_message = _('Error downloading metadata')
            self.progress_report.metadata_exception = e
            self.progress_report.metadata_traceback = sys.exc_info()[2]

            end_time = datetime.now()
            duration = end_time - start_time
            self.progress_report.metadata_execution_time = duration.seconds

            self.progress_report.update_progress()

            return None

        # Parse the retrieved metadata documents
        try:
            metadata = RepositoryMetadata()
            for doc in metadata_json_docs:
                metadata.update_from_json(doc)
        except Exception, e:
            _LOG.exception('Exception parsing metadata for repository <%s>' % self.repo.id)
            self.progress_report.metadata_state = STATE_FAILED
            self.progress_report.metadata_error_message = _('Error parsing repository modules metadata document')
            self.progress_report.metadata_exception = e
            self.progress_report.metadata_traceback = sys.exc_info()[2]

            end_time = datetime.now()
            duration = end_time - start_time
            self.progress_report.metadata_execution_time = duration.seconds

            self.progress_report.update_progress()
#.........这里部分代码省略.........