Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use popen method in yandex-tank

Best Python code snippet using yandex-tank

test_package.py

Source:test_package.py

...29def lib_dir(deps_dir):30    """Return path to lib directory."""31    return os.path.join(deps_dir, "lib_dir")32@pytest.fixture33def mock_popen(lib_dir):34    """Return a Popen mock."""35    with patch("homeassistant.util.package.Popen") as popen_mock:36        popen_mock.return_value.__enter__ = popen_mock37        popen_mock.return_value.communicate.return_value = (38            bytes(lib_dir, "utf-8"),39            b"error",40        )41        popen_mock.return_value.returncode = 042        yield popen_mock43@pytest.fixture44def mock_env_copy():45    """Mock os.environ.copy."""46    with patch("homeassistant.util.package.os.environ.copy") as env_copy:47        env_copy.return_value = {}...

acts_job_test.py

Source:acts_job_test.py

1#!/usr/bin/env python32# Copyright 2016 - The Android Open Source Project3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#     http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15# Import the python3 compatible bytes()16from builtins import bytes17import mock18import os19import sys20import unittest21from acts.libs.proc import job22if os.name == 'posix' and sys.version_info[0] < 3:23    import subprocess32 as subprocess24else:25    import subprocess26class FakePopen(object):27    """A fake version of the object returned from subprocess.Popen()."""28    def __init__(self,29                 stdout=None,30                 stderr=None,31                 returncode=0,32                 will_timeout=False):33        self.returncode = returncode34        self._stdout = bytes(stdout,35                             'utf-8') if stdout is not None else bytes()36        self._stderr = bytes(stderr,37                             'utf-8') if stderr is not None else bytes()38        self._will_timeout = will_timeout39    def communicate(self, timeout=None):40        if self._will_timeout:41            raise subprocess.TimeoutExpired(42                -1, 'Timed out according to test logic')43        return self._stdout, self._stderr44    def kill(self):45        pass46    def wait(self):47        pass48class JobTestCases(unittest.TestCase):49    @mock.patch(50        'acts.libs.proc.job.subprocess.Popen',51        return_value=FakePopen(stdout='TEST\n'))52    def test_run_success(self, popen):53        """Test running a simple shell command."""54        result = job.run('echo TEST')55        self.assertTrue(result.stdout.startswith('TEST'))56    @mock.patch(57        'acts.libs.proc.job.subprocess.Popen',58        return_value=FakePopen(stderr='TEST\n'))59    def test_run_stderr(self, popen):60        """Test that we can read process stderr."""61        result = job.run('echo TEST 1>&2')62        self.assertEqual(len(result.stdout), 0)63        self.assertTrue(result.stderr.startswith('TEST'))64        self.assertFalse(result.stdout)65    @mock.patch(66        'acts.libs.proc.job.subprocess.Popen',67        return_value=FakePopen(returncode=1))68    def test_run_error(self, popen):69        """Test that we raise on non-zero exit statuses."""70        self.assertRaises(job.Error, job.run, 'exit 1')71    @mock.patch(72        'acts.libs.proc.job.subprocess.Popen',73        return_value=FakePopen(returncode=1))74    def test_run_with_ignored_error(self, popen):75        """Test that we can ignore exit status on request."""76        result = job.run('exit 1', ignore_status=True)77        self.assertEqual(result.exit_status, 1)78    @mock.patch(79        'acts.libs.proc.job.subprocess.Popen',80        return_value=FakePopen(will_timeout=True))81    def test_run_timeout(self, popen):82        """Test that we correctly implement command timeouts."""83        self.assertRaises(job.Error, job.run, 'sleep 5', timeout=0.1)84    @mock.patch(85        'acts.libs.proc.job.subprocess.Popen',86        return_value=FakePopen(stdout='TEST\n'))87    def test_run_no_shell(self, popen):88        """Test that we handle running without a wrapping shell."""89        result = job.run(['echo', 'TEST'])90        self.assertTrue(result.stdout.startswith('TEST'))91    @mock.patch(92        'acts.libs.proc.job.subprocess.Popen',93        return_value=FakePopen(stdout='TEST\n'))94    def test_job_env(self, popen):95        """Test that we can set environment variables correctly."""96        test_env = {'MYTESTVAR': '20'}97        result = job.run('printenv', env=test_env.copy())98        popen.assert_called_once()99        _, kwargs = popen.call_args100        self.assertTrue('env' in kwargs)101        self.assertEqual(kwargs['env'], test_env)102if __name__ == '__main__':...

test_easy.py

Source:test_easy.py

