~launchpad-pqm/launchpad/devel

Soyuz Upload Test
-----------------

This test will:

  * Upload packages
  * Import gpg key for katie
  * Register gpg key for katie
  * Register the katie user in the right team
  * Turn on the test keyserver
  * Include the non_free component in the database
  * Run process-upload.py
  * Check result
  * Mark packages as ACCEPTED
  * Runs process-accepted.py
  * Check results
  * Cleanup


Uploading Packages
------------------


First, let's create a temporary directory where we'll put
uploaded files in.

    >>> import os
    >>> import shutil
    >>> import tempfile
    >>> temp_dir = tempfile.mkdtemp()
    >>> incoming_dir = os.path.join(temp_dir, "incoming")
    >>> accepted_dir = os.path.join(temp_dir, "accepted")
    >>> rejected_dir = os.path.join(temp_dir, "rejected")
    >>> failed_dir = os.path.join(temp_dir, "failed")
    >>> os.mkdir(incoming_dir)


Now let's copy the files into separate upload directories, as if they'd
been uploaded over FTP.

    >>> from lp.services.config import config
    >>> from lp.archiveuploader.tagfiles import parse_tagfile
    >>> import glob
    >>> test_files_dir = os.path.join(config.root,
    ...                               "lib/lp/soyuz/scripts/"
    ...                               "tests/upload_test_files/")
    ...
    >>> changes = sorted(glob.glob(test_files_dir + "*.changes"))
    >>> sent_filenames = []
    >>> uploads = []
    >>> package_names = []

    >>> seq = 1
    >>> for changes_filepath in changes:
    ...     tf = parse_tagfile(changes_filepath)
    ...
    ...     if tf.has_key("Source"):
    ...         package_names.append(tf["Source"])
    ...
    ...     send_filepaths = [changes_filepath]
    ...     if tf.has_key("Files"):
    ...         send_filepaths.extend(
    ...             [os.path.join(test_files_dir, line.split()[-1])
    ...              for line in tf["Files"].splitlines() if line])
    ...
    ...     sent_filenames.extend(
    ...         os.path.basename(filepath) for filepath in send_filepaths)
    ...
    ...     upload_dir = os.path.join(
    ...         incoming_dir, 'upload-%06d' % seq, 'ubuntutest')
    ...     os.makedirs(upload_dir)
    ...
    ...     for filepath in send_filepaths:
    ...         shutil.copyfile(
    ...             filepath,
    ...             os.path.join(upload_dir, os.path.basename(filepath)))
    ...
    ...     uploads.append(send_filepaths)
    ...     seq += 1

Check that what we've just uploaded (everything in test_files_dir) is
what we were expecting to have uploaded.

    >>> package_names
    ['drdsl', 'etherwake']

At that point we must have a bunch of directories in the upload base
directory named upload-XXXXXX, as would result from several FTP
sessions.  Below we ensure that, and also that the content of these
files match the uploaded ones.

    >>> import hashlib
    >>> def get_md5(filename):
    ...     return hashlib.md5(open(filename).read()).digest()

    >>> def get_upload_dir(num, dir=incoming_dir):
    ...     """Return the path to the upload, if found in the dir."""
    ...     for upload_dir in os.listdir(dir):
    ...         if upload_dir.endswith("%06d" % num):
    ...             return os.path.join(dir, upload_dir)
    ...     return None

    >>> def find_upload_dir(num):
    ...     """Return a tuple (result, path) for the numbered upload."""
    ...     for name, dir in (("incoming", incoming_dir),
    ...         ("accepted", accepted_dir), ("rejected", rejected_dir),
    ...         ("failed", failed_dir)):
    ...         result = get_upload_dir(num, dir)
    ...         if result is not None:
    ...             return (name, result)
    ...     return (None, None)

    >>> def find_upload_dir_result(num):
    ...     """Return the result for the numbered upload."""
    ...     return find_upload_dir(num)[0]

    >>> def find_upload_dir_path(num):
    ...     """Return the path of the numbered upload."""
    ...     return find_upload_dir(num)[1]

    >>> for i, sent_filenames in enumerate(uploads):
    ...     upload_dir = get_upload_dir(i + 1)
    ...     distro_upload_dir = os.path.join(upload_dir, 'ubuntutest')
    ...     assert len(os.listdir(distro_upload_dir)) == len(sent_filenames)
    ...     for filename in sent_filenames:
    ...         upload_filename = os.path.join(distro_upload_dir,
    ...                                        os.path.basename(filename))
    ...         assert os.path.isfile(upload_filename)
    ...         assert get_md5(filename) == get_md5(upload_filename)

