launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #08221
[Merge] lp:~cjwatson/launchpad/refactor-customupload-tests into lp:launchpad
Colin Watson has proposed merging lp:~cjwatson/launchpad/refactor-customupload-tests into lp:launchpad.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/refactor-customupload-tests/+merge/107381
== Summary ==
I've been working on a branch which adds another custom upload format (not included here), and I really couldn't face writing doctests for it just so that they'd be similar to the existing custom upload tests. Therefore I cracked and decided to rewrite the custom upload doctests as unit tests. It won't do my LoC credit any harm either, which I may need.
The doctests were a mixture of fine-grained testing of the behaviour of the custom upload processors, albeit at one remove via queue processing, and high-level testing of upload and queue manipulation. The latter can be done in much the same way the doctests were doing it (although the doctest-style matching of debug output really doesn't add much of any interest), but the former is much better handled by unit tests that generate tarballs on the fly to exercise specific features. I put the former in archivepublisher near to the custom upload processors, and the latter in soyuz near to the queue processing.
== Tests ==
bin/test -vvct test_ddtp_tarball -t test_debian_installer -t test_dist_upgrader -t tarfile_helpers.txt -t test_distroseriesqueue
--
https://code.launchpad.net/~cjwatson/launchpad/refactor-customupload-tests/+merge/107381
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~cjwatson/launchpad/refactor-customupload-tests into lp:launchpad.
=== added file 'lib/lp/archivepublisher/tests/test_ddtp_tarball.py'
--- lib/lp/archivepublisher/tests/test_ddtp_tarball.py 1970-01-01 00:00:00 +0000
+++ lib/lp/archivepublisher/tests/test_ddtp_tarball.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,102 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test ddtp-tarball custom uploads.
+
+See also lp.soyuz.tests.test_distroseriesqueue_ddtp_tarball for high-level
+tests of ddtp-tarball upload and queue manipulation.
+"""
+
+import os
+
+from lp.archivepublisher.ddtp_tarball import process_ddtp_tarball
+from lp.services.tarfile_helpers import LaunchpadWriteTarFile
+from lp.testing import TestCase
+
+
+class TestDdtpTarball(TestCase):
+
+ def setUp(self):
+ super(TestDdtpTarball, self).setUp()
+ self.temp_dir = self.makeTemporaryDirectory()
+ self.suite = "distroseries"
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+
+ def openArchive(self, version):
+ self.path = os.path.join(
+ self.temp_dir, "translations_main_%s.tar.gz" % version)
+ self.buffer = open(self.path, "wb")
+ self.archive = LaunchpadWriteTarFile(self.buffer)
+
+ def process(self):
+ self.archive.close()
+ self.buffer.close()
+ process_ddtp_tarball(self.temp_dir, self.path, self.suite)
+
+ def getTranslationsPath(self, filename):
+ return os.path.join(
+ self.temp_dir, "dists", self.suite, "main", "i18n", filename)
+
+ def test_basic(self):
+ # Processing a simple correct tar file works.
+ self.openArchive("20060728")
+ self.archive.add_file("i18n/Translation-de", "")
+ self.process()
+ self.assertTrue(os.path.exists(
+ self.getTranslationsPath("Translation-de")))
+
+ def test_ignores_empty_directories(self):
+ # Empty directories in the tarball are not extracted.
+ self.openArchive("20060728")
+ self.archive.add_file("i18n/Translation-de", "")
+ self.archive.add_directory("i18n/foo")
+ self.process()
+ self.assertTrue(os.path.exists(
+ self.getTranslationsPath("Translation-de")))
+ self.assertFalse(os.path.exists(self.getTranslationsPath("foo")))
+
+ def test_partial_update(self):
+ # If a DDTP tarball only contains a subset of published translation
+ # files, these are updated and the rest are left untouched.
+ self.openArchive("20060728")
+ self.archive.add_file("i18n/Translation-bn", "bn")
+ self.archive.add_file("i18n/Translation-ca", "ca")
+ self.process()
+ with open(self.getTranslationsPath("Translation-bn")) as bn_file:
+ self.assertEqual("bn", bn_file.read())
+ with open(self.getTranslationsPath("Translation-ca")) as ca_file:
+ self.assertEqual("ca", ca_file.read())
+ self.openArchive("20060817")
+ self.archive.add_file("i18n/Translation-bn", "new bn")
+ self.process()
+ with open(self.getTranslationsPath("Translation-bn")) as bn_file:
+ self.assertEqual("new bn", bn_file.read())
+ with open(self.getTranslationsPath("Translation-ca")) as ca_file:
+ self.assertEqual("ca", ca_file.read())
+
+ def test_breaks_hard_links(self):
+ # Our archive uses dsync to replace identical files with hard links
+ # in order to save some space. tarfile.extract overwrites
+ # pre-existing files rather than creating new files and moving them
+ # into place, so making this work requires special care. Test that
+ # that care has been taken.
+ self.openArchive("20060728")
+ self.archive.add_file("i18n/Translation-ca", "")
+ self.process()
+ ca = self.getTranslationsPath("Translation-ca")
+ bn = self.getTranslationsPath("Translation-bn")
+ os.link(ca, bn)
+ self.assertTrue(os.path.exists(bn))
+ self.assertEqual(2, os.stat(bn).st_nlink)
+ self.assertEqual(2, os.stat(ca).st_nlink)
+ self.openArchive("20060817")
+ self.archive.add_file("i18n/Translation-bn", "break hard link")
+ self.process()
+ with open(bn) as bn_file:
+ self.assertEqual("break hard link", bn_file.read())
+ with open(ca) as ca_file:
+ self.assertEqual("", ca_file.read())
+ self.assertEqual(1, os.stat(bn).st_nlink)
+ self.assertEqual(1, os.stat(ca).st_nlink)
=== added file 'lib/lp/archivepublisher/tests/test_debian_installer.py'
--- lib/lp/archivepublisher/tests/test_debian_installer.py 1970-01-01 00:00:00 +0000
+++ lib/lp/archivepublisher/tests/test_debian_installer.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,147 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test debian-installer custom uploads.
+
+See also lp.soyuz.tests.test_distroseriesqueue_debian_installer for
+high-level tests of debian-installer upload and queue manipulation.
+"""
+
+import os
+
+from lp.archivepublisher.customupload import CustomUploadBadUmask
+from lp.archivepublisher.debian_installer import (
+ DebianInstallerAlreadyExists,
+ process_debian_installer,
+ )
+from lp.services.tarfile_helpers import LaunchpadWriteTarFile
+from lp.testing import TestCase
+
+
+class TestDebianInstaller(TestCase):
+
+ def setUp(self):
+ super(TestDebianInstaller, self).setUp()
+ self.temp_dir = self.makeTemporaryDirectory()
+ self.suite = "distroseries"
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+
+ def openArchive(self):
+ self.version = "20070214ubuntu1"
+ self.arch = "i386"
+ self.path = os.path.join(
+ self.temp_dir,
+ "debian-installer-images_%s_%s.tar.gz" % (self.version, self.arch))
+ self.buffer = open(self.path, "wb")
+ self.archive = LaunchpadWriteTarFile(self.buffer)
+
+ def addFile(self, path, contents):
+ self.archive.add_file(
+ "installer-%s/%s/%s" % (self.arch, self.version, path), contents)
+
+ def addSymlink(self, path, target):
+ self.archive.add_symlink(
+ "installer-%s/%s/%s" % (self.arch, self.version, path), target)
+
+ def process(self):
+ self.archive.close()
+ self.buffer.close()
+ process_debian_installer(self.temp_dir, self.path, self.suite)
+
+ def getInstallerPath(self, versioned_filename=None):
+ installer_path = os.path.join(
+ self.temp_dir, "dists", self.suite, "main",
+ "installer-%s" % self.arch)
+ if versioned_filename is not None:
+ installer_path = os.path.join(
+ installer_path, self.version, versioned_filename)
+ return installer_path
+
+ def test_basic(self):
+ # Processing a simple correct tar file succeeds.
+ self.openArchive()
+ self.addFile("hello", "world")
+ self.process()
+
+ def test_already_exists(self):
+ # If the target directory already exists, processing fails.
+ self.openArchive()
+ os.makedirs(self.getInstallerPath("."))
+ self.assertRaises(DebianInstallerAlreadyExists, self.process)
+
+ def test_bad_umask(self):
+ # The umask must be 022 to avoid incorrect permissions.
+ self.openArchive()
+ self.addFile("dir/file", "foo")
+ os.umask(002) # cleanup already handled by setUp
+ self.assertRaises(CustomUploadBadUmask, self.process)
+
+ def test_current_symlink(self):
+ # A "current" symlink is created to the last version.
+ self.openArchive()
+ self.addFile("hello", "world")
+ self.process()
+ installer_path = self.getInstallerPath()
+ self.assertContentEqual(
+ [self.version, "current"], os.listdir(installer_path))
+ self.assertEqual(
+ self.version, os.readlink(os.path.join(installer_path, "current")))
+
+ def test_correct_file(self):
+ # Files in the tarball are extracted correctly.
+ self.openArchive()
+ directory = ("images/netboot/ubuntu-installer/i386/"
+ "pxelinux.cfg.serial-9600")
+ filename = os.path.join(directory, "default")
+ long_filename = os.path.join(
+ directory, "very_very_very_very_very_very_long_filename")
+ self.addFile(filename, "hey")
+ self.addFile(long_filename, "long")
+ self.process()
+ with open(self.getInstallerPath(filename)) as f:
+ self.assertEqual("hey", f.read())
+ with open(self.getInstallerPath(long_filename)) as f:
+ self.assertEqual("long", f.read())
+
+ def test_correct_symlink(self):
+ # Symbolic links in the tarball are extracted correctly.
+ self.openArchive()
+ foo_path = "images/netboot/foo"
+ foo_target = "ubuntu-installer/i386/pxelinux.cfg.serial-9600/default"
+ link_to_dir_path = "images/netboot/link_to_dir"
+ link_to_dir_target = "ubuntu-installer/i386/pxelinux.cfg.serial-9600"
+ self.addSymlink(foo_path, foo_target)
+ self.addSymlink(link_to_dir_path, link_to_dir_target)
+ self.process()
+ self.assertEqual(
+ foo_target, os.readlink(self.getInstallerPath(foo_path)))
+ self.assertEqual(
+ link_to_dir_target,
+ os.path.normpath(os.readlink(
+ self.getInstallerPath(link_to_dir_path))))
+
+ def test_top_level_permissions(self):
+ # Top-level directories are set to mode 0755 (see bug 107068).
+ self.openArchive()
+ self.addFile("hello", "world")
+ self.process()
+ installer_path = self.getInstallerPath()
+ self.assertEqual(0755, os.stat(installer_path).st_mode & 0777)
+ self.assertEqual(
+ 0755,
+ os.stat(os.path.join(installer_path, os.pardir)).st_mode & 0777)
+
+ def test_extracted_permissions(self):
+ # Extracted files and directories are set to 0644/0755.
+ self.openArchive()
+ directory = ("images/netboot/ubuntu-installer/i386/"
+ "pxelinux.cfg.serial-9600")
+ filename = os.path.join(directory, "default")
+ self.addFile(filename, "hey")
+ self.process()
+ self.assertEqual(
+ 0644, os.stat(self.getInstallerPath(filename)).st_mode & 0777)
+ self.assertEqual(
+ 0755, os.stat(self.getInstallerPath(directory)).st_mode & 0777)
=== added file 'lib/lp/archivepublisher/tests/test_dist_upgrader.py'
--- lib/lp/archivepublisher/tests/test_dist_upgrader.py 1970-01-01 00:00:00 +0000
+++ lib/lp/archivepublisher/tests/test_dist_upgrader.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,86 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test dist-upgrader custom uploads.
+
+See also lp.soyuz.tests.test_distroseriesqueue_dist_upgrader for high-level
+tests of dist-upgrader upload and queue manipulation.
+"""
+
+import os
+
+from lp.archivepublisher.customupload import CustomUploadBadUmask
+from lp.archivepublisher.dist_upgrader import (
+ DistUpgraderAlreadyExists,
+ DistUpgraderBadVersion,
+ process_dist_upgrader,
+ )
+from lp.services.tarfile_helpers import LaunchpadWriteTarFile
+from lp.testing import TestCase
+
+
+class TestDistUpgrader(TestCase):
+
+ def setUp(self):
+ super(TestDistUpgrader, self).setUp()
+ self.temp_dir = self.makeTemporaryDirectory()
+ self.suite = "distroseries"
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+
+ def openArchive(self, version):
+ self.path = os.path.join(
+ self.temp_dir, "dist-upgrader_%s_all.tar.gz" % version)
+ self.buffer = open(self.path, "wb")
+ self.archive = LaunchpadWriteTarFile(self.buffer)
+
+ def process(self):
+ self.archive.close()
+ self.buffer.close()
+ process_dist_upgrader(self.temp_dir, self.path, self.suite)
+
+ def getUpgraderPath(self):
+ return os.path.join(
+ self.temp_dir, "dists", self.suite, "main", "dist-upgrader-all")
+
+ def test_basic(self):
+ # Processing a simple correct tar file works.
+ self.openArchive("20060302.0120")
+ self.archive.add_file("20060302.0120/hello", "world")
+ self.process()
+
+ def test_already_exists(self):
+ # If the target directory already exists, processing fails.
+ self.openArchive("20060302.0120")
+ self.archive.add_file("20060302.0120/hello", "world")
+ os.makedirs(os.path.join(self.getUpgraderPath(), "20060302.0120"))
+ self.assertRaises(DistUpgraderAlreadyExists, self.process)
+
+ def test_bad_umask(self):
+ # The umask must be 022 to avoid incorrect permissions.
+ self.openArchive("20060302.0120")
+ self.archive.add_file("20060302.0120/file", "foo")
+ os.umask(002) # cleanup already handled by setUp
+ self.assertRaises(CustomUploadBadUmask, self.process)
+
+ def test_current_symlink(self):
+ # A "current" symlink is created to the last version.
+ self.openArchive("20060302.0120")
+ self.archive.add_file("20060302.0120/hello", "world")
+ self.process()
+ upgrader_path = self.getUpgraderPath()
+ self.assertContentEqual(
+ ["20060302.0120", "current"], os.listdir(upgrader_path))
+ self.assertEqual(
+ "20060302.0120",
+ os.readlink(os.path.join(upgrader_path, "current")))
+ self.assertContentEqual(
+ ["hello"],
+ os.listdir(os.path.join(upgrader_path, "20060302.0120")))
+
+ def test_bad_version(self):
+ # Bad versions in the tarball are refused.
+ self.openArchive("20070219.1234")
+ self.archive.add_file("foobar/foobar/dapper.tar.gz", "")
+ self.assertRaises(DistUpgraderBadVersion, self.process)
=== removed file 'lib/lp/archiveuploader/tests/data/ddtp-tarball/translations-main_20060817_all.changes'
--- lib/lp/archiveuploader/tests/data/ddtp-tarball/translations-main_20060817_all.changes 2011-05-20 13:48:46 +0000
+++ lib/lp/archiveuploader/tests/data/ddtp-tarball/translations-main_20060817_all.changes 1970-01-01 00:00:00 +0000
@@ -1,27 +0,0 @@
------BEGIN PGP SIGNED MESSAGE-----
-Hash: SHA1
-
-Origin: Ubuntu/warty
-Format: 1.7
-Date: Thu, 17 Aug 2006 14:20:25 +0300
-Architecture: all
-Version: 20060817
-Distribution: breezy-autotest
-Source: ddtp-tarball
-Binary: ddtp-tarball
-Urgency: low
-Maintainer: Foo Bar <foo.bar@xxxxxxxxxxxxx>
-Changed-By: Foo Bar <foo.bar@xxxxxxxxxxxxx>
-Changes:
- .
- * Partial update, new language index
- .
-Files:
- a774352e50a4b4606a85d53b1cf1b49e 181 raw-ddtp-tarball - translations_main_20060817.tar.gz
------BEGIN PGP SIGNATURE-----
-Version: GnuPG v1.4.11 (GNU/Linux)
-
-iEYEARECAAYFAk3WbpIACgkQjn63CGxkqMVACgCfc3GXQwcDuSs8C+lopI+Igw2c
-w0cAn1YBCAxqTvipznXueOOWsphAMLCP
-=803N
------END PGP SIGNATURE-----
=== removed file 'lib/lp/archiveuploader/tests/data/ddtp-tarball/translations_main_20060817.tar.gz'
Binary files lib/lp/archiveuploader/tests/data/ddtp-tarball/translations_main_20060817.tar.gz 2007-03-27 21:01:50 +0000 and lib/lp/archiveuploader/tests/data/ddtp-tarball/translations_main_20060817.tar.gz 1970-01-01 00:00:00 +0000 differ
=== modified file 'lib/lp/services/doc/tarfile_helpers.txt'
--- lib/lp/services/doc/tarfile_helpers.txt 2010-12-17 14:06:07 +0000
+++ lib/lp/services/doc/tarfile_helpers.txt 2012-05-25 13:45:30 +0000
@@ -5,28 +5,35 @@
easy.
>>> from StringIO import StringIO
+ >>> import tarfile
>>> from lp.services.tarfile_helpers import LaunchpadWriteTarFile
First, we are going to define a function we are going to use to validate the
output we will get.
- >>> def examine_tarfile(tarfile):
- ... names = tarfile.getnames()
+ >>> def examine_tarfile(tar):
+ ... members = tar.getmembers()
... # Calculate the length of the longest name.
- ... max_length = max(len(name) for name in names)
+ ... max_length = max(len(member.name) for member in members)
... # Use this length to generate an appropriate format string.
... format = '%%-%ds | %%s' % max_length
...
- ... for name in names:
- ... file = tarfile.extractfile(name)
+ ... for member in members:
+ ... if member.type == tarfile.REGTYPE:
+ ... file = tar.extractfile(member)
...
- ... # XXX: 2010-04-26, Salgado, bug=570244: This is to make the
- ... # test pass on python2.5 and 2.6.
- ... name = name.rstrip('/')
- ... if file is not None:
- ... print format % (name, file.read())
- ... else:
- ... print format % (name, '')
+ ... # XXX: 2010-04-26, Salgado, bug=570244: This is to make
+ ... # the test pass on python2.5 and 2.6.
+ ... name = member.name.rstrip('/')
+ ... if file is not None:
+ ... print format % (name, file.read())
+ ... else:
+ ... print format % (name, '')
+ ... elif member.type == tarfile.SYMTYPE:
+ ... print format % (
+ ... member.name, "<link to %s>" % member.linkname)
+ ... elif member.type == tarfile.DIRTYPE:
+ ... print format % (member.name, "<directory>")
# Start off by creating a blank archive.
# We'll need a filehandle to store it in.
@@ -43,19 +50,28 @@
>>> archive.add_files({'bar': '2', 'baz': '3'})
+We can add symbolic links.
+
+ >>> archive.add_symlink('link', 'foo')
+
+We can add directories.
+
+ >>> archive.add_directory('dir')
+
Once we are done adding files, the archive needs to be closed.
>>> archive.close()
And now we can inspect the produced file.
- >>> import tarfile
>>> buffer.seek(0)
>>> archive = tarfile.open('', 'r', buffer)
>>> examine_tarfile(archive)
- foo | 1
- bar | 2
- baz | 3
+ foo | 1
+ bar | 2
+ baz | 3
+ link | <link to foo>
+ dir | <directory>
There are also some convenience methods for getting directly from several
files, represented with a dictionary, which have the file name as key and
@@ -99,9 +115,9 @@
... 'uno/dos/tres/cuatro': 'blah'
... })
>>> examine_tarfile(archive)
- uno |
- uno/dos |
- uno/dos/tres |
+ uno | <directory>
+ uno/dos | <directory>
+ uno/dos/tres | <directory>
uno/dos/tres/cuatro | blah
Also, if there is a duplicated file, last one is the one that remains there.
=== modified file 'lib/lp/services/tarfile_helpers.py'
--- lib/lp/services/tarfile_helpers.py 2010-12-17 14:06:07 +0000
+++ lib/lp/services/tarfile_helpers.py 2012-05-25 13:45:30 +0000
@@ -1,4 +1,4 @@
-# Copyright 2010 Canonical Ltd. This software is licensed under the
+# Copyright 2010-2012 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Helpers to work with tar files more easily."""
@@ -25,6 +25,7 @@
# any longer.
# -- Dafydd Harries, 2005-04-07.
+
class LaunchpadWriteTarFile:
"""Convenience wrapper around the tarfile module.
@@ -66,36 +67,46 @@
self.tarfile.close()
self.closed = True
- def add_file(self, path, contents):
- """Add a file to the archive."""
- assert not self.closed, "Can't add a file to a closed archive"
+ def _make_skeleton_tarinfo(self, path, now):
+ """Make a basic TarInfo object to be fleshed out by the caller."""
+ tarinfo = tarfile.TarInfo(path)
+ tarinfo.mtime = now
+ tarinfo.uname = 'launchpad'
+ tarinfo.gname = 'launchpad'
+ return tarinfo
- now = int(time.time())
+ def _ensure_directories(self, path, now):
+ """Ensure that all the directories in the path are present."""
path_bits = path.split(os.path.sep)
- # Ensure that all the directories in the path are present in the
- # archive.
for i in range(1, len(path_bits)):
joined_path = os.path.join(*path_bits[:i])
try:
self.tarfile.getmember(joined_path)
except KeyError:
- tarinfo = tarfile.TarInfo(joined_path)
+ tarinfo = self._make_skeleton_tarinfo(joined_path, now)
tarinfo.type = tarfile.DIRTYPE
- tarinfo.mtime = now
tarinfo.mode = 0755
- tarinfo.uname = 'launchpad'
- tarinfo.gname = 'launchpad'
self.tarfile.addfile(tarinfo)
- tarinfo = tarfile.TarInfo(path)
- tarinfo.time = now
- tarinfo.mtime = now
+ def add_directory(self, path):
+ """Add a directory to the archive."""
+ assert not self.closed, "Can't add a directory to a closed archive"
+
+ now = int(time.time())
+ self._ensure_directories(os.path.join(path, "."), now)
+
+ def add_file(self, path, contents):
+ """Add a file to the archive."""
+ assert not self.closed, "Can't add a file to a closed archive"
+
+ now = int(time.time())
+ self._ensure_directories(path, now)
+
+ tarinfo = self._make_skeleton_tarinfo(path, now)
tarinfo.mode = 0644
tarinfo.size = len(contents)
- tarinfo.uname = 'launchpad'
- tarinfo.gname = 'launchpad'
self.tarfile.addfile(tarinfo, StringIO(contents))
def add_files(self, files):
@@ -106,3 +117,15 @@
for filename in sorted(files.keys()):
self.add_file(filename, files[filename])
+
+ def add_symlink(self, path, target):
+ """Add a symbolic link to the archive."""
+ assert not self.closed, "Can't add a symlink to a closed archive"
+
+ now = int(time.time())
+ self._ensure_directories(path, now)
+
+ tarinfo = self._make_skeleton_tarinfo(path, now)
+ tarinfo.type = tarfile.SYMTYPE
+ tarinfo.linkname = target
+ self.tarfile.addfile(tarinfo)
=== modified file 'lib/lp/services/tests/test_helpers.py'
--- lib/lp/services/tests/test_helpers.py 2011-12-23 23:44:59 +0000
+++ lib/lp/services/tests/test_helpers.py 2012-05-25 13:45:30 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2012 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
from doctest import DocTestSuite
@@ -6,7 +6,7 @@
import unittest
from lp.services import helpers
-from lp.translations.utilities.translation_export import LaunchpadWriteTarFile
+from lp.services.tarfile_helpers import LaunchpadWriteTarFile
def make_test_tarball_1():
=== removed file 'lib/lp/soyuz/doc/distroseriesqueue-ddtp-tarball.txt'
--- lib/lp/soyuz/doc/distroseriesqueue-ddtp-tarball.txt 2012-01-06 11:08:30 +0000
+++ lib/lp/soyuz/doc/distroseriesqueue-ddtp-tarball.txt 1970-01-01 00:00:00 +0000
@@ -1,308 +0,0 @@
-DDTP-TARBALL upload
--------------------
-
-This test will describe how launchpad treats an upload of DDTP
-tarball.
-
-DDTP (Debian Description Translation Project) tarball upload is
-consisted of a tarball (specified in the changesfile as a
-DDTP-TARBALL) containing all the supported index files for the DDTP
-contents driven by component.
-
-The index-files needs to be published in the ubuntu archive, according
-its target suite and component, under the 'i18n' (see
-http://en.wikipedia.org/wiki/Internationalization_and_localization)
-directory.
-
-Soonish APT will be able to download and follow the indexes files,
-then present the appropriated translated package description the users
-in ubuntu.
-
-For further info about DDTP see:
- https://wiki.ubuntu.com/TranslatedPackageDescriptionsSpec
-
- >>> from lp.registry.interfaces.distribution import IDistributionSet
-
- >>> ubuntutest = getUtility(IDistributionSet)['ubuntutest']
- >>> breezy_autotest = ubuntutest['breezy-autotest']
-
- >>> from lp.archiveuploader.nascentupload import NascentUpload
- >>> from lp.archiveuploader.tests import datadir, getPolicy
-
- >>> from lp.testing.gpgkeys import import_public_test_keys
- >>> import_public_test_keys()
-
-Login as an admin (or ubuntutest.archive_admin if we have one), since
-we need to access and modify PackageUpload records and other tables.
-
- >>> login('foo.bar@xxxxxxxxxxxxx')
-
-Set the email address for announcements:
-
- >>> breezy_autotest.changeslist = 'test@xxxxxxxxxxx'
-
-First, test the rejection of a misapplied changesfile name, which
-doesn't follow the accepted format, "<pkg>_<version>_<arch>.changes"
-
- >>> sync_policy = getPolicy(
- ... name='sync', distro='ubuntutest', distroseries=None)
-
- >>> from lp.services.log.logger import DevNullLogger
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('ddtp-tarball/translations-main_20060728.changes'),
- ... sync_policy, DevNullLogger())
- >>> upload.process()
- Traceback (most recent call last):
- ...
- EarlyReturnUploadError: An error occurred that prevented further
- processing.
-
-Now do a proper upload into the system.
-
- >>> insecure_policy = getPolicy(
- ... name='insecure', distro='ubuntutest', distroseries=None)
-
- >>> from lp.services.log.logger import FakeLogger
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('ddtp-tarball/translations-main_20060728_all.changes'),
- ... insecure_policy, FakeLogger())
- DEBUG Verifying signature on translations-main_20060728_all.changes
-
- >>> upload.process()
- DEBUG Beginning processing.
- DEBUG Verifying the changes file.
- DEBUG Verifying files in upload.
- DEBUG Single Custom Upload detected.
- DEBUG Finished checking upload.
-
-It was not rejected.
-
- >>> upload.is_rejected
- False
-
- >>> success = upload.do_accept()
- DEBUG Creating queue entry
- ...
- DEBUG Sent a mail:
- DEBUG Subject: [ubuntutest/breezy-autotest]
- translations_main_20060728.tar.gz - (Accepted)
- DEBUG Sender: Root <root@localhost>
- DEBUG Recipients: Foo Bar <foo.bar@xxxxxxxxxxxxx>
- DEBUG Bcc: Root <root@localhost>
- DEBUG Body:
- DEBUG .
- DEBUG * Duhh
- DEBUG
- DEBUG
- DEBUG Date: Thu, 01 Aug 2006 14:20:25 +0300
- DEBUG Changed-By: Foo Bar <foo.bar@xxxxxxxxxxxxx>
- DEBUG Origin: Ubuntu/warty
- DEBUG
- DEBUG
- DEBUG ==
- DEBUG
- DEBUG Announcing to test@xxxxxxxxxxx
- DEBUG
- DEBUG Thank you for your contribution to Ubuntu Test.
- DEBUG
- DEBUG --
- DEBUG You are receiving this email because you are the uploader,
- maintainer or
- DEBUG signer of the above package.
- DEBUG Sent a mail:
- DEBUG Subject: [ubuntutest/breezy-autotest]
- translations_main_20060728.tar.gz - (Accepted)
- DEBUG Sender: Foo Bar <foo.bar@xxxxxxxxxxxxx>
- DEBUG Recipients: test@xxxxxxxxxxx
- DEBUG Bcc: Root <root@localhost>
- DEBUG Body:
- DEBUG .
- DEBUG * Duhh
- DEBUG
- DEBUG
- DEBUG Date: Thu, 01 Aug 2006 14:20:25 +0300
- DEBUG Changed-By: Foo Bar <foo.bar@xxxxxxxxxxxxx>
- DEBUG Origin: Ubuntu/warty
- DEBUG
-
-And all things worked.
-
- >>> success
- True
-
- >>> not upload.rejection_message
- True
-
-We need to commit the transaction to be able to use the librarian files.
-
- >>> flush_database_updates()
- >>> transaction.commit()
-
-Let's use the script to fetch the ddtp-tarball upload:
-
- >>> import subprocess
- >>> import os
- >>> import sys
- >>> from lp.services.config import config
-
- >>> script = os.path.join(
- ... config.root, "scripts", "ftpmaster-tools", "queue")
-
- >>> process = subprocess.Popen([sys.executable, script, "-Q", "accepted",
- ... "-s", "breezy-autotest", "fetch", "trans",
- ... "-d", "ubuntutest"],
- ... stdout=subprocess.PIPE)
- >>> stdout, stderr = process.communicate()
- >>> process.returncode
- 0
- >>> print stdout
- Initializing connection to queue accepted
- Running: "fetch trans"
- Fetching ubuntutest/breezy-autotest (ACCEPTED) 1/1
- ---------------------------------------------------------------------------
- Constructing translations-main_20060728_all.changes
- Constructing translations_main_20060728.tar.gz
- ---------------------------------------------------------------------------
- 1/1 total
- <BLANKLINE>
-
-Check if the files were written:
-
- >>> os.path.exists('translations-main_20060728_all.changes')
- True
- >>> os.path.exists('translations_main_20060728.tar.gz')
- True
-
-Cleanup the mess:
-
- >>> os.remove('translations-main_20060728_all.changes')
- >>> os.remove('translations_main_20060728.tar.gz')
-
-
-Inspect the QUEUE looking for an ACCEPTED entry corresponding to the
-previous upload:
-
- >>> from lp.soyuz.enums import PackageUploadStatus
- >>> queue_item = breezy_autotest.getPackageUploads(
- ... status=PackageUploadStatus.ACCEPTED)[0]
- >>> queue_item.customfiles[0].libraryfilealias.filename
- u'translations_main_20060728.tar.gz'
-
-Do the publish, i.e process the CustomUpload decompressing and writing
-the tarball contents in the archive, collect and check debug message:
-
-The umask must be 022 in order to perform the upload.
-
- >>> old_mask = os.umask(022)
- >>> pub_records = queue_item.realiseUpload(FakeLogger())
- DEBUG Publishing custom translations_main_20060728.tar.gz to
- ubuntutest/breezy-autotest
- >>> print '%03o' % os.umask(old_mask)
- 022
-
-Check what was published in the target directory:
-
- >>> import os
- >>> archive_dir = '/var/tmp/archive/'
-
-DDTP indexes are published in the "i18n" directory inside archive dists:
-
- >>> upgrade_dir = 'ubuntutest/dists/breezy-autotest/main/i18n'
- >>> target_dir = os.path.join(archive_dir, upgrade_dir)
-
-We simply decompress the contents of the tarball, the files follow
-the format:
-
- Translation-<lang-code>{'', '.gz', '.bz2'}
-
-We have only used uncompressed files for tests:
-
- >>> dir_list = os.listdir(target_dir)
- >>> for filename in sorted(dir_list):
- ... content = open(os.path.join(target_dir, filename), 'rb').read()
- ... print filename, len(content)
- Translation-bn 0
- Translation-ca 0
- Translation-en 16
- Translation-pt_BR 15
-
-** Note that a incomming directory inside the tarball "i18n" directory
-was skipped by the DDTP processor
-
-Upload a new DDTP tarball which is supposed to only add a new language
-index , the "bn" one, and left the rest untouched, this feature is
-know as "partial update" and will save some storage when fixing specific
-language indexes or adding new ones:
-
- >>> insecure_policy = getPolicy(
- ... name='insecure', distro='ubuntutest', distroseries=None)
-
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('ddtp-tarball/translations-main_20060817_all.changes'),
- ... insecure_policy, DevNullLogger())
- >>> upload.process()
- >>> upload.is_rejected
- False
- >>> success = upload.do_accept()
- >>> success
- True
-
-
-We need to commit the transaction to be able to use the librarian files.
-
- >>> flush_database_updates()
- >>> transaction.commit()
-
-
-Our archive uses `dsync` to replace identical files by hard link in
-order to save some space. This feature breaks the tarfile.extract method,
-because it doesn't remove pre-existing files, it simply overwrite them.
-
-Create a hard link to "bn" package file.
-
- >>> src = os.path.join(target_dir, 'Translation-ca')
- >>> dest = os.path.join(target_dir, 'Translation-bn')
- >>> os.remove(dest)
- >>> os.link(src, dest)
-
- >>> os.path.exists(dest)
- True
- >>> os.stat(dest).st_nlink
- 2
- >>> os.stat(src).st_nlink
- 2
-
-
-Retrieve and publish new custom upload:
-
- >>> queue_item = breezy_autotest.getPackageUploads(
- ... status=PackageUploadStatus.ACCEPTED)[0]
-
- >>> pub_records = queue_item.realiseUpload()
-
-
-Check if there is new content for "bn" index:
-
- >>> dir_list = os.listdir(target_dir)
- >>> for filename in sorted(dir_list):
- ... content = open(os.path.join(target_dir, filename), 'rb').read()
- ... print filename, len(content)
- Translation-bn 22
- Translation-ca 0
- Translation-en 16
- Translation-pt_BR 15
-
-
-Check if the contents of "ca" were not affected by the "bn" contents
-(this would happen if "bn" was not removed, left as hard link):
-
- >>> open(os.path.join(target_dir, 'Translation-bn')).read()
- 'hardlinks are bad !!!\n'
-
- >>> open(os.path.join(target_dir, 'Translation-ca')).read()
- ''
-
-Remove the directory to keep the test working.
-
- >>> import shutil
- >>> shutil.rmtree(target_dir)
=== removed file 'lib/lp/soyuz/doc/distroseriesqueue-debian-installer.txt'
--- lib/lp/soyuz/doc/distroseriesqueue-debian-installer.txt 2012-05-23 23:02:08 +0000
+++ lib/lp/soyuz/doc/distroseriesqueue-debian-installer.txt 1970-01-01 00:00:00 +0000
@@ -1,187 +0,0 @@
-= Debian Installer Custom Uploads =
-
-Debian Installer uploads are treated as one of the Soyuz "custom"
-types.
-
-They come from buildd as result of the 'debian-installer' source
-build. This binary upload usually contains:
-
- * The debian-installer binaries (DEB)
- * The debian-installer translation (TAR.GZ, custom ROSETTA_TRANSLATIONS)
- * The debian-installer special archive files (TAR.GZ, custom DEBIAN_INSTALLER)
-
-We have a special processor to publish debian-installer special archive
-file contents.
-
-The debian-installer filename should be something like:
-
-<BASE>_<VERSION>_<ARCH>.tar.gz
-
-Where:
-
- * BASE: base name (usually 'debian-installer-images')
-
- * VERSION: encoded version (something like '20061102ubuntu14')
-
- * ARCH: targeted architecture tag ('i386', 'amd64', etc)
-
-The contents are extracted in the archive in the following path:
-
- * <ARCHIVE>/dists/<SUITE>/main/installer-<ARCH>/
-
-
-Let's try one test upload and verify the results of its publication.
-In order to keep the tests simple we will upload only the
-debian-installer tarball, despite the fact that it's very unlikely to
-happen in production:
-
- >>> from lp.testing.gpgkeys import import_public_test_keys
- >>> import_public_test_keys()
- >>> login('foo.bar@xxxxxxxxxxxxx')
-
- >>> from lp.registry.interfaces.distribution import IDistributionSet
- >>> ubuntutest = getUtility(IDistributionSet)['ubuntutest']
-
- >>> from lp.archiveuploader.nascentupload import NascentUpload
- >>> from lp.archiveuploader.tests import datadir, getPolicy
-
- >>> anything_policy = getPolicy(
- ... name='anything', distro='ubuntutest', distroseries=None)
- >>> anything_policy.setDistroSeriesAndPocket('hoary-test')
-
-Set the email address for annoucements emails:
-
- >>> anything_policy.distroseries.changeslist = 'announce@xxxxxxxxxxx'
-
- >>> from lp.services.log.logger import DevNullLogger
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir(
- ... 'debian-installer/debian-installer_20070214ubuntu1_i386.changes'),
- ... anything_policy, DevNullLogger())
- >>> upload.process()
- >>> success = upload.do_accept()
- >>> success
- True
-
-Upload was successfully 'processed' and 'accepted'. Two email messages
-were generated (acceptance and announcement). Inspect the queue (IDRQ)
-information we find out that it contains one custom upload.
-
- >>> upload.queue_root.status.name
- 'ACCEPTED'
-
-Make librarian files available for publication:
-
- >>> import transaction
- >>> transaction.commit()
-
- >>> from lp.services.mail import stub
- >>> len(stub.test_emails)
- 2
- >>> upload.queue_root.customfiles.count()
- 1
-
-Let's force an error simulating a conflict in archive:
-
- >>> import os
- >>> archive_path = "/var/tmp/archive/"
- >>> installer_path = os.path.join(
- ... archive_path, "ubuntutest/dists/hoary-test/main/installer-i386")
-
- >>> os.makedirs(os.path.join(installer_path, '20070214ubuntu1'))
-
-Try to publish the custom upload, but it will be skipped due the disk
-conflict and the queue item will remain in ACCEPTED
-
- >>> from lp.services.log.logger import FakeLogger
- >>> pub_records = upload.queue_root.realiseUpload(FakeLogger())
- DEBUG Publishing custom debian-installer-images_20070214ubuntu1_i386.tar.gz to ubuntutest/hoary-test
- ERROR Queue item ignored: installer build i386 for architecture 20070214ubuntu1 already exists
- >>> upload.queue_root.status.name
- 'ACCEPTED'
-
-Let's remove the conflicted directory from disk and expect it to work
-next time:
-
- >>> os.rmdir(os.path.join(installer_path, '20070214ubuntu1'))
-
-The upload task requires that the umask be 022, otherwise things will end up
-with the wrong permission.
-
- >>> old_mask = os.umask(002)
- >>> pub_records = upload.queue_root.realiseUpload(FakeLogger())
- DEBUG ...
- ERROR Queue item ignored: Bad umask; expected 022, got 002
-
-Reset the umask, remove any residual directories, and try again.
-
- >>> os.rmdir(os.path.join(installer_path, '20070214ubuntu1'))
- >>> print '%03o' % os.umask(old_mask)
- 002
-
-Process (publish) the custom upload:
-
- >>> pub_records = upload.queue_root.realiseUpload(DevNullLogger())
- >>> upload.queue_root.status.name
- 'DONE'
-
-The current debian-installer directory (20070214ubuntu1) and a link to
-the last version (we only have the one we uploaded) should be
-presented:
-
- >>> sorted(os.listdir(installer_path))
- ['20070214ubuntu1', 'current']
-
-As a consistency check, let's verify one of the files inside the
-tarball:
-
- >>> the_file = ("current/images/netboot/ubuntu-installer/i386/"
- ... "pxelinux.cfg.serial-9600/default")
- >>> the_file_path = os.path.join(installer_path, the_file)
- >>> print open(the_file_path).read()
- hey
-
-Also check if the symbolic link included in the installer tarball
-looks sane:
-
- >>> the_link = "current/images/netboot/foo"
- >>> the_link_path = os.path.join(installer_path, the_link)
- >>> os.readlink(the_link_path)
- 'ubuntu-installer/i386/pxelinux.cfg.serial-9600/default'
-
- >>> the_dir_link = "current/images/netboot/link_to_dir"
- >>> the_dir_link_path = os.path.join(installer_path, the_dir_link)
- >>> os.readlink(the_dir_link_path)
- 'ubuntu-installer/i386/pxelinux.cfg.serial-9600/'
-
-Check permissions on the created directories to satisfy the
-publication path. They are created with os.makedirs() and forced to
-set permission '0755' instead of the default '0777' (see further
-details in bug #107068):
-
- >>> oct(os.stat(installer_path).st_mode & 0777)
- '0755'
-
- >>> the_path = os.path.join(installer_path, os.pardir)
- >>> oct(os.stat(the_path).st_mode & 0777)
- '0755'
-
-Check permissions of created files and directories:
-
- >>> the_file = ("current/images/netboot/ubuntu-installer/i386/"
- ... "pxelinux.cfg.serial-9600/default")
- >>> the_file_path = os.path.join(installer_path, the_file)
- >>> oct(os.stat(the_file_path).st_mode & 0777)
- '0644'
-
- >>> the_dir = ("current/images/netboot/ubuntu-installer/i386/"
- ... "pxelinux.cfg.serial-9600/")
- >>> the_dir_path = os.path.join(installer_path, the_dir)
- >>> oct(os.stat(the_dir_path).st_mode & 0777)
- '0755'
-
-
-Remove local archive contents:
-
- >>> import shutil
- >>> shutil.rmtree(archive_path)
=== removed file 'lib/lp/soyuz/doc/distroseriesqueue-dist-upgrader.txt'
--- lib/lp/soyuz/doc/distroseriesqueue-dist-upgrader.txt 2012-01-06 11:08:30 +0000
+++ lib/lp/soyuz/doc/distroseriesqueue-dist-upgrader.txt 1970-01-01 00:00:00 +0000
@@ -1,339 +0,0 @@
-DIST-UPGRADE upload
-===================
-
-This test will check the upload of dist-upgrade tarballs
-
- >>> from lp.registry.interfaces.distribution import IDistributionSet
-
- >>> from lp.archiveuploader.nascentupload import NascentUpload
- >>> from lp.archiveuploader.tests import datadir, getPolicy
-
- >>> ubuntutest = getUtility(IDistributionSet)['ubuntutest']
- >>> breezy_autotest = ubuntutest['breezy-autotest']
-
- >>> from lp.testing.gpgkeys import import_public_test_keys
- >>> import_public_test_keys()
-
-Login as an admin.
-
- >>> login('foo.bar@xxxxxxxxxxxxx')
-
-First, test the rejection of a misapplied changesfile name:
-
- >>> sync_policy = getPolicy(
- ... name='sync', distro='ubuntutest', distroseries=None)
-
- >>> from lp.services.log.logger import FakeLogger
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('dist-upgrader/dist-upgrader_20060302.0120.changes'),
- ... sync_policy, FakeLogger())
- DEBUG dist-upgrader_20060302.0120.changes can be unsigned.
- >>> upload.process()
- Traceback (most recent call last):
- ...
- EarlyReturnUploadError: An error occurred that prevented further
- processing.
-
-Now do a proper upload into the system.
-
- >>> insecure_policy = getPolicy(
- ... name='insecure', distro='ubuntutest', distroseries=None)
-
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('dist-upgrader/dist-upgrader_20060302.0120_all.changes'),
- ... insecure_policy, FakeLogger())
- DEBUG Verifying signature on dist-upgrader_20060302.0120_all.changes
-
- >>> upload.process()
- DEBUG Beginning processing.
- DEBUG Verifying the changes file.
- DEBUG Verifying files in upload.
- DEBUG Single Custom Upload detected.
- DEBUG Finished checking upload.
-
-It was not rejected.
-
- >>> upload.is_rejected
- False
-
- >>> success = upload.do_accept()
- DEBUG Creating queue entry
- DEBUG Setting it to ACCEPTED
- DEBUG Building recipients list.
- ...
- DEBUG Adding recipient: 'Foo Bar <foo.bar@xxxxxxxxxxxxx>'
- DEBUG Sent a mail:
- ...
- DEBUG signer of the above package.
-
-And all things worked.
-
- >>> success
- True
-
- >>> print upload.rejection_message
-
-We need to commit the transaction to be able to use the librarian files.
-XXX: flush_database_updates() shouldn't be needed. This seems to be
-Bug 3989 -- StuartBishop 20060713
-
- >>> flush_database_updates()
- >>> transaction.commit()
-
-Let's check how queue tool handles dist_upgrader uploads:
-
- >>> import subprocess
- >>> import os
- >>> import sys
- >>> from lp.services.config import config
-
- >>> script = os.path.join(
- ... config.root, "scripts", "ftpmaster-tools", "queue")
-
-
-Rejecting item from accepted queue (very unlikely, normally it would
-be rejecting from unapproved or new, fix bug #54649):
-
- >>> process = subprocess.Popen([
- ... sys.executable, script,
- ... "-Q", "accepted",
- ... "-s", "breezy-autotest", "reject", "dist",
- ... "-d", "ubuntutest", "-M"],
- ... stderr=subprocess.PIPE,
- ... stdout=subprocess.PIPE)
- >>> stdout, stderr = process.communicate()
- >>> process.returncode
- 0
- >>> print stdout
- Initializing connection to queue accepted
- ...
- Rejecting dist-upgrader_20060302.0120_all.tar.gz
- ...
-
-
-Accepting from rejected queue (resurrecting an reject upload is also
-very unlikely, however it's only for testing purposes, fix #54649):
-
- >>> process = subprocess.Popen([sys.executable, script, "-Q", "rejected",
- ... "-s", "breezy-autotest", "accept", "dist",
- ... "-d", "ubuntutest", "-M"],
- ... stderr=subprocess.PIPE,
- ... stdout=subprocess.PIPE)
- >>> stdout, stderr = process.communicate()
- >>> process.returncode
- 0
- >>> print stdout
- Initializing connection to queue rejected
- ...
- Accepting dist-upgrader_20060302.0120_all.tar.gz
- ...
-
-
-Let's use the script to fetch the dist-upgrader upload:
-
- >>> process = subprocess.Popen([sys.executable, script, "-Q", "accepted",
- ... "-s", "breezy-autotest", "fetch", "dist",
- ... "-d", "ubuntutest"],
- ... stdout=subprocess.PIPE)
- >>> stdout, stderr = process.communicate()
- >>> process.returncode
- 0
- >>> print stdout
- Initializing connection to queue accepted
- Running: "fetch dist"
- Fetching ubuntutest/breezy-autotest (ACCEPTED) 1/1
- ---------------------------------------------------------------------------
- Constructing dist-upgrader_20060302.0120_all.changes
- Constructing dist-upgrader_20060302.0120_all.tar.gz
- ---------------------------------------------------------------------------
- 1/1 total
- <BLANKLINE>
-
-
-Check if the files were written:
-
- >>> os.path.exists('dist-upgrader_20060302.0120_all.changes')
- True
- >>> os.path.exists('dist-upgrader_20060302.0120_all.tar.gz')
- True
-
-Cleanup the mess:
-
- >>> os.remove('dist-upgrader_20060302.0120_all.changes')
- >>> os.remove('dist-upgrader_20060302.0120_all.tar.gz')
-
-
-Inspect the QUEUE looking for an ACCEPTED entry corresponding to the
-previous upload:
-
- >>> from lp.soyuz.enums import PackageUploadStatus
- >>> queue_item = breezy_autotest.getPackageUploads(
- ... status=PackageUploadStatus.ACCEPTED)[0]
- >>> queue_item.customfiles[0].libraryfilealias.filename
- u'dist-upgrader_20060302.0120_all.tar.gz'
-
-Do the publish, i.e process the CustomUpload writing the tarball in
-the archive. The umask must be 022 in order to perform the upload.
-
- >>> old_mask = os.umask(022)
- >>> pub_records = queue_item.realiseUpload()
- >>> print '%03o' % os.umask(old_mask)
- 022
-
-Check what was published in the target directory:
-
- >>> import os
- >>> archive_dir = '/var/tmp/archive/'
- >>> upgrade_dir = (
- ... 'ubuntutest/dists/breezy-autotest/main/dist-upgrader-all')
- >>> target_dir = os.path.join(archive_dir, upgrade_dir)
- >>> content = sorted(os.listdir(target_dir))
- >>> content
- ['20060302.0120', 'current']
-
-Check the content of the directories published and the integrity of
-the 'current' symbolic link.
-
- >>> real_path = os.path.join(target_dir, '20060302.0120')
- >>> current_path = os.path.join(target_dir, 'current')
-
- >>> os.path.islink(current_path)
- True
-
- >>> os.path.join(target_dir, os.readlink(current_path)) == real_path
- True
-
- >>> sorted(os.listdir(real_path))
- ['ReleaseAnouncement', 'dapper.tar.gz']
-
-
-Pre-publication version checks
-------------------------------
-
-Dist-Upgrader version is checked before installing any files in the
-archive.
-
-If any problem happen during the custom publication, the error will be
-printed and the queue item will remain in ACCEPTED queue waiting for
-manual intervention.
-
- >>> insecure_policy = getPolicy(
- ... name='insecure', distro='ubuntutest', distroseries=None)
-
- >>> upload = NascentUpload.from_changesfile_path(
- ... datadir('dist-upgrader/dist-upgrader_20070219.1234_all.changes'),
- ... insecure_policy, FakeLogger())
- DEBUG Verifying signature on dist-upgrader_20070219.1234_all.changes
-
- >>> upload.process()
- DEBUG Beginning processing.
- DEBUG Verifying the changes file.
- DEBUG Verifying files in upload.
- DEBUG Single Custom Upload detected.
- DEBUG Finished checking upload.
-
- >>> success = upload.do_accept()
- DEBUG Creating queue entry
- DEBUG Setting it to ACCEPTED
- DEBUG Building recipients list.
- ...
- DEBUG Adding recipient: 'Foo Bar <foo.bar@xxxxxxxxxxxxx>'
- DEBUG Sent a mail:
- ...
- DEBUG signer of the above package.
-
-Make the just-uploaded librarian file visible:
-
- >>> transaction.commit()
-
-Check current archive status:
-
- >>> previous_content = sorted(os.listdir(target_dir))
- >>> previous_content
- ['20060302.0120', 'current']
-
-Publish the custom upload:
-
- >>> pub_records = upload.queue_root.realiseUpload(FakeLogger())
- DEBUG Publishing custom dist-upgrader_20070219.1234_all.tar.gz to
- ubuntutest/breezy-autotest
- ERROR Queue item ignored: bad version found in
- '.../dist-upgrader_20070219.1234_all.tar.gz': Could not parse version: Bad
- upstream version format foobar
-
-Check if the queue item remained in ACCEPTED and not cruft was
-inserted in the archive:
-
- >>> upload.queue_root.status.name
- 'ACCEPTED'
-
- >>> content = sorted(os.listdir(target_dir))
- >>> content == previous_content
- True
-
-Remove the directory to keep the test working.
-
- >>> import shutil
- >>> shutil.rmtree(target_dir)
-
-
-Dist Upgrader to PPA
---------------------
-
-Custom uploads for PPAs should work as expected, i.e, publish its
-files in the respective PPA location.
-
- >>> from lp.registry.interfaces.person import IPersonSet
- >>> from lp.soyuz.enums import ArchivePurpose
- >>> from lp.soyuz.interfaces.archive import IArchiveSet
-
- >>> name16 = getUtility(IPersonSet).getByName("name16")
- >>> foobar_archive = getUtility(IArchiveSet).new(
- ... distribution=ubuntutest, purpose=ArchivePurpose.PPA,
- ... owner=name16)
-
- >>> insecure_policy.archive = foobar_archive
-
- >>> ppa_upload = NascentUpload.from_changesfile_path(
- ... datadir('dist-upgrader/dist-upgrader_20060302.0120_all.changes'),
- ... insecure_policy, FakeLogger())
- DEBUG Verifying signature on dist-upgrader_20060302.0120_all.changes
-
- >>> ppa_upload.process()
- DEBUG Beginning processing.
- DEBUG Verifying the changes file.
- DEBUG Verifying files in upload.
- DEBUG Single Custom Upload detected.
- DEBUG Finished checking upload.
-
- >>> success = ppa_upload.do_accept()
- DEBUG Creating queue entry
- DEBUG Setting it to ACCEPTED
- DEBUG Building recipients list.
- DEBUG Adding recipient: 'Foo Bar <foo.bar@xxxxxxxxxxxxx>'
- DEBUG Sent a mail:
- ...
- DEBUG PPA package.
-
- >>> transaction.commit()
-
-The upload task requires that the umask be 022, otherwise things will end up
-with the wrong permission.
-
- >>> old_mask = os.umask(022)
- >>> pub_records = ppa_upload.queue_root.realiseUpload(FakeLogger())
- DEBUG Publishing custom dist-upgrader_20060302.0120_all.tar.gz to
- ubuntutest/breezy-autotest
- >>> ignore = os.umask(old_mask)
-
- >>> ppa_root = config.personalpackagearchive.root
- >>> ppa_dir = os.path.join(ppa_root, 'name16', 'ppa')
- >>> target_dir = os.path.join(ppa_dir, upgrade_dir)
- >>> content = sorted(os.listdir(target_dir))
- >>> content
- ['20060302.0120', 'current']
-
-Remove the directory to keep the test working.
-
- >>> import shutil
- >>> shutil.rmtree(target_dir)
=== added file 'lib/lp/soyuz/tests/test_distroseriesqueue_ddtp_tarball.py'
--- lib/lp/soyuz/tests/test_distroseriesqueue_ddtp_tarball.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/tests/test_distroseriesqueue_ddtp_tarball.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,104 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test upload and queue manipulation of DDTP tarballs.
+
+DDTP (Debian Description Translation Project) uploads consist of a tarball
+containing translations of package descriptions for a component. These
+translations should be published in the Ubuntu archive under the
+dists/SUITE/COMPONENT/i18n/ directory.
+
+ https://wiki.ubuntu.com/TranslatedPackageDescriptionsSpec
+
+See also lp.archivepublisher.tests.test_ddtp_tarball for detailed tests of
+ddtp-tarball extraction.
+"""
+
+import os
+
+import transaction
+
+from lp.archiveuploader.nascentupload import (
+ EarlyReturnUploadError,
+ NascentUpload,
+ )
+from lp.archiveuploader.tests import (
+ datadir,
+ getPolicy,
+ )
+from lp.services.log.logger import DevNullLogger
+from lp.soyuz.scripts.queue import (
+ CommandRunner,
+ name_queue_map,
+ )
+from lp.soyuz.tests.test_publishing import TestNativePublishingBase
+from lp.testing.gpgkeys import import_public_test_keys
+
+
+class TestDistroSeriesQueueDdtpTarball(TestNativePublishingBase):
+
+ def setUp(self):
+ super(TestDistroSeriesQueueDdtpTarball, self).setUp()
+ import_public_test_keys()
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+ self.anything_policy = getPolicy(
+ name="anything", distro="ubuntutest", distroseries=None)
+ self.absolutely_anything_policy = getPolicy(
+ name="absolutely-anything", distro="ubuntutest", distroseries=None)
+ self.logger = DevNullLogger()
+
+ def test_rejects_misspelled_changesfile_name(self):
+ upload = NascentUpload.from_changesfile_path(
+ datadir("ddtp-tarball/translations-main_20060728.changes"),
+ self.absolutely_anything_policy, self.logger)
+ self.assertRaises(EarlyReturnUploadError, upload.process)
+
+ def uploadTestData(self, version):
+ upload = NascentUpload.from_changesfile_path(
+ datadir("ddtp-tarball/translations-main_%s_all.changes" % version),
+ self.anything_policy, self.logger)
+ upload.process()
+ self.assertFalse(upload.is_rejected)
+ self.assertTrue(upload.do_accept())
+ self.assertFalse(upload.rejection_message)
+ return upload
+
+ def test_accepts_correct_upload(self):
+ self.uploadTestData("20060728")
+
+ def runQueueCommand(self, queue_name, args):
+ def null_display(text):
+ pass
+
+ queue = name_queue_map[queue_name]
+ runner = CommandRunner(
+ queue, "ubuntutest", "breezy-autotest", True, None, None, None,
+ display=null_display)
+ runner.execute(args)
+
+ def test_queue_tool_behaviour(self):
+ # The queue tool can fetch ddtp-tarball uploads.
+ self.uploadTestData("20060728")
+ # Make sure that we can use the librarian files.
+ transaction.commit()
+ # Fetch upload into a temporary directory.
+ self.useTempDir()
+ self.runQueueCommand("accepted", ["fetch", "trans"])
+ expected_entries = [
+ "translations-main_20060728_all.changes",
+ "translations_main_20060728.tar.gz",
+ ]
+ self.assertContentEqual(expected_entries, os.listdir("."))
+
+ def test_publish(self):
+ upload = self.uploadTestData("20060728")
+ transaction.commit()
+ upload.queue_root.realiseUpload(self.logger)
+ target_dir = os.path.join(
+ self.config.distroroot, "ubuntutest", "dists", "breezy-autotest",
+ "main", "i18n")
+ # In this high-level test, we only care that something was unpacked.
+ self.assertTrue([name for name in os.listdir(target_dir)
+ if name.startswith("Translation-")])
=== added file 'lib/lp/soyuz/tests/test_distroseriesqueue_debian_installer.py'
--- lib/lp/soyuz/tests/test_distroseriesqueue_debian_installer.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/tests/test_distroseriesqueue_debian_installer.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,74 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test upload and queue manipulation of debian-installer custom uploads.
+
+See also lp.archivepublisher.tests.test_debian_installer for detailed tests
+of debian-installer custom upload extraction.
+"""
+
+import os
+
+import transaction
+
+from lp.archivepublisher.debian_installer import DebianInstallerAlreadyExists
+from lp.archiveuploader.nascentupload import NascentUpload
+from lp.archiveuploader.tests import (
+ datadir,
+ getPolicy,
+ )
+from lp.services.log.logger import DevNullLogger
+from lp.services.mail import stub
+from lp.soyuz.tests.test_publishing import TestNativePublishingBase
+from lp.testing.gpgkeys import import_public_test_keys
+
+
+class TestDistroSeriesQueueDebianInstaller(TestNativePublishingBase):
+
+ def setUp(self):
+ super(TestDistroSeriesQueueDebianInstaller, self).setUp()
+ import_public_test_keys()
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+ self.anything_policy = getPolicy(
+ name="anything", distro="ubuntutest", distroseries=None)
+ self.logger = DevNullLogger()
+
+ def uploadTestData(self):
+ upload = NascentUpload.from_changesfile_path(
+ datadir(
+ "debian-installer/"
+ "debian-installer_20070214ubuntu1_i386.changes"),
+ self.anything_policy, self.logger)
+ upload.process()
+ self.assertFalse(upload.is_rejected)
+ self.assertTrue(upload.do_accept())
+ self.assertFalse(upload.rejection_message)
+ return upload
+
+ def test_accepts_correct_upload(self):
+ upload = self.uploadTestData()
+ self.assertEqual(1, upload.queue_root.customfiles.count())
+
+ def test_generates_mail(self):
+ # Two e-mail messages were generated (acceptance and announcement).
+ self.anything_policy.setDistroSeriesAndPocket("hoary-test")
+ self.anything_policy.distroseries.changeslist = "announce@xxxxxxxxxxx"
+ self.uploadTestData()
+ self.assertEqual(2, len(stub.test_emails))
+
+ def test_bad_upload_remains_in_accepted(self):
+ # Bad debian-installer uploads remain in accepted. Simulate an
+ # on-disk conflict to force an error.
+ upload = self.uploadTestData()
+ # Make sure that we can use the librarian files.
+ transaction.commit()
+ os.makedirs(os.path.join(
+ self.config.distroroot, "ubuntutest", "dists", "hoary-test",
+ "main", "installer-i386", "20070214ubuntu1"))
+ self.assertFalse(upload.queue_root.realiseUpload(self.logger))
+ self.assertRaises(
+ DebianInstallerAlreadyExists,
+ upload.queue_root.customfiles[0].publish, self.logger)
+ self.assertEqual("ACCEPTED", upload.queue_root.status.name)
=== added file 'lib/lp/soyuz/tests/test_distroseriesqueue_dist_upgrader.py'
--- lib/lp/soyuz/tests/test_distroseriesqueue_dist_upgrader.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/tests/test_distroseriesqueue_dist_upgrader.py 2012-05-25 13:45:30 +0000
@@ -0,0 +1,133 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test upload and queue manipulation of dist-upgrader tarballs.
+
+See also lp.archivepublisher.tests.test_dist_upgrader for detailed tests of
+dist-upgrader tarball extraction.
+"""
+
+import os
+import shutil
+
+import transaction
+
+from lp.archivepublisher.dist_upgrader import DistUpgraderBadVersion
+from lp.archiveuploader.nascentupload import (
+ EarlyReturnUploadError,
+ NascentUpload,
+ )
+from lp.archiveuploader.tests import (
+ datadir,
+ getPolicy,
+ )
+from lp.services.config import config
+from lp.services.log.logger import DevNullLogger
+from lp.soyuz.scripts.queue import (
+ CommandRunner,
+ name_queue_map,
+ )
+from lp.soyuz.tests.test_publishing import TestNativePublishingBase
+from lp.testing.gpgkeys import import_public_test_keys
+
+
+class TestDistroSeriesQueueDistUpgrader(TestNativePublishingBase):
+
+ def setUp(self):
+ super(TestDistroSeriesQueueDistUpgrader, self).setUp()
+ import_public_test_keys()
+ # CustomUpload.installFiles requires a umask of 022.
+ old_umask = os.umask(022)
+ self.addCleanup(os.umask, old_umask)
+ self.anything_policy = getPolicy(
+ name="anything", distro="ubuntutest", distroseries=None)
+ self.absolutely_anything_policy = getPolicy(
+ name="absolutely-anything", distro="ubuntutest", distroseries=None)
+ self.logger = DevNullLogger()
+
+ def tearDown(self):
+ super(TestDistroSeriesQueueDistUpgrader, self).tearDown()
+ if os.path.exists(config.personalpackagearchive.root):
+ shutil.rmtree(config.personalpackagearchive.root)
+
+ def test_rejects_misspelled_changesfile_name(self):
+ upload = NascentUpload.from_changesfile_path(
+ datadir("dist-upgrader/dist-upgrader_20060302.0120.changes"),
+ self.absolutely_anything_policy, self.logger)
+ self.assertRaises(EarlyReturnUploadError, upload.process)
+
+ def uploadTestData(self, version):
+ upload = NascentUpload.from_changesfile_path(
+ datadir("dist-upgrader/dist-upgrader_%s_all.changes" % version),
+ self.anything_policy, self.logger)
+ upload.process()
+ self.assertFalse(upload.is_rejected)
+ self.assertTrue(upload.do_accept())
+ self.assertFalse(upload.rejection_message)
+ return upload
+
+ def test_accepts_correct_upload(self):
+ self.uploadTestData("20060302.0120")
+
+ def runQueueCommand(self, queue_name, args):
+ def null_display(text):
+ pass
+
+ queue = name_queue_map[queue_name]
+ runner = CommandRunner(
+ queue, "ubuntutest", "breezy-autotest", True, None, None, None,
+ display=null_display)
+ runner.execute(args)
+
+ def test_queue_tool_behaviour(self):
+ # The queue tool can accept, reject, and fetch dist-upgrader
+ # uploads. See bug #54649.
+ upload = self.uploadTestData("20060302.0120")
+ # Make sure that we can use the librarian files.
+ transaction.commit()
+ # Reject from accepted queue (unlikely, would normally be from
+ # unapproved or new).
+ self.runQueueCommand("accepted", ["reject", "dist"])
+ self.assertEqual("REJECTED", upload.queue_root.status.name)
+ # Accept from rejected queue (also unlikely, but only for testing).
+ self.runQueueCommand("rejected", ["accept", "dist"])
+ self.assertEqual("ACCEPTED", upload.queue_root.status.name)
+ # Fetch upload into a temporary directory.
+ self.useTempDir()
+ self.runQueueCommand("accepted", ["fetch", "dist"])
+ expected_entries = [
+ "dist-upgrader_20060302.0120_all.changes",
+ "dist-upgrader_20060302.0120_all.tar.gz",
+ ]
+ self.assertContentEqual(expected_entries, os.listdir("."))
+
+ def test_bad_upload_remains_in_accepted(self):
+ # Bad dist-upgrader uploads remain in ACCEPTED.
+ upload = self.uploadTestData("20070219.1234")
+ # Make sure that we can use the librarian files.
+ transaction.commit()
+ self.assertFalse(upload.queue_root.realiseUpload(self.logger))
+ self.assertEqual(1, upload.queue_root.customfiles.count())
+ self.assertRaises(
+ DistUpgraderBadVersion, upload.queue_root.customfiles[0].publish,
+ self.logger)
+ self.assertEqual("ACCEPTED", upload.queue_root.status.name)
+
+ def test_ppa_publishing_location(self):
+ # A PPA dist-upgrader upload is published to the right place.
+ archive = self.factory.makeArchive(distribution=self.ubuntutest)
+ self.anything_policy.archive = archive
+ ppa_upload = self.uploadTestData("20060302.0120")
+ ppa_upload = NascentUpload.from_changesfile_path(
+ datadir("dist-upgrader/dist-upgrader_20060302.0120_all.changes"),
+ self.anything_policy, self.logger)
+ ppa_upload.process()
+ self.assertTrue(ppa_upload.do_accept())
+ transaction.commit()
+ ppa_upload.queue_root.realiseUpload(self.logger)
+ ppa_root = config.personalpackagearchive.root
+ ppa_dir = os.path.join(ppa_root, archive.owner.name, archive.name)
+ target_dir = os.path.join(
+ ppa_dir, "ubuntutest/dists/breezy-autotest/main/dist-upgrader-all")
+ self.assertContentEqual(
+ ["20060302.0120", "current"], os.listdir(target_dir))
=== modified file 'lib/lp/soyuz/tests/test_doc.py'
--- lib/lp/soyuz/tests/test_doc.py 2012-05-16 01:09:00 +0000
+++ lib/lp/soyuz/tests/test_doc.py 2012-05-25 13:45:30 +0000
@@ -77,24 +77,6 @@
tearDown(test)
-def distroseriesqueueSetUp(test):
- setUp(test)
- # The test requires that the umask be set to 022, and in fact this comment
- # was made in irc on 13-Apr-2007:
- #
- # (04:29:18 PM) kiko: barry, cprov says that the local umask is controlled
- # enough for us to rely on it
- #
- # Setting it here reproduces the environment that the doctest expects.
- # Save the old umask so we can reset it in the tearDown().
- test.old_umask = os.umask(022)
-
-
-def distroseriesqueueTearDown(test):
- os.umask(test.old_umask)
- tearDown(test)
-
-
def uploadQueueSetUp(test):
lobotomize_stevea()
test_dbuser = config.uploadqueue.dbuser
@@ -142,11 +124,6 @@
setUp=setUp, tearDown=tearDown,
layer=LaunchpadZopelessLayer
),
- 'distroseriesqueue-debian-installer.txt': LayeredDocFileSuite(
- '../doc/distroseriesqueue-debian-installer.txt',
- setUp=distroseriesqueueSetUp, tearDown=distroseriesqueueTearDown,
- layer=LaunchpadFunctionalLayer
- ),
'closing-bugs-from-changelogs.txt': LayeredDocFileSuite(
'../doc/closing-bugs-from-changelogs.txt',
setUp=uploadQueueSetUp,
Follow ups
