KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use __popen method in yandex-tank

Best Python code snippet using yandex-tank

advanced.py

Source:advanced.py

1#2#    t.py: utility for contest problem development3#    Copyright (C) 2009-2017 Oleg Davydov4#5#    This program is free software; you can redistribute it and/or modify6#    it under the terms of the GNU General Public License as published by7#    the Free Software Foundation; either version 2 of the License, or8#    (at your option) any later version.9#10#    This program is distributed in the hope that it will be useful,11#    but WITHOUT ANY WARRANTY; without even the implied warranty of12#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the13#    GNU General Public License for more details.14#15#    You should have received a copy of the GNU General Public License along16#    with this program; if not, write to the Free Software Foundation, Inc.,17#    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.18#19import os20import time21import resource22import subprocess23from tlib import Log, Module24from .common import *25class Process:26    class CONTINUE:27        pass28    def __init__ ( self,29        command, *,30        directory=None,31        stdin=None, stdout=None, stderr=None,32        limit_time=None, limit_idle=None, limit_memory=None33    ):34        self.__start = time.time ()35        if type (stdin) is str:36            stdin = open (stdin, "rb")37        if type (stdout) is str:38            stdout = open (stdout, "wb")39        if type (stderr) is str:40            stderr = open (stderr, "wb")41        self.__popen = subprocess.Popen (42            command, cwd=directory, stdin=stdin, stdout=stdout, stderr=stderr43        )44        self.__pid = self.__popen.pid45        self.__limit_time = limit_time46        self.__limit_idle = limit_idle47        self.__limit_memory = limit_memory48        self.__usage_time = 0.049        self.__usage_idle = 0.050        self.__usage_memory = 051        self.__peak_memory = 052    usage_time = property (lambda self: self.__usage_time)53    usage_memory = property (lambda self: self.__usage_memory)54    peak_memory = property (lambda self: self.__peak_memory)55    def peaks ( self ):56        return {57            'peak_time': self.__usage_time,58            'peak_memory': self.__peak_memory59        }60    def check ( self ):61        code = self.__popen.returncode62        if code is not None:63            return code64        try:  # ÑÐ°Ðº Ð¼Ð¾Ð¶ÐµÑ ÑÐ»ÑÑÐ¸ÑÑÑÑ, ÑÑÐ¾ Ð¿ÑÐ¾ÑÐµÑÑ Ð·Ð°Ð²ÐµÑÑÐ¸ÑÑÑ Ð² ÑÐ°Ð¼ÑÐ¹ Ð¸Ð½ÑÐµÑÐµÑÐ½ÑÐ¹ Ð¼Ð¾Ð¼ÐµÐ½Ñ65            with open("/proc/%d/stat" % self.__pid, 'r') as f:66                stats = f.readline ().split ()67            with open("/proc/%d/statm" % self.__pid, 'r') as f:68                stats_m = f.readline ().split ()69            self.__usage_time = (int (stats[13]) + int (stats[14])) / \70                os.sysconf (os.sysconf_names['SC_CLK_TCK'])71            self.__usage_idle = time.time () - self.__start72            self.__usage_memory = int (stats_m[0]) * 102473            self.__peak_memory = max (self.__peak_memory, self.__usage_memory)74            if self.__limit_time is not None and self.__usage_time > self.__limit_time:75                return RunResult.limitTime ('cpu usage: %.2f' % self.__usage_time, **self.peaks ())76                self.__popen.kill ()77            if self.__limit_idle is not None and self.__usage_idle > self.__limit_idle:78                return RunResult.limitIdle ('time usage: %.2f' % self.__usage_idle, **self.peaks ())79                self.__popen.kill ()80            if self.__limit_memory is not None and self.__usage_memory > self.__limit_memory:81                return RunResult.limitMemory ('memory usage: %d' % self.__usage_memory, **self.peaks ())82                self.__popen.kill ()83            return Process.CONTINUE84        except IOError:85            return None86    def kill ( self ):87        return self.__popen.kill ()88    def communicate ( self, *args, **kwargs ):89        return self.__popen.communicate (*args, **kwargs)90class Runner (Module):91    def __init__ ( self, *, t ):92        super (Runner, self).__init__ (t=t)93        resource.setrlimit (resource.RLIMIT_STACK, (-1, -1))  # set unlimited stack size94        self._log.warning ("advanced runner doesn\'t support security limitations")95    def run (self,96        command, *,97        interactor=None,98        wait=True,99        verbose=False,100        stdin=None, stdout=None, stderr=None,101        **kwargs102    ):103        if stdin is None:104            stdin = subprocess.DEVNULL105        if stdout is None and self._log.policy is Log.BRIEF:106            stdout = subprocess.DEVNULL107        if stderr is None and self._log.policy is Log.BRIEF:108            stderr = subprocess.DEVNULL109        process = Process (command, stdin=stdin, stdout=stdout, stderr=stderr, **kwargs)110        if not wait:111            return process112        outputs = None113        try:114            time_cpu = 0.0115            while True:116                result_process = process.check ()117                if result_process is None:118                    continue119                if isinstance (result_process, (RunResult, int)):120                    break121                if interactor is not None:122                    result_interactor = interactor.check ()123                    if result_interactor is None:124                        continue125                    # self._log.debug ("interactor result: ", result_interactor)126                    if isinstance (result_interactor, (RunResult, int)):127                        process.kill ()128                if verbose:129                    line = "%.3fs, %.2fMiB" % (process.usage_time, process.usage_memory / 2**20)130                    line = line + '\b' * len (line)131                    self._t.log (line, prefix=False, end='')132                try:133                    outputs = process.communicate (timeout=0.01)134                except subprocess.TimeoutExpired:135                    pass136                if interactor is not None:137                    try:138                        # self._log.debug ("interactor.communicate")139                        interactor.communicate (timeout=0)140                    except subprocess.TimeoutExpired:141                        pass142            if interactor is not None:143                interactor.communicate ()144                result_interactor = interactor.check ()145        finally:146            if interactor is not None:147                interactor.kill ()148            process.kill ()149        150        if outputs is None:151            outputs = process.communicate ()152        if isinstance (result_process, int) and result_process:153            result_process = RunResult.runtime (result_process, 'exit code: %d' % result_process, outputs=outputs, **process.peaks ())154        if isinstance (result_process, int) and result_process == 0:155            result_process = RunResult.ok (outputs=outputs, **process.peaks ())156        if verbose:157            line = "[%.3fs, %.2fMiB] " % (process.usage_time, process.peak_memory / 2**20)158            self._t.log (line, prefix=False, end='')159        assert type (result_process) is RunResult160        if interactor is not None:161            if isinstance (result_interactor, int) and result_interactor:162                result_interactor = RunResult.runtime (result_interactor, 'exit code: %d' % result_interactor, **interactor.peaks ())163            if isinstance (result_interactor, int) and result_interactor == 0:164                result_interactor = RunResult.ok (**interactor.peaks ())165            assert type (result_interactor) is RunResult166            return (result_interactor, result_process)...