1from paver import easy2from paver.tests.mock import patch, Mock3import subprocess # for easy.sh tests4@patch(subprocess, "Popen")5@patch(easy, "error")6def test_sh_raises_BuildFailure(popen, error):7    popen.return_value = Mock()8    popen.return_value.returncode = 19    popen.return_value.stderr.read.return_value = 'some stderr'10    try:11        easy.sh('foo')12    except easy.BuildFailure, e:13        args = e.args14        assert args == ('Subprocess return code: 1', )15    else:16        assert False, 'Failed to raise BuildFailure'17    assert popen.called18    assert popen.call_args[0][0] == 'foo'19    assert popen.call_args[1]['shell'] == True20    assert 'stdout' not in popen.call_args[1]21    assert error.called22    assert error.call_args == (('some stderr', ), {})23@patch(subprocess, "Popen")24def test_sh_with_capture_raises_BuildFailure(popen):25    popen.return_value = Mock()26    popen.return_value.returncode = 127    popen.return_value.stderr.read.return_value = 'some stderr'28    try:29        easy.sh('foo', capture=True)30    except easy.BuildFailure, e:31        args = e.args32        assert args == ('Subprocess return code: 1', )33    else:34        assert False, 'Failed to raise BuildFailure'35    assert popen.called36    assert popen.call_args[0][0] == 'foo'37    assert popen.call_args[1]['shell'] == True38    assert popen.call_args[1]['stdout'] == subprocess.PIPE39    assert popen.call_args[1]['stderr'] == subprocess.PIPE40@patch(subprocess, "Popen")41def test_sh_ignores_error(popen):42    popen.return_value = Mock()43    popen.return_value.returncode = 144    popen.return_value.stderr.read.return_value = 'some stderr'45    easy.sh('foo', ignore_error=True)46    assert popen.called47    assert popen.call_args[0][0] == 'foo'48    assert popen.call_args[1]['shell'] == True49    assert 'stdout' not in popen.call_args[1]50    assert popen.call_args[1]['stderr'] == subprocess.PIPE51@patch(subprocess, "Popen")52def test_sh_ignores_error_with_capture(popen):53    popen.return_value = Mock()54    popen.return_value.returncode = 155    popen.return_value.stderr.read.return_value = 'some stderr'56    easy.sh('foo', capture=True, ignore_error=True)57    assert popen.called58    assert popen.call_args[0][0] == 'foo'59    assert popen.call_args[1]['shell'] == True60    assert popen.call_args[1]['stdout'] == subprocess.PIPE...

qnxnto.py

Source:qnxnto.py

1#!/usr/bin/env python2# encoding: utf-83# JÃ©rÃ´me Carretero 2011 (zougloub)4# QNX neutrino compatibility functions5import sys, os6from waflib import Utils7class Popen(object):8	"""9	Popen cannot work on QNX from a threaded program:10	Forking in threads is not implemented in neutrino.11	Python's os.popen / spawn / fork won't work when running in threads (they will if in the main program thread)12	In waf, this happens mostly in build.13	And the use cases can be replaced by os.system() calls.14	"""15	__slots__ = ["prog", "kw", "popen", "verbose"]16	verbose = 017	def __init__(self, prog, **kw):18		try:19			self.prog = prog20			self.kw = kw21			self.popen = None22			if Popen.verbose:23				sys.stdout.write("Popen created: %r, kw=%r..." % (prog, kw))24			do_delegate = kw.get('stdout') == -1 and kw.get('stderr') == -125			if do_delegate:26				if Popen.verbose:27					print("Delegating to real Popen")28				self.popen = self.real_Popen(prog, **kw)29			else:30				if Popen.verbose:31					print("Emulating")32		except Exception as e:33			if Popen.verbose:34				print("Exception: %s" % e)35			raise36	def __getattr__(self, name):37		if Popen.verbose:38			sys.stdout.write("Getattr: %s..." % name)39		if name in Popen.__slots__:40			return object.__getattribute__(self, name)41		else:42			if self.popen is not None:43				if Popen.verbose:44					print("from Popen")45				return getattr(self.popen, name)46			else:47				if name == "wait":48					return self.emu_wait49				else:50					raise Exception("subprocess emulation: not implemented: %s" % name)51	def emu_wait(self):52		if Popen.verbose:53			print("emulated wait (%r kw=%r)" % (self.prog, self.kw))54		if isinstance(self.prog, str):55			cmd = self.prog56		else:57			cmd = " ".join(self.prog)58		if 'cwd' in self.kw:59			cmd = 'cd "%s" && %s' % (self.kw['cwd'], cmd)60		return os.system(cmd)61if sys.platform == "qnx6":62	Popen.real_Popen = Utils.subprocess.Popen...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.