pulp_smash.utils

Location: Pulp SmashAPI Documentationpulp_smash.utils

Utility functions for Pulp tests.

This module may make use of pulp_smash.api and pulp_smash.cli, but the reverse should not be done.

class pulp_smash.utils.BaseAPICrudTestCase(methodName='runTest')

A parent class for API CRUD test cases.

create_body() and update_body() should be overridden by concrete child classes. The bodies of these two methods are encoded to JSON and used as the bodies of HTTP requests for creating and updating a repository, respectively. Be careful to return appropriate data when overriding these methods: the various test* methods assume the repository is fairly simple.

Relevant Pulp documentation:

Create
http://docs.pulpproject.org/en/latest/dev-guide/integration/rest-api/repo/cud.html#create-a-repository
Read
http://docs.pulpproject.org/en/latest/dev-guide/integration/rest-api/repo/retrieval.html#retrieve-a-single-repository
Update
http://docs.pulpproject.org/en/latest/dev-guide/integration/rest-api/repo/cud.html#update-a-repository
Delete
http://docs.pulpproject.org/en/latest/dev-guide/integration/rest-api/repo/cud.html#delete-a-repository
static create_body()

Return a dict for creating a repository. Should be overridden.

Raises:NotImplementedError if not implemented by a child class.
classmethod setUpClass()

Create, update, read and delete a repository.

test_create()

Assert the created repository has all requested attributes.

Walk through each of the attributes returned by create_body() and verify the attribute is present in the repository.

NOTE: Any attribute whose name starts with importer or distributor is not verified.

test_importer_config()

Validate the config attribute of each importer.

test_importer_type_id()

Validate the repo importer’s importer_type_id attribute.

test_number_importers()

Assert the repository has one importer.

test_read()

Assert the repo update response has the requested changes.

test_status_codes()

Assert each response has a correct status code.

test_update()

Assert the repo update response has the requested changes.

static update_body()

Return a dict for updating a repository. Should be overridden.

Raises:NotImplementedError if not implemented by a child class.
class pulp_smash.utils.BaseAPITestCase(methodName='runTest')

A class with behaviour that is of use in many API test cases.

This test case provides set-up and tear-down behaviour that is common to many API test cases. It is not necessary to use this class as the parent of all API test cases, but it serves well in many cases.

classmethod setUpClass()

Provide a server config and an iterable of resources to delete.

The following class attributes are created this method:

cfg
A pulp_smash.config.PulpSmashConfig object.
resources
A set object. If a child class creates some resources that should be deleted when the test is complete, the child class should add that resource’s href to this set.
classmethod tearDownClass()

Delete all resources named by resources.

class pulp_smash.utils.DuplicateUploadsMixin

A mixin that adds tests for the “duplicate upload” test cases.

Consider the following procedure:

  1. Create a new feed-less repository of any content unit type.
  2. Upload a content unit into the repository.
  3. Upload the same content unit into the same repository.

The second upload should silently fail for all Pulp releases in the 2.x series. See:

This mixin adds tests for this case. Child classes should do the following:

  • Create a repository. Content units will be uploaded into this repository.
  • Create a class or instance attribute named upload_import_unit_args. It should be an iterable whose contents match the signature of upload_import_unit().
test_01_first_upload()

Upload a content unit to a repository.

test_02_second_upload()

Upload the same content unit to the same repository.

pulp_smash.utils.get_broker(server_config)

Build an object for managing the target system’s AMQP broker.

Talk to the host named by server_config and use simple heuristics to determine which AMQP broker is installed. If Qpid or RabbitMQ appear to be installed, return the name of that service. Otherwise, raise an exception.

Parameters:server_config (pulp_smash.config.PulpSmashConfig) – Information about the system on which an AMQP broker exists.
Returns:A string such as ‘qpidd’ or ‘rabbitmq’.
Raises:pulp_smash.exceptions.NoKnownBrokerError – If unable to find any AMQP brokers on the target system.
pulp_smash.utils.get_sha256_checksum(url)

Return the sha256 checksum of the file at the given URL.

When a URL is encountered for the first time, do the following:

  1. Download the file and calculate its sha256 checksum.
  2. Cache the URL-checksum pair.
  3. Return the checksum.

On subsequent calls, return a cached checksum.

pulp_smash.utils.get_unit_type_ids(server_config)

Tell which content unit types are supported by the target Pulp server.

Each Pulp plugin adds one (or more?) content unit types to Pulp, and each content unit type has a unique identifier. For example, the Python plugin [1] adds the Python content unit type [2], and Python content units have an ID of python_package. This function queries the server and returns those unit type IDs.

