def test_publisher(self):
     # setup
     units = self.populate()
     # test
     # publish
     repo_id = 'test_repo'
     base_url = 'file://'
     publish_dir = os.path.join(self.tmpdir, 'nodes/repos')
     repo_publish_dir = os.path.join(publish_dir, repo_id)
     virtual_host = (publish_dir, publish_dir)
     with HttpPublisher(base_url, virtual_host, repo_id, repo_publish_dir) as p:
         p.publish(units)
         p.commit()
     # verify
     conf = DownloaderConfig()
     downloader = LocalFileDownloader(conf)
     manifest_path = p.manifest_path()
     working_dir = os.path.join(self.tmpdir, 'working_dir')
     os.makedirs(working_dir)
     url = pathlib.url_join(base_url, manifest_path)
     manifest = RemoteManifest(url, downloader, working_dir)
     manifest.fetch()
     manifest.fetch_units()
     self.assertTrue(manifest.has_valid_units())
     units = manifest.get_units()
     n = 0
     for unit, ref in units:
         self.assertEqual(
             manifest.publishing_details[constants.BASE_URL],
             pathlib.url_join(base_url, publish_dir, repo_id))
         if n == 0:  # TARBALL
             path = pathlib.join(publish_dir, repo_id, unit[constants.TARBALL_PATH])
             self.assertTrue(os.path.isfile(path))
         else:
             path = pathlib.join(publish_dir, repo_id, unit[constants.RELATIVE_PATH])
             self.assertTrue(os.path.islink(path))
             self.assertEqual(unit[constants.FILE_SIZE], os.path.getsize(path))
         if n == 0:  # TARBALL
             path = pathlib.join(publish_dir, repo_id, unit[constants.TARBALL_PATH])
             tb = tarfile.open(path)
             try:
                 files = sorted(tb.getnames())
             finally:
                 tb.close()
             self.assertEqual(len(files), self.NUM_TARED_FILES)
         else:
             path = pathlib.join(publish_dir, repo_id, unit[constants.RELATIVE_PATH])
             with open(path, 'rb') as fp:
                 unit_content = fp.read()
                 self.assertEqual(unit_content, unit_content)
         self.assertEqual(unit['unit_key']['n'], n)
         n += 1

 def _url_and_destination(self, base_url, unit):
     """
     Get the download URL and download destination.
     :param base_url: The base URL.
     :type base_url: str
     :param unit: A content unit.
     :type unit: dict
     :return: (url, destination)
     :rtype: tuple(2)
     """
     storage_path = unit[constants.STORAGE_PATH]
     tar_path = unit.get(constants.TARBALL_PATH)
     if not tar_path:
         # The pulp/nodes/content endpoint provides all content.
         # This replaced the publishing of individual links for each unit.
         parsed = urlparse(base_url)
         relative_path = unit[constants.RELATIVE_PATH]
         path = pathlib.join(constants.CONTENT_PATH, pathlib.quote(relative_path))
         base_url = ParseResult(
             scheme=parsed.scheme,
             netloc=parsed.netloc,
             path=path,
             params=parsed.params,
             query=parsed.query,
             fragment=parsed.fragment)
         return base_url.geturl(), storage_path
     else:
         return pathlib.url_join(base_url, pathlib.quote(tar_path)),\
             pathlib.join(os.path.dirname(storage_path), os.path.basename(tar_path))

 def test_import_cached_manifest_missing_units(self, *unused):
     # Setup
     self.populate()
     with mock_config.patch({'server': {'storage_dir': self.parentfs}}):
         dist = NodesHttpDistributor()
         working_dir = os.path.join(self.childfs, 'working_dir')
         os.makedirs(working_dir)
         repo = Repository(self.REPO_ID, working_dir)
         configuration = self.dist_conf()
         conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
         dist.publish_repo(repo, conduit, configuration)
         model.Distributor.objects.delete()
         RepoContentUnit.get_collection().remove()
         unit_db.clean()
         self.define_plugins()
         publisher = dist.publisher(repo, configuration)
         manifest_path = publisher.manifest_path()
         manifest = Manifest(manifest_path)
         manifest.read()
         shutil.copy(manifest_path, os.path.join(working_dir, MANIFEST_FILE_NAME))
         # Test
         importer = NodesHttpImporter()
         manifest_url = pathlib.url_join(publisher.base_url, manifest_path)
         configuration = {
             constants.MANIFEST_URL_KEYWORD: manifest_url,
             constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
         }
         configuration = PluginCallConfiguration(configuration, {})
         conduit = RepoSyncConduit(self.REPO_ID, constants.HTTP_IMPORTER, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             importer.sync_repo(repo, conduit, configuration)
         # Verify
         units = conduit.get_units()
         self.assertEquals(len(units), self.NUM_UNITS)

    def test_publish(self, mock_repo_ctrl):
        # Setup
        self.populate()

        with mock_config.patch({'server': {'storage_dir': self.parentfs}}):
            # Test
            dist = NodesHttpDistributor()
            repo = Repository(self.REPO_ID)
            conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
            dist.publish_repo(repo, conduit, self.dist_conf())
            # Verify
            conf = DownloaderConfig()
            downloader = LocalFileDownloader(conf)
            pub = dist.publisher(repo, self.dist_conf())
            url = pathlib.url_join(pub.base_url, pub.manifest_path())
            working_dir = self.childfs
            manifest = RemoteManifest(url, downloader, working_dir)
            manifest.fetch()
            manifest.fetch_units()
            units = [u for u, r in manifest.get_units()]
            self.assertEqual(len(units), self.NUM_UNITS)
            for n in range(0, self.NUM_UNITS):
                unit = units[n]
                created = self.units[n]
                for p, v in unit['unit_key'].items():
                    self.assertEqual(created[p], v)
                for p, v in unit['metadata'].items():
                    if p in ('_ns', '_content_type_id'):
                        continue
                    self.assertEqual(created[p], v)
                self.assertEqual(created.get('_storage_path'), unit['storage_path'])
                self.assertEqual(unit['type_id'], self.UNIT_TYPE_ID)

 def test_publisher(self):
     # setup
     units = self.populate()
     # test
     # publish
     repo_id = 'test_repo'
     base_url = 'file://'
     publish_dir = os.path.join(self.tmpdir, 'nodes/repos')
     virtual_host = (publish_dir, publish_dir)
     with HttpPublisher(base_url, virtual_host, repo_id) as p:
         p.publish(units)
         p.commit()
     # verify
     conf = DownloaderConfig()
     downloader = HTTPSCurlDownloader(conf)
     manifest_path = p.manifest_path()
     working_dir = os.path.join(self.tmpdir, 'working_dir')
     os.makedirs(working_dir)
     manifest = Manifest()
     url = pathlib.url_join(base_url, manifest_path)
     manifest.fetch(url, working_dir, downloader)
     manifest.fetch_units(url, downloader)
     units = manifest.get_units()
     n = 0
     for unit, ref in units:
         if n == 0:
             self.assertTrue(unit[constants.PUBLISHED_AS_TARBALL])
         else:
             self.assertFalse(unit.get(constants.PUBLISHED_AS_TARBALL, False))
         path = pathlib.join(publish_dir, repo_id, unit[constants.RELATIVE_PATH])
         self.assertEqual(
             manifest.publishing_details[constants.BASE_URL],
             pathlib.url_join(base_url, publish_dir, repo_id))
         if n == 0:
             self.assertTrue(os.path.isfile(path))
         else:
             self.assertTrue(os.path.islink(path))
         if n == 0:
             with tarfile.open(path) as tb:
                 files = sorted(tb.getnames())
             self.assertEqual(len(files), self.NUM_TARED_FILES + 1)
         else:
             with open(path, 'rb') as fp:
                 unit_content = fp.read()
                 self.assertEqual(unit_content, unit_content)
         self.assertEqual(unit['unit_key']['n'], n)
         n += 1

 def test_import_modified_units(self, mock_get_working, *mocks):
     # Setup
     self.populate()
     mock_get_working.return_value = self.temp_dir
     max_concurrency = 5
     max_bandwidth = 12345
     with mock_config.patch({'server': {'storage_dir': self.parentfs}}):
         dist = NodesHttpDistributor()
         working_dir = os.path.join(self.childfs, 'working_dir')
         os.makedirs(working_dir)
         repo = Repository(self.REPO_ID, working_dir)
         cfg = self.dist_conf()
         conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
         dist.publish_repo(repo, conduit, cfg)
         # make the published unit have a newer _last_updated.
         collection = connection.get_collection(unit_db.unit_collection_name(self.UNIT_TYPE_ID))
         # N=0 (no file)
         unit = collection.find_one({'N': 0})
         unit['age'] = 84  # this will be updated back to 42.
         unit['_last_updated'] -= 1
         unit['_storage_path'] = None
         collection.update({'N': 0}, unit)
         # N=1
         unit = collection.find_one({'N': 1})
         unit['age'] = 85   # this will be updated back to 42.
         unit['_last_updated'] -= 1
         collection.update({'N': 1}, unit)
         # Test
         importer = NodesHttpImporter()
         publisher = dist.publisher(repo, cfg)
         manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
         configuration = {
             constants.MANIFEST_URL_KEYWORD: manifest_url,
             constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
             importer_constants.KEY_MAX_DOWNLOADS: max_concurrency,
             importer_constants.KEY_MAX_SPEED: max_bandwidth,
         }
         configuration = PluginCallConfiguration(configuration, {})
         conduit = RepoSyncConduit(self.REPO_ID, constants.HTTP_IMPORTER, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             importer.sync_repo(repo, conduit, configuration)
         # Verify
         unit = collection.find_one({'N': 0})
         self.assertEqual(unit['age'], 42)
         unit = collection.find_one({'N': 1})
         self.assertEqual(unit['age'], 42)

 def test_import_unit_files_already_exist_size_mismatch(self, *mocks):
     # Setup
     self.populate()
     pulp_conf.set('server', 'storage_dir', self.parentfs)
     dist = NodesHttpDistributor()
     working_dir = os.path.join(self.childfs, 'working_dir')
     os.makedirs(working_dir)
     repo = Repository(self.REPO_ID, working_dir)
     cfg = self.dist_conf()
     conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
     dist.publish_repo(repo, conduit, cfg)
     Repo.get_collection().remove()
     RepoDistributor.get_collection().remove()
     RepoContentUnit.get_collection().remove()
     unit_db.clean()
     self.define_plugins()
     parent_content = os.path.join(self.parentfs, 'content')
     child_content = os.path.join(self.childfs, 'content')
     shutil.copytree(parent_content, child_content)
     for fn in os.listdir(child_content):
         path = os.path.join(child_content, fn)
         if os.path.isdir(path):
             continue
         with open(path, 'w') as fp:
             fp.truncate()
     # Test
     importer = NodesHttpImporter()
     publisher = dist.publisher(repo, cfg)
     manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
     configuration = {
         constants.MANIFEST_URL_KEYWORD: manifest_url,
         constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
     }
     configuration = PluginCallConfiguration(configuration, {})
     conduit = RepoSyncConduit(
         self.REPO_ID,
         constants.HTTP_IMPORTER,
         RepoContentUnit.OWNER_TYPE_IMPORTER,
         constants.HTTP_IMPORTER)
     pulp_conf.set('server', 'storage_dir', self.childfs)
     importer.sync_repo(repo, conduit, configuration)
     # Verify
     units = conduit.get_units()
     self.assertEquals(len(units), self.NUM_UNITS)
     mock_importer_config_to_nectar_config = mocks[0]
     mock_importer_config_to_nectar_config.assert_called_with(configuration.flatten())

 def publish(self, units):
     """
     Publish the specified units.
     Writes the units.json file and symlinks each of the
     files associated to the unit.storage_path.
     :param units: A list of units to publish.
     :type units: iterable
     :return: The path to the written manifest.
     :rtype: str
     """
     manifest_path = super(self.__class__, self).publish(units)
     manifest = Manifest()
     manifest.read(manifest_path)
     manifest.publishing_details[constants.BASE_URL] = \
         pathlib.url_join(self.base_url, self.alias[0], self.repo_id)
     manifest.write(manifest_path)
     return manifest_path

 def _importer_conf(self, repo, config, binding_config):
     """
     Build the nodes importer configuration.
     :param repo: A repo object.
     :type repo: pulp.plugins.model.Repository
     :param config: plugin configuration
     :type  config: pulp.plugins.config.PluginCallConfiguration
     :param binding_config: The configuration stored on the binding.
     :type binding_config: dict
     :return: The importer configuration.
     :rtype: dict
     """
     publisher = self.publisher(repo, config)
     manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
     strategy = binding_config.get(constants.STRATEGY_KEYWORD, constants.DEFAULT_STRATEGY)
     configuration = {constants.STRATEGY_KEYWORD: strategy, constants.MANIFEST_URL_KEYWORD: manifest_url}
     return configuration

 def test_import_cached_manifest_matched(self, mock_fetch, *unused):
     # Setup
     self.populate()
     pulp_conf.set('server', 'storage_dir', self.parentfs)
     dist = NodesHttpDistributor()
     working_dir = os.path.join(self.childfs, 'working_dir')
     os.makedirs(working_dir)
     repo = Repository(self.REPO_ID, working_dir)
     configuration = self.dist_conf()
     conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
     dist.publish_repo(repo, conduit, configuration)
     Repo.get_collection().remove()
     RepoDistributor.get_collection().remove()
     RepoContentUnit.get_collection().remove()
     unit_db.clean()
     self.define_plugins()
     publisher = dist.publisher(repo, configuration)
     manifest_path = publisher.manifest_path()
     units_path = os.path.join(os.path.dirname(manifest_path), UNITS_FILE_NAME)
     manifest = Manifest(manifest_path)
     manifest.read()
     shutil.copy(manifest_path, os.path.join(working_dir, MANIFEST_FILE_NAME))
     shutil.copy(units_path, os.path.join(working_dir, UNITS_FILE_NAME))
     # Test
     importer = NodesHttpImporter()
     manifest_url = pathlib.url_join(publisher.base_url, manifest_path)
     configuration = {
         constants.MANIFEST_URL_KEYWORD: manifest_url,
         constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
     }
     configuration = PluginCallConfiguration(configuration, {})
     conduit = RepoSyncConduit(
         self.REPO_ID,
         constants.HTTP_IMPORTER,
         RepoContentUnit.OWNER_TYPE_IMPORTER,
         constants.HTTP_IMPORTER)
     pulp_conf.set('server', 'storage_dir', self.childfs)
     importer.sync_repo(repo, conduit, configuration)
     # Verify
     units = conduit.get_units()
     self.assertEquals(len(units), self.NUM_UNITS)
     self.assertFalse(mock_fetch.called)

 def test_import(self, *mocks):
     # Setup
     self.populate()
     max_concurrency = 5
     max_bandwidth = 12345
     pulp_conf.set('server', 'storage_dir', self.parentfs)
     dist = NodesHttpDistributor()
     working_dir = os.path.join(self.childfs, 'working_dir')
     os.makedirs(working_dir)
     repo = Repository(self.REPO_ID, working_dir)
     cfg = self.dist_conf()
     conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
     dist.publish_repo(repo, conduit, cfg)
     Repo.get_collection().remove()
     RepoDistributor.get_collection().remove()
     RepoContentUnit.get_collection().remove()
     unit_db.clean()
     self.define_plugins()
     # Test
     importer = NodesHttpImporter()
     publisher = dist.publisher(repo, cfg)
     manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
     configuration = {
         constants.MANIFEST_URL_KEYWORD: manifest_url,
         constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
         importer_constants.KEY_MAX_DOWNLOADS: max_concurrency,
         importer_constants.KEY_MAX_SPEED: max_bandwidth,
     }
     configuration = PluginCallConfiguration(configuration, {})
     conduit = RepoSyncConduit(
         self.REPO_ID,
         constants.HTTP_IMPORTER,
         RepoContentUnit.OWNER_TYPE_IMPORTER,
         constants.HTTP_IMPORTER)
     pulp_conf.set('server', 'storage_dir', self.childfs)
     importer.sync_repo(repo, conduit, configuration)
     # Verify
     units = conduit.get_units()
     self.assertEquals(len(units), self.NUM_UNITS)
     mock_importer_config_to_nectar_config = mocks[0]
     mock_importer_config_to_nectar_config.assert_called_with(configuration.flatten())

 def test_import_modified_units(self, *mocks):
     # Setup
     self.populate()
     max_concurrency = 5
     max_bandwidth = 12345
     pulp_conf.set('server', 'storage_dir', self.parentfs)
     dist = NodesHttpDistributor()
     working_dir = os.path.join(self.childfs, 'working_dir')
     os.makedirs(working_dir)
     repo = Repository(self.REPO_ID, working_dir)
     cfg = self.dist_conf()
     conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
     dist.publish_repo(repo, conduit, cfg)
     # make the published unit have a newer _last_updated.
     collection = connection.get_collection(unit_db.unit_collection_name(self.UNIT_TYPE_ID))
     unit = collection.find_one({'N': 0})
     unit['age'] = 84
     unit['_last_updated'] -= 1
     collection.update({'N': 0}, unit, safe=True)
     # Test
     importer = NodesHttpImporter()
     publisher = dist.publisher(repo, cfg)
     manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
     configuration = {
         constants.MANIFEST_URL_KEYWORD: manifest_url,
         constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
         importer_constants.KEY_MAX_DOWNLOADS: max_concurrency,
         importer_constants.KEY_MAX_SPEED: max_bandwidth,
     }
     configuration = PluginCallConfiguration(configuration, {})
     conduit = RepoSyncConduit(
         self.REPO_ID,
         constants.HTTP_IMPORTER,
         RepoContentUnit.OWNER_TYPE_IMPORTER,
         constants.HTTP_IMPORTER)
     pulp_conf.set('server', 'storage_dir', self.childfs)
     importer.sync_repo(repo, conduit, configuration)
     # Verify
     unit = collection.find_one({'N': 0})
     self.assertEqual(unit['age'], 42)

 def _add_units(self, request, unit_inventory):
     """
     Determine the list of units contained in the parent inventory
     but are not contained in the child inventory and add them.
     For each unit, this is performed in the following steps:
       1. Download the file (if defined) associated with the unit.
       2. Add the unit to the child inventory.
       3. Associate the unit to the repository.
     The unit is added only:
       1. If no file is associated with unit.
       2. The file associated with the unit is successfully downloaded.
     For units with files, the unit is added to the inventory as part of the
     unit download manager callback.
     :param request: A synchronization request.
     :type request: SyncRequest
     :param unit_inventory: The inventory of both parent and child content units.
     :type unit_inventory: UnitInventory
     """
     download_list = []
     units = unit_inventory.units_on_parent_only()
     request.progress.begin_adding_units(len(units))
     listener = ContentDownloadListener(self, request)
     for unit, unit_ref in units:
         if request.cancelled():
             return
         self._reset_storage_path(unit)
         if not self._needs_download(unit):
             # unit has no file associated
             self.add_unit(request, unit_ref.fetch())
             continue
         unit_path, destination = self._path_and_destination(unit)
         unit_URL = pathlib.url_join(unit_inventory.base_URL, unit_path)
         _request = listener.create_request(unit_URL, destination, unit, unit_ref)
         download_list.append(_request)
     if request.cancelled():
         return
     container = ContentContainer()
     request.summary.sources = \
         container.download(request.cancel_event, request.downloader, download_list, listener)
     request.summary.errors.extend(listener.error_list)

 def _add_units(self, request, unit_inventory):
     """
     Determine the list of units contained in the parent inventory
     but are not contained in the child inventory and add them.
     For each unit, this is performed in the following steps:
       1. Download the file (if defined) associated with the unit.
       2. Add the unit to the child inventory.
       3. Associate the unit to the repository.
     The unit is added only:
       1. If no file is associated with unit.
       2. The file associated with the unit is successfully downloaded.
     For units with files, the unit is added to the inventory as part of the
     unit download manager callback.
     :param request: A synchronization request.
     :type request: SyncRequest
     :param unit_inventory: The inventory of both parent and child content units.
     :type unit_inventory: UnitInventory
     """
     download_list = []
     units = unit_inventory.units_on_parent_only()
     request.progress.begin_adding_units(len(units))
     manager = UnitDownloadManager(self, request)
     publishing_details = unit_inventory.manifest.publishing_details
     for unit, unit_ref in units:
         if request.cancelled():
             return
         self._update_storage_path(unit)
         if not self._needs_download(unit):
             # unit has no file associated
             self.add_unit(request, unit_ref.fetch())
             continue
         url = pathlib.url_join(publishing_details[constants.BASE_URL], pathlib.quote(unit[constants.RELATIVE_PATH]))
         storage_path = unit[constants.STORAGE_PATH]
         _request = manager.create_request(url, storage_path, unit_ref)
         download_list.append(_request)
     if request.cancelled():
         return
     request.downloader.event_listener = manager
     request.downloader.download(download_list)
     request.summary.errors.extend(manager.error_list())

 def test_import_unit_files_already_exist(self, mock_get_working, *mocks):
     # Setup
     self.populate()
     mock_get_working.return_value = self.temp_dir
     with mock_config.patch({'server': {'storage_dir': self.parentfs}}):
         dist = NodesHttpDistributor()
         working_dir = os.path.join(self.childfs, 'working_dir')
         os.makedirs(working_dir)
         repo = Repository(self.REPO_ID, working_dir)
         cfg = self.dist_conf()
         conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
         dist.publish_repo(repo, conduit, cfg)
         model.Distributor.objects.delete()
         RepoContentUnit.get_collection().remove()
         unit_db.clean()
         self.define_plugins()
         parent_content = os.path.join(self.parentfs, 'content')
         child_content = os.path.join(self.childfs, 'content')
         shutil.copytree(parent_content, child_content)
         # Test
         importer = NodesHttpImporter()
         publisher = dist.publisher(repo, cfg)
         manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
         configuration = {
             constants.MANIFEST_URL_KEYWORD: manifest_url,
             constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
         }
         configuration = PluginCallConfiguration(configuration, {})
         conduit = RepoSyncConduit(self.REPO_ID, constants.HTTP_IMPORTER, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             importer.sync_repo(repo, conduit, configuration)
         # Verify
         units = conduit.get_units()
         self.assertEquals(len(units), self.NUM_UNITS)
         mock_importer_config_to_nectar_config = mocks[0]
         mock_importer_config_to_nectar_config.assert_called_with(configuration.flatten())