test_builder.py

Source:test_builder.py

1# -*- coding: utf-8 -*-2import os3import unittest4try:5    import catkin.builder6except ImportError as e:7    raise ImportError(8        'Please adjust your pythonpath before running this test: %s' % str(e)9    )10class BuilderTest(unittest.TestCase):11    # TODO: Add tests for catkin_make and catkin_make_isolated12    def test_run_command_unicode_string(self):13        backup_Popen = catkin.builder.subprocess.Popen14        class StdOut(object):15            def __init__(self, popen):16                self.__popen = popen17            def readline(self):18                self.__popen.returncode = 019                try:20                    # for Python 2 compatibility only21                    return unichr(2018)22                except NameError:23                    return chr(2018)24        class MockPopen(object):25            def __init__(self, *args, **kwargs):26                self.returncode = None27                self.stdout = StdOut(self)28            def wait(self):29                return True30        try:31            catkin.builder.subprocess.Popen = MockPopen32            catkin.builder.run_command(['false'], os.getcwd(), True, True)33        finally:34            catkin.builder.subprocess.Popen = backup_Popen35    def test_run_command_unicode_bytes(self):36        backup_Popen = catkin.builder.subprocess.Popen37        class StdOut(object):38            def __init__(self, popen):39                self.__popen = popen40            def readline(self):41                self.__popen.returncode = 042                try:43                    # for Python 2 compatibility only44                    s = unichr(2018)45                except NameError:46                    s = chr(2018)47                return s.encode('utf8')48        class MockPopen(object):49            def __init__(self, *args, **kwargs):50                self.returncode = None51                self.stdout = StdOut(self)52            def wait(self):53                return True54        try:55            catkin.builder.subprocess.Popen = MockPopen56            catkin.builder.run_command(['false'], os.getcwd(), True, True)57        finally:58            catkin.builder.subprocess.Popen = backup_Popen59    def test_extract_jobs_flags(self):60        valid_mflags = [61            '-j8 -l8', 'j8 ', '-j', 'j', '-l8', 'l8',62            '-l', 'l', '-j18', ' -j8 l9', '-j1 -l1',63            '--jobs=8', '--jobs 8', '--jobs', '--load-average',64            '--load-average=8', '--load-average 8', '--jobs=8 -l9'65        ]66        results = [67            '-j8 -l8', 'j8', '-j', 'j', '-l8', 'l8',68            '-l', 'l', '-j18', '-j8 l9', '-j1 -l1',69            '--jobs=8', '--jobs 8', '--jobs', '--load-average',70            '--load-average=8', '--load-average 8', '--jobs=8 -l9'71        ]72        for mflag, result in zip(valid_mflags, results):73            match = catkin.builder.extract_jobs_flags(mflag)74            assert match == result, "should match '{0}'".format(mflag)75            print('--')76            print("input:    '{0}'".format(mflag))77            print("matched:  '{0}'".format(match))78            print("expected: '{0}'".format(result))79        invalid_mflags = ['', '--jobs= 8', '--jobs8']80        for mflag in invalid_mflags:81            match = catkin.builder.extract_jobs_flags(mflag)...

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.