Finally, we'll just create an entirely empty upload folder. We rely for
our tests on a poppy-like naming system, ie. that the upload folder
end with 000004 (being our fourth upload).

    >>> os.mkdir("%s/fake_upload_000004" % incoming_dir)


Processing Uploads
------------------

Before asking the system to process the upload, we must prepare the
database to receive it. This consists mainly of adding the katie
user, since that's the email used in the Changed-By field for the
.changes files we are going to process, and the ftpmaster@canonical.com
GPG key, since that's the one used to sign the .changes file.

We don't have to check the .dsc file, since we're using the 'sync'
policy in process-upload.py.

# XXX: gustavo 2005-12-10
#     It might be interesting to move these entries into the sample data
#     rather than leaving it here. On the other hand, it's nice to have
#     it here as we have a good reference of what the uploading
#     procedure depends upon.

So, load the GPG key:

    >>> from zope.component import getUtility
    >>> from lp.services.gpg.interfaces import IGPGHandler
    >>> from lp.testing.gpgkeys import gpgkeysdir
    >>> gpg_handler = getUtility(IGPGHandler)
    >>> key_path = os.path.join(gpgkeysdir, 'ftpmaster@canonical.com.pub')
    >>> key_data = open(key_path).read()
    >>> key = gpg_handler.importPublicKey(key_data)
    >>> assert key is not None
    >>> print key.fingerprint
    33C0A61893A5DC5EB325B29E415A12CAC2F30234


Create the katie user and register it in a team that is allowed to
do uploads:

    >>> from lp.services.identity.interfaces.emailaddress import (
    ...     IEmailAddressSet)
    >>> from lp.registry.interfaces.gpg import (
    ...     GPGKeyAlgorithm,
    ...     IGPGKeySet,
    ...     )
    >>> from lp.registry.interfaces.person import (
    ...     IPersonSet,
    ...     PersonCreationRationale,
    ...     )
    >>> name, address = "Katie", "katie@rockhopper.ubuntu.com"
    >>> user = getUtility(IPersonSet).ensurePerson(
    ...     address, name, PersonCreationRationale.OWNER_CREATED_LAUNCHPAD)
    >>> assert user is not None
    >>> email = getUtility(IEmailAddressSet).getByEmail(address)
    >>> user.validateAndEnsurePreferredEmail(email)

    >>> uploader_team = getUtility(IPersonSet).getByName("ubuntu-team")
    >>> assert uploader_team is not None

    >>> login("foo.bar@canonical.com")
    >>> unused = uploader_team.addMember(
    ...     user, reviewer=uploader_team.teamowner)
    >>> login("test@canonical.com")


Assign the loaded GPG key to the katie user.

    >>> key_set = getUtility(IGPGKeySet)
    >>> user_key = key_set.new(
    ...     ownerID=user.id, keyid=key.keyid, fingerprint=key.fingerprint,
    ...     algorithm=GPGKeyAlgorithm.items[key.algorithm],
    ...       keysize=key.keysize, can_encrypt=key.can_encrypt, active=True)


Now we want to turn on the test key server to provide the key we
just imported. Remember that process-upload.py is running as
a different process.

    >>> from lp.testing.keyserver import KeyServerTac
    >>> keyserver = KeyServerTac()
    >>> keyserver.setUp()


Include non-free in the database. This will be done by the
NascentUpload in the 'sync' policy in the future.

    >>> from lp.soyuz.interfaces.component import IComponentSet
    >>> component_set = getUtility(IComponentSet)
    >>> non_free = component_set.new("non-free")
    >>> contrib = component_set.new("contrib")
    >>> import transaction
    >>> transaction.commit()

Now we are ready to process the uploaded packages.
This is done by running process-upload.py on each upload directory.

    >>> import subprocess, sys
    >>> script = os.path.join(config.root, "scripts/process-upload.py")

First, we will test process-upload's -J option, which limits which uploads
should be processed. We'll do this by locating and uploading initially
just upload number 1.

    >>> upload_dir_1_path = get_upload_dir(1)
    >>> upload_dir_1_name = os.path.basename(upload_dir_1_path)
    >>> process = subprocess.Popen([
    ...     sys.executable, script, "--no-mails", "-vv",
    ...     "-C", "sync", "-J", upload_dir_1_name, temp_dir,
    ...     ],
    ...     stdout=subprocess.PIPE,
    ...     stderr=subprocess.PIPE)
    >>> stdout, stderr = process.communicate()
    >>> process.returncode
    0

Check the four uploads are all where we expect - number 1 in rejected,
the other three still in incoming.

    >>> for i in range(4):
    ...     find_upload_dir_result(i + 1)
    'rejected'
    'incoming'
    'incoming'
    'incoming'


Now continue with the real upload.

    >>> process = subprocess.Popen([
    ...     sys.executable, script, "--no-mails", "-vv",
    ...     "-C", "sync", temp_dir,
    ...     ],
    ...     stdout=subprocess.PIPE,
    ...     stderr=subprocess.PIPE)

    >>> stdout, stderr = process.communicate()
    >>> if process.returncode != 0:
    ...     print stdout
    ...     print stderr


Let's check if packages were uploaded correctly.

    >>> from lp.registry.model.sourcepackagename import SourcePackageName
    >>> from lp.soyuz.model.sourcepackagerelease import SourcePackageRelease
    >>> from pprint import pprint
    >>> spn = SourcePackageName.selectOneBy(name="drdsl")
    >>> spn.name
    u'drdsl'
    >>> spr = SourcePackageRelease.selectOneBy(sourcepackagenameID=spn.id)
    >>> spr.title
    u'drdsl - 1.2.0-0ubuntu1'
    >>> spr.name
    u'drdsl'
    >>> spr.version
    u'1.2.0-0ubuntu1'
    >>> spr.component.name
    u'multiverse'
    >>> spr.section.name
    u'comm'
    >>> spr.maintainer.displayname
    u'Matthias Klose'
    >>> pprint(sorted([sprf.libraryfile.filename for sprf in spr.files]))
    [u'drdsl_1.2.0-0ubuntu1.diff.gz',
     u'drdsl_1.2.0-0ubuntu1.dsc',
     u'drdsl_1.2.0.orig.tar.gz']
    >>> spr.format.name
    'DPKG'
    >>> spr.urgency.name
    'LOW'
    >>> spr.upload_distroseries.name
    u'breezy-autotest'


Same thing for etherwake:

    >>> spn = SourcePackageName.selectOneBy(name="etherwake")
    >>> spn.name
    u'etherwake'
    >>> spr = SourcePackageRelease.selectOneBy(sourcepackagenameID=spn.id)
    >>> spr.title
    u'etherwake - 1.08-1'
    >>> spr.name
    u'etherwake'
    >>> spr.version
    u'1.08-1'
    >>> spr.component.name
    u'universe'
    >>> spr.section.name
    u'net'
    >>> spr.maintainer.displayname
    u'Alain Schroeder'
    >>> pprint(sorted([sprf.libraryfile.filename for sprf in spr.files]))
    [u'etherwake_1.08-1.diff.gz',
     u'etherwake_1.08-1.dsc',
     u'etherwake_1.08.orig.tar.gz']
    >>> spr.format.name
    'DPKG'
    >>> spr.urgency.name
    'LOW'
    >>> spr.upload_distroseries.name
    u'breezy-autotest'


Check the four uploads all ended up where we expected.

    >>> for i in range(0, 4):
    ...     find_upload_dir_result(i + 1)
    'rejected'
    'failed'

Also check the upload folders contain all the files we uploaded.

# XXX cprov 2006-12-06: hardcoded 'ubuntutest' directory is a hack see
# above around line 313.

    >>> for i, sent_filenames in enumerate(uploads):
    ...     upload_dir = find_upload_dir_path(i + 1)
    ...     if upload_dir is None:
    ...         continue
    ...     distro_upload_dir = os.path.join(upload_dir, 'ubuntutest')
    ...     assert len(os.listdir(distro_upload_dir)) == len(sent_filenames)
    ...     for filename in sent_filenames:
    ...         upload_filename = os.path.join(distro_upload_dir,
    ...                                        os.path.basename(filename))
    ...         assert os.path.isfile(upload_filename)
    ...         assert get_md5(filename) == get_md5(upload_filename)


Now let's see if all of the valid uploads are in the Upload queue marked
as NEW and RELEASE.

    >>> from lp.soyuz.model.queue import PackageUploadSource
    >>> for name in package_names:
    ...     print name
    ...     spn = SourcePackageName.selectOneBy(name=name)
    ...     spr = SourcePackageRelease.selectOneBy(sourcepackagenameID=spn.id)
    ...     us = PackageUploadSource.selectOneBy(
    ...         sourcepackagereleaseID=spr.id)
    ...     assert us.packageupload.status.name == 'NEW'
    ...     assert us.packageupload.pocket.name == 'RELEASE'
    drdsl
    etherwake


Processing NEW Items
----------------------

