launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #29636
[Merge] ~cjwatson/launchpad:remove-lp-testing-run-script into launchpad:master
Colin Watson has proposed merging ~cjwatson/launchpad:remove-lp-testing-run-script into launchpad:master.
Commit message:
Remove lp.testing.run_script
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/+git/launchpad/+merge/436843
`lp.testing.script.run_script` has a better API, since it expects arguments as a list rather than embedded in a string that's passed to a shell. I made some small adjustments to give us the union of the good bits of both functions.
--
Your team Launchpad code reviewers is requested to review the proposed merge of ~cjwatson/launchpad:remove-lp-testing-run-script into launchpad:master.
diff --git a/lib/lp/answers/tests/test_questionjob.py b/lib/lp/answers/tests/test_questionjob.py
index 214b83a..3c3a7f7 100644
--- a/lib/lp/answers/tests/test_questionjob.py
+++ b/lib/lp/answers/tests/test_questionjob.py
@@ -3,6 +3,8 @@
"""Tests for QuestionJobs classes."""
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -26,9 +28,10 @@ from lp.services.mail import stub
from lp.services.mail.sendmail import format_address, format_address_for_person
from lp.services.scripts import log
from lp.services.worlddata.interfaces.language import ILanguageSet
-from lp.testing import TestCaseWithFactory, person_logged_in, run_script
+from lp.testing import TestCaseWithFactory, person_logged_in
from lp.testing.dbuser import dbuser
from lp.testing.layers import CeleryJobLayer, DatabaseFunctionalLayer
+from lp.testing.script import run_script
class QuestionJobTestCase(TestCaseWithFactory):
@@ -352,9 +355,12 @@ class QuestionEmailJobTestCase(TestCaseWithFactory):
question.target.addAnswerContact(user, user)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IQuestionEmailJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IQuestionEmailJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
self.addDetail("stderr", text_content(err))
@@ -372,7 +378,6 @@ class QuestionEmailJobTestCase(TestCaseWithFactory):
class TestViaCelery(TestCaseWithFactory):
-
layer = CeleryJobLayer
def test_run(self):
diff --git a/lib/lp/archivepublisher/tests/test_publish_ftpmaster.py b/lib/lp/archivepublisher/tests/test_publish_ftpmaster.py
index 5f686ad..770c667 100644
--- a/lib/lp/archivepublisher/tests/test_publish_ftpmaster.py
+++ b/lib/lp/archivepublisher/tests/test_publish_ftpmaster.py
@@ -45,9 +45,10 @@ from lp.soyuz.enums import (
PackageUploadCustomFormat,
)
from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
-from lp.testing import TestCase, TestCaseWithFactory, run_script
+from lp.testing import TestCase, TestCaseWithFactory
from lp.testing.fakemethod import FakeMethod
from lp.testing.layers import LaunchpadZopelessLayer
+from lp.testing.script import run_script
def path_exists(*path_components):
@@ -229,7 +230,9 @@ class TestPublishFTPMasterScript(
def test_script_runs_successfully(self):
self.prepareUbuntu()
self.layer.txn.commit()
- stdout, stderr, retval = run_script(self.SCRIPT_PATH + " -d ubuntu")
+ retval, stdout, stderr = run_script(
+ self.SCRIPT_PATH, args=["-d", "ubuntu"]
+ )
self.assertEqual(0, retval, "Script failure:\n" + stderr)
def test_getConfigs_maps_distro_and_purpose_to_matching_config(self):
diff --git a/lib/lp/codehosting/scripts/tests/test_upgrade_all_branches.py b/lib/lp/codehosting/scripts/tests/test_upgrade_all_branches.py
index 87752a6..fe9590e 100644
--- a/lib/lp/codehosting/scripts/tests/test_upgrade_all_branches.py
+++ b/lib/lp/codehosting/scripts/tests/test_upgrade_all_branches.py
@@ -12,8 +12,9 @@ from fixtures import TempDir
from lp.code.bzr import branch_changed
from lp.codehosting.upgrade import Upgrader
from lp.services.config import config
-from lp.testing import TestCaseWithFactory, person_logged_in, run_script
+from lp.testing import TestCaseWithFactory, person_logged_in
from lp.testing.layers import AppServerLayer
+from lp.testing.script import run_script
class TestUpgradeAllBranchesScript(TestCaseWithFactory):
@@ -29,11 +30,13 @@ class TestUpgradeAllBranchesScript(TestCaseWithFactory):
"""Run the script to upgrade all branches."""
transaction.commit()
if finish:
- flags = " --finish "
+ flags = ["--finish"]
else:
- flags = " "
+ flags = []
return run_script(
- "scripts/upgrade_all_branches.py" + flags + target, cwd=self.cwd
+ "scripts/upgrade_all_branches.py",
+ args=flags + [target],
+ cwd=self.cwd,
)
def prepare(self):
@@ -52,7 +55,7 @@ class TestUpgradeAllBranchesScript(TestCaseWithFactory):
def test_start_upgrade(self):
"""Test that starting the upgrade behaves as expected."""
upgrader = self.prepare()
- stdout, stderr, retcode = self.upgrade_all_branches(
+ retcode, stdout, stderr = self.upgrade_all_branches(
upgrader.target_dir
)
self.assertIn(
@@ -68,7 +71,7 @@ class TestUpgradeAllBranchesScript(TestCaseWithFactory):
"""Test that finishing the upgrade behaves as expected."""
upgrader = self.prepare()
upgrader.start_upgrade()
- stdout, stderr, retcode = self.upgrade_all_branches(
+ retcode, stdout, stderr = self.upgrade_all_branches(
upgrader.target_dir, finish=True
)
self.assertIn(
diff --git a/lib/lp/registry/tests/test_distributionmirror_prober.py b/lib/lp/registry/tests/test_distributionmirror_prober.py
index 8aba075..5925995 100644
--- a/lib/lp/registry/tests/test_distributionmirror_prober.py
+++ b/lib/lp/registry/tests/test_distributionmirror_prober.py
@@ -84,13 +84,13 @@ from lp.testing import (
TestCaseWithFactory,
admin_logged_in,
clean_up_reactor,
- run_script,
)
from lp.testing.layers import (
LaunchpadZopelessLayer,
TwistedLayer,
ZopelessDatabaseLayer,
)
+from lp.testing.script import run_script
class HTTPServerTestSetup(TacTestSetup):
@@ -1324,9 +1324,14 @@ class TestDistroMirrorProberFunctional(TestCaseWithFactory):
mirror = self.makeMirror(content_type=MirrorContent.RELEASE)
transaction.commit()
- out, err, exit_code = run_script(
- "cronscripts/distributionmirror-prober.py --no-remote-hosts "
- "--content-type=cdimage --no-owner-notification --force"
+ exit_code, out, err = run_script(
+ "cronscripts/distributionmirror-prober.py",
+ args=[
+ "--no-remote-hosts",
+ "--content-type=cdimage",
+ "--no-owner-notification",
+ "--force",
+ ],
)
self.assertEqual(0, exit_code, err)
@@ -1376,9 +1381,14 @@ class TestDistroMirrorProberFunctional(TestCaseWithFactory):
)
transaction.commit()
- out, err, exit_code = run_script(
- "cronscripts/distributionmirror-prober.py --no-remote-hosts "
- "--content-type=archive --no-owner-notification --force"
+ exit_code, out, err = run_script(
+ "cronscripts/distributionmirror-prober.py",
+ args=[
+ "--no-remote-hosts",
+ "--content-type=archive",
+ "--no-owner-notification",
+ "--force",
+ ],
)
self.assertEqual(0, exit_code, err)
diff --git a/lib/lp/registry/tests/test_membership_notification_job.py b/lib/lp/registry/tests/test_membership_notification_job.py
index ef9d7e4..9becfe6 100644
--- a/lib/lp/registry/tests/test_membership_notification_job.py
+++ b/lib/lp/registry/tests/test_membership_notification_job.py
@@ -3,6 +3,8 @@
"""Tests of `MembershipNotificationJob`."""
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -19,14 +21,10 @@ from lp.registry.model.persontransferjob import MembershipNotificationJob
from lp.services.features.testing import FeatureFixture
from lp.services.job.interfaces.job import JobStatus
from lp.services.job.tests import block_on_job, pop_remote_notifications
-from lp.testing import (
- TestCaseWithFactory,
- login_person,
- person_logged_in,
- run_script,
-)
+from lp.testing import TestCaseWithFactory, login_person, person_logged_in
from lp.testing.layers import CeleryJobLayer, DatabaseFunctionalLayer
from lp.testing.sampledata import ADMIN_EMAIL
+from lp.testing.script import run_script
class MembershipNotificationJobTest(TestCaseWithFactory):
@@ -116,9 +114,12 @@ class MembershipNotificationJobTest(TestCaseWithFactory):
)
job_repr = repr(job)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IMembershipNotificationJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IMembershipNotificationJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
self.addDetail("stderr", text_content(err))
diff --git a/lib/lp/registry/tests/test_person_merge_job.py b/lib/lp/registry/tests/test_person_merge_job.py
index 1a39e6b..c536471 100644
--- a/lib/lp/registry/tests/test_person_merge_job.py
+++ b/lib/lp/registry/tests/test_person_merge_job.py
@@ -3,6 +3,8 @@
"""Tests of `PersonMergeJob`."""
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -23,9 +25,10 @@ from lp.services.job.tests import block_on_job
from lp.services.log.logger import BufferLogger
from lp.services.mail.sendmail import format_address_for_person
from lp.services.scripts import log
-from lp.testing import TestCaseWithFactory, person_logged_in, run_script
+from lp.testing import TestCaseWithFactory, person_logged_in
from lp.testing.dbuser import dbuser
from lp.testing.layers import CeleryJobLayer, DatabaseFunctionalLayer
+from lp.testing.script import run_script
def create_job(factory):
@@ -141,9 +144,12 @@ class TestPersonMergeJob(TestCaseWithFactory):
)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IPersonMergeJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IPersonMergeJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
diff --git a/lib/lp/registry/tests/test_productjob.py b/lib/lp/registry/tests/test_productjob.py
index c6e2961..3927271 100644
--- a/lib/lp/registry/tests/test_productjob.py
+++ b/lib/lp/registry/tests/test_productjob.py
@@ -3,6 +3,7 @@
"""Tests for ProductJobs."""
+import os
from datetime import datetime, timedelta
import pytz
@@ -47,13 +48,14 @@ from lp.services.job.interfaces.job import JobStatus
from lp.services.log.logger import BufferLogger
from lp.services.propertycache import clear_property_cache
from lp.services.webapp.publisher import canonical_url
-from lp.testing import TestCaseWithFactory, person_logged_in, run_script
+from lp.testing import TestCaseWithFactory, person_logged_in
from lp.testing.layers import (
DatabaseFunctionalLayer,
LaunchpadZopelessLayer,
ZopelessAppServerLayer,
)
from lp.testing.mail_helpers import pop_notifications
+from lp.testing.script import run_script
class CommercialHelpers:
@@ -151,7 +153,7 @@ class DailyProductJobsTestCase(TestCaseWithFactory, CommercialHelpers):
# ProductJobManagerTestCase.test_createAllDailyJobs
self.make_test_products()
transaction.commit()
- stdout, stderr, retcode = run_script(
+ retcode, stdout, stderr = run_script(
"cronscripts/daily_product_jobs.py"
)
self.addDetail("stdout", text_content(stdout))
@@ -622,9 +624,12 @@ class CommericialExpirationMixin(CommercialHelpers):
proprietary_job = self.JOB_CLASS.create(proprietary_product, reviewer)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % self.JOB_SOURCE_INTERFACE.getName()
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", self.JOB_SOURCE_INTERFACE.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
self.addDetail("stderr", text_content(err))
diff --git a/lib/lp/registry/tests/test_sharingjob.py b/lib/lp/registry/tests/test_sharingjob.py
index e017e72..a088b02 100644
--- a/lib/lp/registry/tests/test_sharingjob.py
+++ b/lib/lp/registry/tests/test_sharingjob.py
@@ -3,6 +3,8 @@
"""Tests for SharingJobs."""
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -42,17 +44,13 @@ from lp.services.job.interfaces.job import JobStatus
from lp.services.job.tests import block_on_job
from lp.services.mail.sendmail import format_address_for_person
from lp.snappy.interfaces.snap import SNAP_TESTING_FLAGS
-from lp.testing import (
- TestCaseWithFactory,
- login_person,
- person_logged_in,
- run_script,
-)
+from lp.testing import TestCaseWithFactory, login_person, person_logged_in
from lp.testing.layers import (
CeleryJobLayer,
DatabaseFunctionalLayer,
LaunchpadZopelessLayer,
)
+from lp.testing.script import run_script
class SharingJobTestCase(TestCaseWithFactory):
@@ -266,9 +264,12 @@ class TestRunViaCron(TestCaseWithFactory):
)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (job_type)
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", job_type],
+ env=env,
)
self.addDetail("stdout", text_content(out))
self.addDetail("stderr", text_content(err))
diff --git a/lib/lp/services/scripts/tests/test_all_scripts.py b/lib/lp/services/scripts/tests/test_all_scripts.py
index fc3be73..a5e770c 100644
--- a/lib/lp/services/scripts/tests/test_all_scripts.py
+++ b/lib/lp/services/scripts/tests/test_all_scripts.py
@@ -10,7 +10,8 @@ from testscenarios import WithScenarios, load_tests_apply_scenarios
from testtools.matchers import DocTestMatches
from lp.services.scripts.tests import find_lp_scripts
-from lp.testing import TestCase, run_script
+from lp.testing import TestCase
+from lp.testing.script import run_script
def make_id(script_path):
@@ -27,8 +28,7 @@ class ScriptsTestCase(WithScenarios, TestCase):
def test_script(self):
# Run self.script_path with '-h' to make sure it runs cleanly.
- cmd_line = self.script_path + " -h"
- out, err, returncode = run_script(cmd_line)
+ returncode, out, err = run_script(self.script_path, args=["-h"])
self.assertThat(err, DocTestMatches("", doctest.REPORT_NDIFF))
self.assertEqual("", err)
self.assertEqual(os.EX_OK, returncode)
diff --git a/lib/lp/soyuz/tests/test_packagecopyjob.py b/lib/lp/soyuz/tests/test_packagecopyjob.py
index 645683b..b16350f 100644
--- a/lib/lp/soyuz/tests/test_packagecopyjob.py
+++ b/lib/lp/soyuz/tests/test_packagecopyjob.py
@@ -62,7 +62,6 @@ from lp.testing import (
TestCaseWithFactory,
admin_logged_in,
person_logged_in,
- run_script,
verifyObject,
)
from lp.testing.dbuser import dbuser, switch_dbuser
@@ -75,6 +74,7 @@ from lp.testing.layers import (
)
from lp.testing.mail_helpers import pop_notifications
from lp.testing.matchers import Provides
+from lp.testing.script import run_script
def get_dsd_comments(dsd):
@@ -725,9 +725,12 @@ class PlainPackageCopyJobTests(TestCaseWithFactory, LocalTestHelper):
archive2.newComponentUploader(requester, "main")
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IPlainPackageCopyJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IPlainPackageCopyJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
diff --git a/lib/lp/soyuz/tests/test_packagediffjob.py b/lib/lp/soyuz/tests/test_packagediffjob.py
index b3d9b2c..54c5a65 100644
--- a/lib/lp/soyuz/tests/test_packagediffjob.py
+++ b/lib/lp/soyuz/tests/test_packagediffjob.py
@@ -1,6 +1,8 @@
# Copyright 2013-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -18,9 +20,10 @@ from lp.soyuz.interfaces.packagediffjob import (
)
from lp.soyuz.model.packagediffjob import PackageDiffJob
from lp.soyuz.tests.test_packagediff import create_proper_job
-from lp.testing import TestCaseWithFactory, run_script, verifyObject
+from lp.testing import TestCaseWithFactory, verifyObject
from lp.testing.fakemethod import FakeMethod
from lp.testing.layers import CeleryJobLayer, LaunchpadZopelessLayer
+from lp.testing.script import run_script
class TestPackageDiffJob(TestCaseWithFactory):
@@ -76,9 +79,12 @@ class TestPackageDiffJob(TestCaseWithFactory):
def test_smoke(self):
diff = create_proper_job(self.factory)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IPackageDiffJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IPackageDiffJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
diff --git a/lib/lp/soyuz/tests/test_packagetranslationsuploadjob.py b/lib/lp/soyuz/tests/test_packagetranslationsuploadjob.py
index 6564fe4..650f39d 100644
--- a/lib/lp/soyuz/tests/test_packagetranslationsuploadjob.py
+++ b/lib/lp/soyuz/tests/test_packagetranslationsuploadjob.py
@@ -1,6 +1,8 @@
# Copyright 2013-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
+import os
+
import transaction
from testtools.content import text_content
from zope.component import getUtility
@@ -18,15 +20,11 @@ from lp.soyuz.interfaces.packagetranslationsuploadjob import (
from lp.soyuz.model.packagetranslationsuploadjob import (
PackageTranslationsUploadJob,
)
-from lp.testing import (
- TestCaseWithFactory,
- person_logged_in,
- run_script,
- verifyObject,
-)
+from lp.testing import TestCaseWithFactory, person_logged_in, verifyObject
from lp.testing.dbuser import dbuser
from lp.testing.fakemethod import FakeMethod
from lp.testing.layers import CeleryJobLayer, LaunchpadZopelessLayer
+from lp.testing.script import run_script
from lp.translations.interfaces.translationimportqueue import (
ITranslationImportQueue,
)
@@ -132,9 +130,12 @@ class TestPackageTranslationsUploadJob(LocalTestHelper):
}
spr, sp, job = self.makeJob(tar_content=tar_content)
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IPackageTranslationsUploadJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IPackageTranslationsUploadJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
diff --git a/lib/lp/soyuz/tests/test_processacceptedbugsjob.py b/lib/lp/soyuz/tests/test_processacceptedbugsjob.py
index 6ed8b1d..f2e29bf 100644
--- a/lib/lp/soyuz/tests/test_processacceptedbugsjob.py
+++ b/lib/lp/soyuz/tests/test_processacceptedbugsjob.py
@@ -4,6 +4,7 @@
"""Tests for jobs to close bugs for accepted package uploads."""
import io
+import os
from itertools import product
from textwrap import dedent
@@ -38,7 +39,6 @@ from lp.testing import (
TestCaseWithFactory,
celebrity_logged_in,
person_logged_in,
- run_script,
verifyObject,
)
from lp.testing.fakemethod import FakeMethod
@@ -47,6 +47,7 @@ from lp.testing.layers import (
DatabaseFunctionalLayer,
LaunchpadZopelessLayer,
)
+from lp.testing.script import run_script
class TestBugIDsFromChangesFile(TestCaseWithFactory):
@@ -447,9 +448,12 @@ class TestProcessAcceptedBugsJob(TestCaseWithFactory):
self.makeJob(spr=spr, bug_ids=[bug.id])
transaction.commit()
- out, err, exit_code = run_script(
- "LP_DEBUG_SQL=1 cronscripts/process-job-source.py -vv %s"
- % (IProcessAcceptedBugsJobSource.getName())
+ env = os.environ.copy()
+ env["LP_DEBUG_SQL"] = "1"
+ exit_code, out, err = run_script(
+ "cronscripts/process-job-source.py",
+ args=["-vv", IProcessAcceptedBugsJobSource.getName()],
+ env=env,
)
self.addDetail("stdout", text_content(out))
diff --git a/lib/lp/testing/__init__.py b/lib/lp/testing/__init__.py
index 4b78f9a..d1cf6fd 100644
--- a/lib/lp/testing/__init__.py
+++ b/lib/lp/testing/__init__.py
@@ -33,7 +33,6 @@ __all__ = [
"record_statements",
"reset_logging",
"run_process",
- "run_script",
"run_with_login",
"run_with_storm_debug",
"RunIsolatedTest",
@@ -1367,42 +1366,11 @@ def time_counter(origin=None, delta=timedelta(seconds=5)):
now += delta
-def run_script(cmd_line, env=None, cwd=None, universal_newlines=True):
- """Run the given command line as a subprocess.
-
- :param cmd_line: A command line 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.
- :param universal_newlines: If True, return stdout and stderr as text.
- Defaults to True.
- :return: A 3-tuple of stdout, stderr, and the process' return code.
- """
- if env is None:
- env = os.environ.copy()
- env.pop("PYTHONPATH", None)
- process = subprocess.Popen(
- cmd_line,
- shell=True,
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- env=env,
- cwd=cwd,
- universal_newlines=universal_newlines,
- )
- (out, err) = process.communicate()
- return out, err, process.returncode
-
-
def run_process(cmd, env=None, universal_newlines=True):
"""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.
+ This 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
diff --git a/lib/lp/testing/script.py b/lib/lp/testing/script.py
index ee010dc..4f4e5e9 100644
--- a/lib/lp/testing/script.py
+++ b/lib/lp/testing/script.py
@@ -8,16 +8,28 @@ __all__ = [
"run_script",
]
+import os
import subprocess
-import sys
+from typing import Dict, List, Optional, Union
-def run_command(command, args=None, input=None, universal_newlines=True):
+# XXX cjwatson 2023-02-03: This type could be stricter: the type of `input`
+# depends on the value of `universal_newlines`.
+def run_command(
+ command: str,
+ args: Optional[List[str]] = None,
+ env: Optional[Dict[str, str]] = None,
+ cwd: Optional[str] = None,
+ input: Optional[Union[bytes, str]] = None,
+ universal_newlines: bool = True,
+):
"""Run an external command in a separate process.
:param command: executable to run.
:param args: optional list of command-line arguments.
:param input: optional text to feed to command's standard input.
+ :param env: optional, passed to `subprocess.Popen`.
+ :param cwd: optional, passed to `subprocess.Popen`.
:param universal_newlines: passed to `subprocess.Popen`, defaulting to
True.
:return: tuple of returncode, standard output, and standard error.
@@ -35,6 +47,8 @@ def run_command(command, args=None, input=None, universal_newlines=True):
stdin=stdin,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
+ env=env,
+ cwd=cwd,
universal_newlines=universal_newlines,
)
stdout, stderr = child.communicate(input)
@@ -42,23 +56,40 @@ def run_command(command, args=None, input=None, universal_newlines=True):
return (returncode, stdout, stderr)
-def run_script(script, args=None, input=None, universal_newlines=True):
+# XXX cjwatson 2023-02-03: This type could be stricter: the type of `input`
+# depends on the value of `universal_newlines`.
+def run_script(
+ script: str,
+ args: Optional[List[str]] = None,
+ env: Optional[Dict[str, str]] = None,
+ cwd: Optional[str] = None,
+ input: Optional[Union[bytes, str]] = None,
+ universal_newlines: bool = True,
+):
"""Run a Python script in a child process, using current interpreter.
:param script: Python script to run.
:param args: optional list of command-line arguments.
+ :param env: optional environment dict; if none is given, the script will
+ get a copy of the environment of the calling process. In either
+ case, `PYTHONPATH` is removed since inheriting it may break some
+ scripts.
+ :param env: optional, passed to `subprocess.Popen`.
+ :param cwd: optional, passed to `subprocess.Popen`.
:param input: optional string to feed to standard input.
:param universal_newlines: passed to `subprocess.Popen`, defaulting to
True.
:return: tuple of return value, standard output, and standard error.
"""
- interpreter_args = [script]
- if args:
- interpreter_args.extend(args)
+ if env is None:
+ env = os.environ.copy()
+ env.pop("PYTHONPATH", None)
return run_command(
- sys.executable,
- interpreter_args,
- input,
+ script,
+ args=args,
+ env=env,
+ cwd=cwd,
+ input=input,
universal_newlines=universal_newlines,
)