Parameters:server_config (pulp_smash.config.PulpSmashConfig) – Information about the Pulp deployment being targeted.
Returns:A set of content unit type IDs. For example: {'ostree', 'python_package'}.
[1]http://docs.pulpproject.org/plugins/pulp_python/
[2]http://docs.pulpproject.org/plugins/pulp_python/reference/python-type.html
pulp_smash.utils.http_get(url, **kwargs)

Issue a HTTP request to the url and return the response content.

This is useful for downloading file contents over HTTP[S].

Parameters:
  • url – the URL where the content should be get.
  • kwargs – additional kwargs to be passed to requests.get.
Returns:

the response content of a GET request to url.

pulp_smash.utils.publish_repo(cfg, repo, json=None)

Publish a repository.

Parameters:
  • cfg (pulp_smash.config.PulpSmashConfig) – Information about the Pulp host.
  • repo – A dict of detailed information about the repository to be published.
  • json – Data to be encoded as JSON and sent as the request body. Defaults to {'id': repo['distributors'][0]['id']}.
Raises:

ValueError when json is not passed, and repo does not have exactly one distributor.

Returns:

The server’s reponse. Call .json() on the response to get a call report.

pulp_smash.utils.pulp_admin_login(server_config)

Execute pulp-admin login.

Parameters:server_config (pulp_smash.config.PulpSmashConfig) – Information about the Pulp server being targeted.
Returns:The completed process.
Return type:pulp_smash.cli.CompletedProcess
pulp_smash.utils.reset_pulp(server_config)

Stop Pulp, reset its database, remove certain files, and start it.

Parameters:server_config (pulp_smash.config.PulpSmashConfig) – Information about the Pulp server being targeted.
Returns:Nothing.
pulp_smash.utils.reset_squid(cfg)

Stop Squid, reset its cache directory, and restart it.

Parameters:cfg (pulp_smash.config.PulpSmashConfig) – Information about a Pulp host.
Returns:Nothing.
pulp_smash.utils.search_units(cfg, repo, criteria=None, response_handler=None)

Find content units in a repo.

Parameters:
Returns:

Whatever is dictated by response_handler.

pulp_smash.utils.skip_if_type_is_unsupported(unit_type_id, server_config=None)

Raise SkipTest if support for the named type is not availalble.

Parameters:
Raises:

unittest.SkipTest if support is unavailable.

Returns:

Nothing.

pulp_smash.utils.sync_repo(cfg, repo)

Sync a repository.

Parameters:
Returns:

The server’s reponse. Call .json() on the response to get a call report.

pulp_smash.utils.upload_import_erratum(server_config, erratum, repo_href)

Upload an erratum to a Pulp server and import it into a repository.

For most content types, use upload_import_unit().

Parameters:
  • server_config (pulp_smash.config.PulpSmashConfig) – Information about the Pulp server being targeted.
  • erratum – A dict, with keys such as “id,” “status,” “issued,” and “references.”
  • repo_href – The path to the repository into which erratum will be imported.
Returns:

The call report returned when importing the erratum.

pulp_smash.utils.upload_import_unit(cfg, unit, import_params, repo)

Upload a content unit to a Pulp server and import it into a repository.

This procedure only works for some unit types, such as rpm or python_package. Others, like package_group, require an alternate procedure. The procedure encapsulated by this function is as follows:

  1. Create an upload request.
  2. Upload the content unit to Pulp, in small chunks.
  3. Import the uploaded content unit into a repository.
  4. Delete the upload request.

The default set of parameters sent to Pulp during step 3 are:

{'unit_key': {}, 'upload_id': '…'}

The actual parameters required by Pulp depending on the circumstances, and the parameters sent to Pulp may be customized via the import_params argument. For example, if uploading a Python content unit, import_params should be the following:

{'unit_key': {'filename': '…'}, 'unit_type_id': 'python_package'}

This would result in the following upload parameters being used:

{
    'unit_key': {'filename': '…'},
    'unit_type_id': 'python_package',
    'upload_id': '…',
}
Parameters:
  • cfg (pulp_smash.config.PulpSmashConfig) – Information about a Pulp host.
  • unit – The unit to be uploaded and imported, as a binary blob.
  • import_params – A dict of parameters to be merged into the default set of import parameters during step 3.
  • repo – A dict of information about the target repository.
Returns:

The call report returned when importing the unit.

pulp_smash.utils.uuid4()

Return a random UUID, as a unicode string.