The processing of NEW-queue-entries checks the integrity of uploads
candidates and promote them to ACCEPTED, the failures are kept
as NEW

    >>> from lp.registry.interfaces.distribution import IDistributionSet
    >>> from lp.soyuz.enums import PackageUploadStatus
    >>> from lp.soyuz.interfaces.queue import (
    ...     QueueInconsistentStateError)

Since we landed correct security adapters for Upload,
we need to perform further actions logged in as an admins, which have
launchpad.Edit on the records:

    >>> from lp.testing import login
    >>> login("foo.bar@canonical.com")

    >>> distro = getUtility(IDistributionSet).getByName('ubuntutest')
    >>> series = distro['breezy-autotest']

We use getPackageUploads to inspect the current NEW queue and accept items.

    >>> queue_items = series.getPackageUploads(status=PackageUploadStatus.NEW)
    >>> L = []
    >>> for queue_item in queue_items:
    ...      try:
    ...          queue_item.setAccepted()
    ...      except QueueInconsistentStateError, e:
    ...          L.append("%s %s" % (queue_item.sourcepackagerelease.name, e))
    ...      else:
    ...          L.append("%s %s" % (queue_item.sourcepackagerelease.name,
    ...                              'ACCEPTED'))
    >>> L.sort()
    >>> print "\n".join(L)
    drdsl ACCEPTED
    etherwake ACCEPTED

Now we process the accepted queue items, one more time.

    >>> transaction.commit()
    >>> script = os.path.join(config.root, "scripts", "process-accepted.py")
    >>> process = subprocess.Popen(
    ...     [sys.executable, script, "ubuntutest", "-q"])
    >>> process.wait()
    0

These packages must now be in the publishing history. Let's check it.

    >>> from lp.soyuz.model.publishing import (
    ...    SourcePackagePublishingHistory as SSPPH)
    >>> package_names.sort()
    >>> for name in package_names:
    ...     spn = SourcePackageName.selectOneBy(name=name)
    ...     spr = SourcePackageRelease.selectOneBy(sourcepackagenameID=spn.id)
    ...     sspph = SSPPH.selectOneBy(sourcepackagereleaseID=spr.id)
    ...     if sspph:
    ...         print name, sspph.status.title
    ...     else:
    ...         print name, 'not Published'
    drdsl Pending
    etherwake Pending


Invoke Publisher script against the 'ubuntutest' distribution:

    >>> script = os.path.join(config.root, "scripts", "publish-distro.py")
    >>> process = subprocess.Popen([sys.executable, script, "-vvCq",
    ...                             "-d", "ubuntutest"],
    ...                            stdout=subprocess.PIPE,
    ...                            stderr=subprocess.PIPE)
    >>> stdout, stderr = process.communicate()
    >>> print stdout
    <BLANKLINE>
    >>> print stderr
    DEBUG Enabled by DEFAULT section
    INFO  Creating lockfile: ...
    DEBUG     Distribution: ubuntutest
    ...
    DEBUG   Added /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08.orig.tar.gz from library
    DEBUG   Added /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08-1.diff.gz from library
    DEBUG   Added /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08-1.dsc from library
    ...

    >>> transaction.commit()

Check if the 'etherwake' source package was correctly published and is
in the filesystem archive, we are looking for the DSC, the gzipped
original source and the gzipped package diff:

    >>> len(os.listdir(
    ...     "/var/tmp/archive/ubuntutest/pool/universe/e/etherwake"))
    3

Check the generation of a correct Sources tag file for the main
component of ubuntutest/breezy-autotest, containing the only the
required entry for 'etherwake':

    >>> sources = open(
    ...    "/var/tmp/archive/ubuntutest/dists/breezy-autotest/universe/source"
    ...    "/Sources").read()
    >>> import re
    >>> sources = re.subn(r'(?sm)^Checksums-.*?(?=^[^ ])', '', sources)[0]
    >>> print sources + '\nEND'
    Package: etherwake
    Binary: etherwake
    Version: 1.08-1
    Section: universe/net
    Maintainer: Alain Schroeder <...@...org>
    Build-Depends: debhelper (>> 2.0)
    Architecture: any
    Standards-Version: 3.5.10.0
    Format: 1.0
    Directory: pool/universe/e/etherwake
    Files:
     f13711c5b8261fbb77b43ae0e8ba9360 566 etherwake_1.08-1.dsc
     c2dc10f98bac012b900fd0b46721fc80 4455 etherwake_1.08.orig.tar.gz
     95c1e89e3ad7bc8740793bdf7aeb7334 4145 etherwake_1.08-1.diff.gz
    <BLANKLINE>
    <BLANKLINE>
    END

# XXX: maxb 2010-04-15 bug=563503
# The regex munging above can be removed once the tests no longer need to pass
# on Karmic and earlier.

