How to use popen method in yandex-tank

Best Python code snippet using yandex-tank

test_package.py

Source:test_package.py Github

copy

Full Screen

...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 = {}...

Full Screen

Full Screen

acts_job_test.py

Source:acts_job_test.py Github

copy

Full Screen

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__':...

Full Screen

Full Screen

test_easy.py

Source:test_easy.py Github

copy

Full Screen

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...

Full Screen

Full Screen

qnxnto.py

Source:qnxnto.py Github

copy

Full Screen

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...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run yandex-tank automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful