launchpad-reviewers team mailing list archive

[Gavin Panella <gavin.panella@xxxxxxxxxxxxx>]

Gavin Panella has proposed merging lp:~allenap/launchpad/bug-export-test-import into lp:launchpad.

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~allenap/launchpad/bug-export-test-import/+merge/83209

This adds a --testing flag to scripts/bug-import.py, which automates
step 4.2 in the bug importing instructions [1].

In doing so it also refactors the existing test for the script, and
adds a helper function to lp.testing called run_process(). The latter
is, imo, a better version of run_script(), though they're intended for
much the same thing.

[1] https://dev.launchpad.net/PolicyAndProcess/BugImportHowto

-- 
https://code.launchpad.net/~allenap/launchpad/bug-export-test-import/+merge/83209
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~allenap/launchpad/bug-export-test-import into lp:launchpad.

=== modified file 'lib/lp/bugs/scripts/tests/test_bugimport.py'
--- lib/lp/bugs/scripts/tests/test_bugimport.py	2011-10-05 18:46:05 +0000
+++ lib/lp/bugs/scripts/tests/test_bugimport.py	2011-11-23 18:03:03 +0000
@@ -1,15 +1,14 @@
-# Copyright 2009-2010 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 import os
 import re
 import shutil
-import subprocess
-import sys
 import tempfile
 import unittest
 
 import pytz
+from testtools.content import text_content
 import transaction
 from zope.component import getUtility
 from zope.interface import implements
@@ -17,7 +16,6 @@
 
 from canonical.config import config
 from canonical.database.sqlbase import cursor