Now we invoke changeOverride on just published etherwake, moving it to
component 'multiverse'.

    >>> ubuntutest = getUtility(IDistributionSet)['ubuntutest']
    >>> breezy_autotest = ubuntutest['breezy-autotest']
    >>> etherwake = breezy_autotest.getSourcePackage('etherwake')
    >>> etherwake_drspr = etherwake.currentrelease
    >>> override = etherwake_drspr.current_published.changeOverride(
    ...     new_component=getUtility(IComponentSet)['multiverse'])

Check if we have new pending publishing record as expected

    >>> for pub in SSPPH.selectBy(
    ...    sourcepackagereleaseID=etherwake_drspr.sourcepackagerelease.id,
    ...    orderBy=['id']):
    ...    print pub.status.name, pub.component.name, pub.pocket.name
    PUBLISHED universe RELEASE
    PENDING multiverse RELEASE

Force database changes, so they can be used by the external script properly.

    >>> transaction.commit()

Invoke Publisher script again to land our changes in the archive

    >>> script = os.path.join(config.root, "scripts", "publish-distro.py")
    >>> process = subprocess.Popen([sys.executable, script, "-vvCq",
    ...                             "-d", "ubuntutest"],
    ...                            stdout=subprocess.PIPE,
    ...                            stderr=subprocess.PIPE)
    >>> stdout, stderr = process.communicate()
    >>> process.returncode
    0

Check careful publishing took place, as requested with -C. In careful
publishing mode, publish-distro will attempt to publish files which are
already marked as published in the database and, if the files are
already on disk, verify the contents are as expected.

    >>> print stderr
    DEBUG Enabled by DEFAULT section
    INFO  Creating lockfile: ...
    DEBUG     Distribution: ubuntutest
    ...
    DEBUG   /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08.orig.tar.gz is already in pool with the same content.
    DEBUG   /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08-1.diff.gz is already in pool with the same content.
    DEBUG   /var/tmp/archive/ubuntutest/pool/universe/e/etherwake/etherwake_1.08-1.dsc is already in pool with the same content.
    ...

Check the publishing history again

    >>> for pub in SSPPH.selectBy(
    ...    sourcepackagereleaseID=etherwake_drspr.sourcepackagerelease.id,
    ...    orderBy=['id']):
    ...    print pub.status.name, pub.component.name, pub.pocket.name
    SUPERSEDED universe RELEASE
    PUBLISHED multiverse RELEASE

Check if the package was moved properly to the component 'multiverse':

    >>> main_sources = open(
    ...     "/var/tmp/archive/ubuntutest/dists/breezy-autotest"
    ...     "/main/source/Sources").read()
    >>> print main_sources + '\nEND'
    <BLANKLINE>
    END

    >>> multiverse_sources = open(
    ...     "/var/tmp/archive/ubuntutest/dists/breezy-autotest"
    ...     "/multiverse/source/Sources").read()
    >>> print multiverse_sources + '\nEND'
    Package: drdsl
    ...
    Package: etherwake
    ...
    END

Release File
------------

The publish-distro.py script will write an appropriate Release file
containing the suite in question and a list of checksums (MD5, SHA1
and SHA256) for each index published.

# XXX cprov 2006-12-13: trailing space on Architectures is a side-effect
# caused by the absence of published binaries in this suite. It should
# no happen in real conditions.

    >>> releasefile_contents = open("/var/tmp/archive/ubuntutest/dists/"
    ...                             "breezy-autotest/Release").read()
    >>> print releasefile_contents + '\nEND' #doctest: -NORMALIZE_WHITESPACE
    Origin: ubuntutest
    Label: ubuntutest
    Suite: breezy-autotest
    Version: 6.6.6
    Codename: breezy-autotest
    Date: ...
    Architectures:
    Components: main restricted universe multiverse
    Description: ubuntutest Breezy Badger Autotest 6.6.6
    MD5Sum:
     a5e5742a193740f17705c998206e18b6              114 main/source/Release
    ...
    SHA1:
     6222b7e616bcc20a32ec227254ad9de8d4bd5557              114 main/source/Release
    ...
    SHA256:
     297125e9b0f5da85552691597c9c4920aafd187e18a4e01d2ba70d8d106a6338              114 main/source/Release
    ...
    END


Nice! That's enough for now.. let's kill the process and clean
everything up.

    >>> import shutil
    >>> shutil.rmtree(temp_dir)

Remove the test archive from filesystem.

    >>> shutil.rmtree("/var/tmp/archive/")
    >>> keyserver.tearDown()


Feito! ;-)


vim:ft=doctest:ts=4:sw=4:et