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

How to use test_stdout method in green

Best Python code snippet using green

test_handle_process.py

Source:test_handle_process.py

1# Copyright(c) 2016 Nippon Telegraph and Telephone Corporation2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import mock15import socket16import testtools17import eventlet18from oslo_utils import timeutils19import masakarimonitors.conf20from masakarimonitors.ha import masakari21from masakarimonitors.objects import event_constants as ec22from masakarimonitors.processmonitor.process_handler import handle_process23from masakarimonitors import utils24CONF = masakarimonitors.conf.CONF25eventlet.monkey_patch(os=False)26MOCK_PROCESS_LIST = [27    {28        'id': 1,29        'process_name': 'mock_process_name_A',30        'start_command': 'mock_start_command',31        'pre_start_command': 'mock_pre_start_command',32        'post_start_command': 'mock_post_start_command',33        'restart_command': 'mock_restart_command',34        'pre_restart_command': 'mock_pre_restart_command',35        'post_restart_command': 'mock_post_restart_command',36        'run_as_root': True37    },38]39MOCK_DOWN_PROCESS_LIST = [40    {41        'id': 1,42        'process_name': 'mock_process_name_A',43        'start_command': 'mock_start_command',44        'pre_start_command': 'mock_pre_start_command',45        'post_start_command': 'mock_post_start_command',46        'restart_command': 'mock_restart_command',47        'pre_restart_command': 'mock_pre_restart_command',48        'post_restart_command': 'mock_post_restart_command',49        'run_as_root': True50    },51]52PS_RESULT = "\n" \53            "UID  PID   PPID C STIME TTY TIME     CMD\n" \54            "root 11187 1    0 18:52 ?   00:00:00 mock_process_name_A\n"55class TestHandleProcess(testtools.TestCase):56    def setUp(self):57        super(TestHandleProcess, self).setUp()58    def test_set_process_list(self):59        process_list = MOCK_PROCESS_LIST60        obj = handle_process.HandleProcess()61        obj.set_process_list(process_list)62        self.assertEqual(process_list, obj.process_list)63    @mock.patch.object(utils, 'execute')64    def test_start_processes(self,65                             mock_execute):66        process_list = MOCK_PROCESS_LIST67        obj = handle_process.HandleProcess()68        obj.set_process_list(process_list)69        mock_execute.side_effect = [('test_stdout', ''),70                                    ('test_stdout', ''),71                                    ('test_stdout', 'test_stderr')]72        obj.start_processes()73        mock_execute.assert_any_call(74            MOCK_PROCESS_LIST[0].get('pre_start_command'),75            run_as_root=MOCK_PROCESS_LIST[0].get('run_as_root'))76        mock_execute.assert_any_call(77            MOCK_PROCESS_LIST[0].get('start_command'),78            run_as_root=MOCK_PROCESS_LIST[0].get('run_as_root'))79        mock_execute.assert_any_call(80            MOCK_PROCESS_LIST[0].get('post_start_command'),81            run_as_root=MOCK_PROCESS_LIST[0].get('run_as_root'))82    @mock.patch.object(utils, 'execute')83    def test_start_processes_pre_cmd_fail(self, mock_execute):84        process_list = MOCK_PROCESS_LIST85        obj = handle_process.HandleProcess()86        obj.set_process_list(process_list)87        mock_execute.return_value = ('test_stdout', 'test_stderr')88        obj.start_processes()89        mock_execute.assert_called_once_with(90            MOCK_PROCESS_LIST[0].get('pre_start_command'),91            run_as_root=MOCK_PROCESS_LIST[0].get('run_as_root'))92    @mock.patch.object(utils, 'execute')93    def test_monitor_processes(self,94                               mock_execute):95        process_list = MOCK_PROCESS_LIST96        obj = handle_process.HandleProcess()97        obj.set_process_list(process_list)98        mock_execute.return_value = (PS_RESULT, '')99        down_process_list = obj.monitor_processes()100        self.assertEqual([], down_process_list)101        mock_execute.assert_called_once_with(102            'ps', '-ef', run_as_root=False)103    @mock.patch.object(utils, 'execute')104    def test_monitor_processes_not_found(self, mock_execute):105        process_list = MOCK_PROCESS_LIST106        obj = handle_process.HandleProcess()107        obj.set_process_list(process_list)108        mock_execute.return_value = ('', '')109        down_process_list = obj.monitor_processes()110        self.assertEqual(MOCK_PROCESS_LIST, down_process_list)111        mock_execute.assert_called_once_with(112            'ps', '-ef', run_as_root=False)113    @mock.patch.object(utils, 'execute')114    def test_monitor_processes_exception(self, mock_execute):115        process_list = MOCK_PROCESS_LIST116        obj = handle_process.HandleProcess()117        obj.set_process_list(process_list)118        mock_execute.side_effect = Exception("Test exception.")119        down_process_list = obj.monitor_processes()120        self.assertEqual([], down_process_list)121    @mock.patch.object(utils, 'execute')122    def test_restart_processes(self,123                               mock_execute):124        process_list = MOCK_PROCESS_LIST125        obj = handle_process.HandleProcess()126        obj.set_process_list(process_list)127        down_process_list = MOCK_DOWN_PROCESS_LIST128        mock_execute.side_effect = [('test_stdout', ''),129                                    ('test_stdout', ''),130                                    ('test_stdout', '')]131        obj.restart_processes(down_process_list)132        mock_execute.assert_any_call(133            down_process_list[0].get('pre_restart_command'),134            run_as_root=down_process_list[0].get('run_as_root'))135        mock_execute.assert_any_call(136            down_process_list[0].get('restart_command'),137            run_as_root=down_process_list[0].get('run_as_root'))138        mock_execute.assert_any_call(139            down_process_list[0].get('post_restart_command'),140            run_as_root=down_process_list[0].get('run_as_root'))141        self.assertEqual([], obj.restart_failure_list)142    @mock.patch.object(utils, 'execute')143    def test_restart_processes_failed_to_restart_previously(144        self, mock_execute):145        process_list = MOCK_PROCESS_LIST146        obj = handle_process.HandleProcess()147        obj.set_process_list(process_list)148        restart_failure_list = [MOCK_DOWN_PROCESS_LIST[0].get('process_name')]149        obj.restart_failure_list = restart_failure_list150        down_process_list = MOCK_DOWN_PROCESS_LIST151        obj.restart_processes(down_process_list)152        self.assertEqual(restart_failure_list, obj.restart_failure_list)153        mock_execute.assert_not_called()154    @mock.patch.object(masakari.SendNotification, 'send_notification')155    @mock.patch.object(timeutils, 'utcnow')156    @mock.patch.object(eventlet.greenthread, 'sleep')157    @mock.patch.object(utils, 'execute')158    def test_restart_processes_pre_restart_command_retry_over(159        self, mock_execute, mock_sleep, mock_utcnow, mock_send_notification):160        process_list = MOCK_PROCESS_LIST161        obj = handle_process.HandleProcess()162        obj.set_process_list(process_list)163        down_process_list = MOCK_DOWN_PROCESS_LIST164        mock_execute.side_effect = [('test_stdout', 'test_stderr'),165                                    ('test_stdout', 'test_stderr'),166                                    ('test_stdout', 'test_stderr'),167                                    ('test_stdout', 'test_stderr')]168        mock_sleep.return_value = None169        current_time = timeutils.utcnow()170        mock_utcnow.return_value = current_time171        mock_send_notification.return_value = None172        obj.restart_processes(down_process_list)173        pre_execute_count = CONF.process.restart_retries + 1174        self.assertEqual(pre_execute_count, mock_execute.call_count)175        for var in range(0, mock_execute.call_count):176            args, kwargs = mock_execute.call_args_list[var]177            self.assertEqual(178                (down_process_list[0].get('pre_restart_command'),),179                args)180            self.assertEqual({'run_as_root': True}, kwargs)181        event = {182            'notification': {183                'type': ec.EventConstants.TYPE_PROCESS,184                'hostname': socket.gethostname(),185                'generated_time': current_time,186                'payload': {187                    'event': ec.EventConstants.EVENT_STOPPED,188                    'process_name': down_process_list[0].get('process_name')189                }190            }191        }192        mock_send_notification.assert_called_once_with(193            CONF.process.api_retry_max,194            CONF.process.api_retry_interval,195            event)196        self.assertEqual(197            [down_process_list[0].get('process_name')],198            obj.restart_failure_list)199    @mock.patch.object(masakari.SendNotification, 'send_notification')200    @mock.patch.object(timeutils, 'utcnow')201    @mock.patch.object(eventlet.greenthread, 'sleep')202    @mock.patch.object(utils, 'execute')203    def test_restart_processes_restart_command_retry_over(204        self, mock_execute, mock_sleep, mock_utcnow, mock_send_notification):205        process_list = MOCK_PROCESS_LIST206        obj = handle_process.HandleProcess()207        obj.set_process_list(process_list)208        down_process_list = MOCK_DOWN_PROCESS_LIST209        mock_execute.side_effect = [('test_stdout', ''),210                                    ('test_stdout', 'test_stderr'),211                                    ('test_stdout', ''),212                                    ('test_stdout', 'test_stderr'),213                                    ('test_stdout', ''),214                                    ('test_stdout', 'test_stderr'),215                                    ('test_stdout', ''),216                                    ('test_stdout', 'test_stderr')]217        mock_sleep.return_value = None218        current_time = timeutils.utcnow()219        mock_utcnow.return_value = current_time220        mock_send_notification.return_value = None221        obj.restart_processes(down_process_list)222        pre_execute_count = CONF.process.restart_retries + 1223        execute_count = CONF.process.restart_retries + 1224        total_execute_count = pre_execute_count + execute_count225        self.assertEqual(total_execute_count, mock_execute.call_count)226        for var in range(0, mock_execute.call_count):227            # Execute order of restart_command is the second.228            execute_order = 2229            if (var + 1) % execute_order == 0:230                args, kwargs = mock_execute.call_args_list[var]231                self.assertEqual(232                    (down_process_list[0].get('restart_command'),),233                    args)234                self.assertEqual({'run_as_root': True}, kwargs)235        event = {236            'notification': {237                'type': ec.EventConstants.TYPE_PROCESS,238                'hostname': socket.gethostname(),239                'generated_time': current_time,240                'payload': {241                    'event': ec.EventConstants.EVENT_STOPPED,242                    'process_name': down_process_list[0].get('process_name')243                }244            }245        }246        mock_send_notification.assert_called_once_with(247            CONF.process.api_retry_max,248            CONF.process.api_retry_interval,249            event)250        self.assertEqual(251            [down_process_list[0].get('process_name')],252            obj.restart_failure_list)253    @mock.patch.object(masakari.SendNotification, 'send_notification')254    @mock.patch.object(timeutils, 'utcnow')255    @mock.patch.object(eventlet.greenthread, 'sleep')256    @mock.patch.object(utils, 'execute')257    def test_restart_processes_post_restart_command_retry_over(258        self, mock_execute, mock_sleep, mock_utcnow, mock_send_notification):259        process_list = MOCK_PROCESS_LIST260        obj = handle_process.HandleProcess()261        obj.set_process_list(process_list)262        down_process_list = MOCK_DOWN_PROCESS_LIST263        mock_execute.side_effect = [('test_stdout', ''),264                                    ('test_stdout', ''),265                                    ('test_stdout', 'test_stderr'),266                                    ('test_stdout', ''),267                                    ('test_stdout', ''),268                                    ('test_stdout', 'test_stderr'),269                                    ('test_stdout', ''),270                                    ('test_stdout', ''),271                                    ('test_stdout', 'test_stderr'),272                                    ('test_stdout', ''),273                                    ('test_stdout', ''),274                                    ('test_stdout', 'test_stderr')]275        mock_sleep.return_value = None276        current_time = timeutils.utcnow()277        mock_utcnow.return_value = current_time278        mock_send_notification.return_value = None279        obj.restart_processes(down_process_list)280        pre_execute_count = CONF.process.restart_retries + 1281        execute_count = CONF.process.restart_retries + 1282        post_execute_count = CONF.process.restart_retries + 1283        total_execute_count = \284            pre_execute_count + execute_count + post_execute_count285        self.assertEqual(total_execute_count, mock_execute.call_count)286        for var in range(0, mock_execute.call_count):287            # Execute order of restart_command is the third.288            execute_order = 3289            if (var + 1) % execute_order == 0:290                args, kwargs = mock_execute.call_args_list[var]291                self.assertEqual(292                    (down_process_list[0].get('post_restart_command'),),293                    args)294                self.assertEqual({'run_as_root': True}, kwargs)295        event = {296            'notification': {297                'type': ec.EventConstants.TYPE_PROCESS,298                'hostname': socket.gethostname(),299                'generated_time': current_time,300                'payload': {301                    'event': ec.EventConstants.EVENT_STOPPED,302                    'process_name': down_process_list[0].get('process_name')303                }304            }305        }306        mock_send_notification.assert_called_once_with(307            CONF.process.api_retry_max,308            CONF.process.api_retry_interval,309            event)310        self.assertEqual(311            [down_process_list[0].get('process_name')],312            obj.restart_failure_list)313    @mock.patch.object(utils, 'execute')314    def test_execute_cmd_exception(self, mock_execute):315        mock_execute.side_effect = Exception("Test exception.")316        obj = handle_process.HandleProcess()317        ret = obj._execute_cmd(MOCK_PROCESS_LIST[0].get('start_command'),318                               MOCK_PROCESS_LIST[0].get('run_as_root'))...