-from lp.bugs.model.bugnotification import BugNotification
 from canonical.launchpad.ftests import (
     login,
     logout,
@@ -37,6 +35,7 @@
 from lp.bugs.interfaces.bugtracker import BugTrackerType
 from lp.bugs.interfaces.bugwatch import IBugWatch
 from lp.bugs.interfaces.externalbugtracker import UNKNOWN_REMOTE_IMPORTANCE
+from lp.bugs.model.bugnotification import BugNotification
 from lp.bugs.scripts import bugimport
 from lp.bugs.scripts.bugimport import ET
 from lp.bugs.scripts.checkwatches import (
@@ -50,7 +49,11 @@
     )
 from lp.registry.interfaces.product import IProductSet
 from lp.registry.model.person import generate_nick
-from lp.testing import TestCaseWithFactory
+from lp.testing import (
+    run_process,
+    TestCase,
+    TestCaseWithFactory,
+    )
 
 
 class UtilsTestCase(unittest.TestCase):
@@ -769,53 +772,80 @@
         importer.importBugs(self.layer.txn)
 
 
-class BugImportScriptTestCase(unittest.TestCase):
+class BugImportScriptTestCase(TestCase):
     """Test that the driver script can be called, and does its job."""
+
     layer = LaunchpadZopelessLayer
 
     def setUp(self):
+        super(BugImportScriptTestCase, self).setUp()
         self.tmpdir = tempfile.mkdtemp()
 
     def tearDown(self):
+        super(BugImportScriptTestCase, self).tearDown()
         shutil.rmtree(self.tmpdir, ignore_errors=True)
         # We ran a subprocess that may have changed the database, so
         # force the test system to treat it as dirty.
         self.layer.force_dirty_database()
 
+    def write_example_xml(self):
+        xml_file = os.path.join(self.tmpdir, 'bugs.xml')
+        with open(xml_file, 'w') as fp:
+            ns = "https://launchpad.net/xmlns/2006/bugs";
+            fp.write('<launchpad-bugs xmlns="%s">\n' % ns)
+            fp.write(sample_bug)
+            fp.write('</launchpad-bugs>\n')
+        return xml_file
+
     def test_bug_import_script(self):
-        # Test that the bug import script can do its job
-        xml_file = os.path.join(self.tmpdir, 'bugs.xml')
-        fp = open(xml_file, 'w')
-        fp.write('<launchpad-bugs '
-                 'xmlns="https://launchpad.net/xmlns/2006/bugs";>\n')
-        fp.write(sample_bug)
-        fp.write('</launchpad-bugs>\n')
-        fp.close()
+        # Test that the bug import script can do its job.
+        script = os.path.join(config.root, 'scripts', 'bug-import.py')
         cache_filename = os.path.join(self.tmpdir, 'bug-map.pickle')
-        # Run the bug import script as a subprocess:
-        proc = subprocess.Popen(
-            [sys.executable,
-             os.path.join(config.root, 'scripts', 'bug-import.py'),
-             '--product', 'netapplet',
-             '--cache', cache_filename,
-             xml_file],
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE)
-        output, error = proc.communicate()
-        self.assertEqual(proc.returncode, 0)
+        stdout, stderr, returncode = run_process(
+            (script, '--product', 'netapplet', '--cache', cache_filename,
+             self.write_example_xml()))
+        self.addDetail("stdout", text_content(stdout))
+        self.addDetail("stderr", text_content(stderr))
+        self.assertEqual(0, returncode)
 
         # Find the imported bug number:
-        match = re.search(r'Creating Launchpad bug #(\d+)', error)
+        match = re.search(r'Creating Launchpad bug #(\d+)', stderr)
         self.assertNotEqual(match, None)
         bug_id = int(match.group(1))
 
         # Abort transaction so we can see the result:
-        self.layer.txn.abort()
+        transaction.abort()
         bug = getUtility(IBugSet).get(bug_id)
         self.assertEqual(bug.title, 'A test bug')
         self.assertEqual(bug.bugtasks[0].product.name, 'netapplet')
 
+    def test_bug_import_script_in_testing_mode(self):
+        # Test that the bug import script works with --testing.
+        script = os.path.join(config.root, 'scripts', 'bug-import.py')
+        cache_filename = os.path.join(self.tmpdir, 'bug-map.pickle')
+        stdout, stderr, returncode = run_process(
+            (script, '--testing', '--cache', cache_filename,
+             self.write_example_xml()))
+        self.addDetail("stdout", text_content(stdout))
+        self.addDetail("stderr", text_content(stderr))
+        self.assertEqual(0, returncode)
+
+        # Find the product that was created:
+        match = re.search(r'Product ([^ ]+) created', stderr)
+        self.assertNotEqual(match, None)
+        product_name = match.group(1)
+
+        # Find the imported bug number:
+        match = re.search(r'Creating Launchpad bug #(\d+)', stderr)
+        self.assertNotEqual(match, None)
+        bug_id = int(match.group(1))
+
+        # Abort transaction so we can see the result:
+        transaction.abort()
+        bug = getUtility(IBugSet).get(bug_id)
+        self.assertEqual(bug.title, 'A test bug')
+        self.assertEqual(bug.bugtasks[0].product.name, product_name)
+
 
 class FakeResultSet:
 

=== modified file 'lib/lp/testing/__init__.py'
--- lib/lp/testing/__init__.py	2011-11-21 14:37:25 +0000
+++ lib/lp/testing/__init__.py	2011-11-23 18:03:03 +0000
@@ -35,6 +35,7 @@
     'person_logged_in',
     'quote_jquery_expression',
     'record_statements',
+    'run_process',
     'run_script',
     'run_with_login',
     'run_with_storm_debug',
@@ -1127,6 +1128,30 @@
     return out, err, process.returncode
 
 
+def run_process(cmd, env=None):
+    """Run the given command as a subprocess.
+
+    This differs from `run_script` in that it does not execute via a shell and
+    it explicitly connects stdin to /dev/null so that processes will not be
+    able to hang, waiting for user input.
+
+    :param cmd_line: A command suitable for passing to `subprocess.Popen`.
+    :param env: An optional environment dict. If none is given, the script
+        will get a copy of your present environment. Either way, PYTHONPATH
+        will be removed from it because it will break the script.
+    :return: A 3-tuple of stdout, stderr, and the process' return code.
+    """
+    if env is None:
+        env = os.environ.copy()
+    env.pop('PYTHONPATH', None)
+    with open(os.devnull, "rb") as devnull:
+        process = subprocess.Popen(
+            cmd, stdin=devnull, stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE, env=env)
+        stdout, stderr = process.communicate()
+        return stdout, stderr, process.returncode
+
+
 def normalize_whitespace(string):
     """Replace all sequences of whitespace with a single space."""
     # In Python 2.4, splitting and joining a string to normalize

=== modified file 'scripts/bug-import.py'
--- scripts/bug-import.py	2011-09-26 07:21:53 +0000
+++ scripts/bug-import.py	2011-11-23 18:03:03 +0000
@@ -3,16 +3,18 @@
 # Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import _pythonpath
+
 import logging
 
-# pylint: disable-msg=W0403
-import _pythonpath
+import transaction
 from zope.component import getUtility
 
 from canonical.config import config
 from lp.bugs.scripts.bugimport import BugImporter
 from lp.registry.interfaces.product import IProductSet
 from lp.services.scripts.base import LaunchpadScript
+from lp.testing.factory import LaunchpadObjectFactory
 
 
 class BugImportScript(LaunchpadScript):
@@ -37,10 +39,29 @@
             '--dont-verify-users', dest='verify_users',
             help="Don't verify newly created users", action='store_false',
             default=True)
+        self.parser.add_option(
+            '--testing', dest="testing", action="store_true", help=(
+                "Testing mode; create the product specified with --product "
+                "with the object factory. Do *not* use in production!"),
+            default=False)
+
+    def create_test_product(self):
+        """Create a test product with `LaunchpadObjectFactory`.
+
+        Returns the new object's name.
+        """
+        factory = LaunchpadObjectFactory()
+        product = factory.makeProduct(official_malone=True)
+        transaction.commit()
+        return product.name
 
     def main(self):
         if self.options.product is None:
-            self.parser.error('No product specified')
+            if self.options.testing:
+                self.options.product = self.create_test_product()
+                self.logger.info("Product %s created", self.options.product)
+            else:
+                self.parser.error('No product specified')
         if len(self.args) != 1:
             self.parser.error('Please specify a bug XML file to import')
         bugs_filename = self.args[0]