test_fprint.py

Source:test_fprint.py

...90    with pytest.raises(ValueError, match='Duplicate background'):91        colorise.fprint('Hello {bg=red,bg=red}world')92@pytest.mark.skip_on_windows93def test_valid_named_fprint_output(test_stdout):94    test_stdout(95        colorise.fprint,96        '\x1b[0m\x1b[31mHello\x1b[0m' + os.linesep,97        '{fg=red}Hello',98    )99@pytest.mark.require_colors(256)100@pytest.mark.skip_on_windows101def test_valid_256_index_fprint_output(test_stdout):102    test_stdout(103        colorise.fprint,104        '\x1b[0m\x1b[38;5;201mHello\x1b[0m' + os.linesep,105        '{fg=201}Hello',106    )107@pytest.mark.require_colors(256**3)108@pytest.mark.skip_on_windows109def test_valid_truecolor_fprint_output(test_stdout):110    tests = [111        ('fg=0xa696ff',112         '\x1b[0m\x1b[38;2;166;150;255mHello\x1b[0m' + os.linesep),113        ('fg=0xa9f',114         '\x1b[0m\x1b[38;2;170;153;255mHello\x1b[0m' + os.linesep),115        ('fg=hls(249.08;79.4;100)',116         '\x1b[0m\x1b[38;2;165;149;255mHello\x1b[0m' + os.linesep),117        ('fg=hsv(249.2;41.0;100)',118         '\x1b[0m\x1b[38;2;166;150;255mHello\x1b[0m' + os.linesep),119        ('fg=rgb(167;151;255)',120         '\x1b[0m\x1b[38;2;167;151;255mHello\x1b[0m' + os.linesep),121    ]122    for color, expected in tests:123        test_stdout(colorise.fprint, expected, '{' + color + '}Hello')124@pytest.mark.skip_on_windows125def test_fprint_disabled(test_stdout):126    test_stdout(127        colorise.fprint,128        'Hello' + os.linesep,129        '{fg=red}Hello',130        enabled=False,131    )132    test_stdout(133        colorise.fprint,134        'Hello' + os.linesep,135        '{fg=red}Hel{bg=blue}lo{fg=green}',136        enabled=False,137    )138@pytest.mark.skip_on_windows139def test_fprint_proper_reset(redirect):140    with redirect('stdout') as stdout:141        colorise.set_color(fg='red')142        colorise.fprint('Hel{bg=blue}lo', file=sys.stdout)143        assert stdout.value == '\x1b[0mHel\x1b[44mlo\x1b[0m' + os.linesep144@pytest.mark.skip_on_windows145def test_fprint_autoreset(test_stdout):146    text = '{fg=red}Hello {bg=blue}world!'147    test_stdout(148        colorise.fprint,149        '\x1b[0m\x1b[31mHello \x1b[44mworld!\x1b[0m' + os.linesep,150        text,151        autoreset=False,152    )153    test_stdout(154        colorise.fprint,155        '\x1b[0m\x1b[31mHello \x1b[0m\x1b[44mworld!\x1b[0m' + os.linesep,156        text,157        autoreset=True,...

test_classmethod.py

Source:test_classmethod.py

...23Foo.new().bar()24""", """4225""")26    def test_classmethod20(self):27        def test_stdout(stdout):28            self._test_regexp(r"""<Class [0-9a-fA-Z]+>29""", stdout)30        self._test("""31class Foo32  def bar()33    puts(self)34  end35  bar = classmethod(bar)36end37Foo.bar()38""", stdout=test_stdout)39    def test_classmethod30(self):40        def test_stdout(stdout):41            self._test_regexp(r"""<Class [0-9a-fA-Z]+>42""", stdout)43        self._test("""44class Foo45  def bar()46    puts(self)47  end48  bar = classmethod(bar)49end50Foo.new().bar()51""", stdout=test_stdout)52    def test_classmethod40(self):53        def test_stdout(stdout):54            self._test_regexp(r"""<Class [0-9a-fA-Z]+>55""", stdout)56        self._test("""57class Foo58  def bar()59    puts(self)60  end61  bar = classmethod(function: bar)62end63Foo.new().bar()64""", stdout=test_stdout)...

test_examples.py

Source:test_examples.py

1from testbook import testbook2@testbook('examples/roux_lib_df.ipynb', execute=True)3def test_stdout(tb):4    # assert tb.cell_output_text(1) == 'hello world!'5    assert 'False' in tb.cell_output_text('validate_no_dups') # 0-based6    assert 'sepal_length_x' in tb.cell_output_text('merge') # 0-based7    8@testbook('examples/roux_lib_io.ipynb', execute=True)9def test_stdout(tb):10    assert "data/table.tsv" in tb.cell_output_text('to_table')    11    12@testbook('examples/roux_query.ipynb', execute=True)13def test_stdout(tb):14    assert "data/biomart/00_raw.tsv" in tb.cell_output_text('to_table')    15@testbook('examples/roux_stat_cluster.ipynb', execute=True)16def test_stdout(tb):17    assert 'data/biomart/01_dedup.tsv' in tb.cell_output_text('to_table')    18@testbook('examples/roux_viz_annot.ipynb', execute=True)19def test_stdout(tb):20    assert 'data/biomart/01_dedup.tsv' in tb.cell_output_text('to_table')    21@testbook('examples/roux_viz_io.ipynb', execute=True)22def test_stdout(tb):...

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.