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

How to use join_command method in autotest

Best Python code snippet using autotest_python

test_pending_start_handler.py

Source:test_pending_start_handler.py

1import unittest2from game_state import GameState3from pending_start_handler import PendingStartHandler, START_COMMAND, JOIN_COMMAND4from test.mock_slack_client import MockSlackClient5from game_end_exception import GameEndException6from generate_game_handler import GenerateGameHandler78class TestPendingStartHandler(unittest.TestCase):9    def setUp(self):10        self.slack_client = MockSlackClient()11        self.game_state = GameState(self.slack_client)12        self.handler = PendingStartHandler(self.game_state)13        14    def _start(self):15        data = {'text': START_COMMAND, 'channel': 'foo'}16        self.handler.process(data)17        18    def test_start(self):19        self._start()20        self.assertEquals('foo', self.game_state.channel)21        22    def test_players_join_success(self):23        self._start()24        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p1'}25        self.handler.process(data)26        27        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p2'}28        self.handler.process(data)29        30        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p3'}31        self.handler.process(data)32        33        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p4'}34        self.handler.process(data)35        36        self.assertEqual(4, len(self.game_state.team_manager.players))37        last_api_call = self.slack_client.api_calls[-1]38        self.assertEquals("You have joined the game!", last_api_call[1])39        40    def _count_down(self):41        self.handler.tick()42        last_api_call = self.slack_client.api_calls[-1]43        self.assertEquals("Starting Codenames game in 30 seconds. Reply `cn join` to play", last_api_call[1])44        self.handler.tick()45        self.handler.tick()46        self.handler.tick()47        self.handler.tick()48        self.handler.tick()49        self.handler.tick()50        self.handler.tick()51        self.handler.tick()52        self.handler.tick()53        self.handler.tick()54        self.handler.tick()55        self.handler.tick()56        self.handler.tick()57        self.handler.tick()58        self.handler.tick()59        self.handler.tick()60        self.handler.tick()61        self.handler.tick()62        self.handler.tick()63        self.handler.tick()64        self.handler.tick()65        self.handler.tick()66        self.handler.tick()67        self.handler.tick()68        self.handler.tick()69        self.handler.tick()70        self.handler.tick()71        self.handler.tick()72        self.handler.tick()7374    def test_players_join_fail(self):75        self._start()76        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p1'}77        self.handler.process(data)78        79        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p2'}80        self.handler.process(data)81        82        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p3'}83        self.handler.process(data)84        85        self._count_down()86        87        last_api_call = self.slack_client.api_calls[-1]88        self.assertEquals("Starting Codenames game in 1 seconds. Reply `cn join` to play", last_api_call[1])89        90        caught_exception = False91        try:92            self.handler.tick()93        except GameEndException:94            caught_exception = True95            96        self.assertTrue(caught_exception)97        98    def test_players_join_success_start(self):99        self._start()100        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p1'}101        self.handler.process(data)102        103        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p2'}104        self.handler.process(data)105        106        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p3'}107        self.handler.process(data)108        109        data = {'text': JOIN_COMMAND, 'channel': 'foo', 'user': 'p4'}110        self.handler.process(data)111        112        self._count_down()113        self.handler.tick()114        115        self.assertTrue(isinstance(self.game_state.handler, GenerateGameHandler))116        
...

configure_node.py

Source:configure_node.py

1#!/usr/bin/env python2#3# Copyright (c) 2017-2019 Cloudify Platform Ltd. All rights reserved4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9#        http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.16import subprocess17from cloudify import ctx18from cloudify.state import ctx_parameters as inputs19def execute_command(_command):20    ctx.logger.debug('_command {0}.'.format(_command))21    subprocess_args = {22        'args': _command.split(),23        'stdout': subprocess.PIPE,24        'stderr': subprocess.PIPE25    }26    ctx.logger.debug('subprocess_args {0}.'.format(subprocess_args))27    process = subprocess.Popen(**subprocess_args)28    output, error = process.communicate()29    ctx.logger.debug('command: {0} '.format(_command))30    ctx.logger.debug('output: {0} '.format(output))31    ctx.logger.debug('error: {0} '.format(error))32    ctx.logger.debug('process.returncode: {0} '.format(process.returncode))33    if process.returncode:34        ctx.logger.error('Running `{0}` returns error.'.format(_command))35        return False36    return output37if __name__ == '__main__':38    join_command = inputs['join_command']39    join_command = 'sudo {0} --skip-preflight-checks'.format(join_command)40    execute_command(join_command)41    # Install weave-related utils42    execute_command('sudo curl -L git.io/weave -o /usr/local/bin/weave')43    execute_command('sudo chmod a+x /usr/local/bin/weave')44    execute_command('sudo curl -L git.io/scope -o /usr/local/bin/scope')45    execute_command('sudo chmod a+x /usr/local/bin/scope')46    execute_command('/usr/local/bin/scope launch')47    hostname = execute_command('hostname')...

fabfile.py

Source:fabfile.py

1import fabricio2import os3from fabric import api as fab4from fabricio import tasks, docker5from fabricio.misc import AvailableVagrantHosts6hosts = AvailableVagrantHosts(guest_network_interface='eth1')7swarm_host='192.168.0.199'8stack = tasks.DockerTasks(9    service=docker.Stack(10        name='example_docker',11        options={12            'compose-file': 'docker/docker-compose.yml',13            'env': f'PWD={os.getcwd()}',14        },15    ),16    hosts=hosts,17    # rollback_command=True,  # show `rollback` command in the list18    # migrate_commands=True,  # show `migrate` and `migrate-back` commands in the list19    # backup_commands=True,  # show `backup` and `restore` commands in the list20    # pull_command=True,  # show `pull` command in the list21    # update_command=True,  # show `update` command in the list22    # revert_command=True,  # show `revert` command in the list23    # destroy_command=True,  # show `destroy` command in the list24)25@fab.task(name='swarm-init')26@fab.serial27def swarm_init():28    """29    enable Docker swarm mode30    """31    join_token = fabricio.run(32        'docker swarm join-token --quiet manager',33        ignore_errors=True34    )35    def init():36        if not init.join_command:37            init.join_command = (38                'docker swarm join --token {join_token} {host}:2377'39            ).format(join_token=join_token, host=swarm_init_ip)40        else:41            fabricio.run(init.join_command, ignore_errors=True, quiet=False)42    init.join_command = None43    with fab.settings(hosts=hosts):44        fab.execute(init)45@fab.task(name='swarm-reset')46def swarm_reset():47    """48    enable Docker swarm mode49    """50    def reset():51        fabricio.run('docker swarm leave --force', ignore_errors=True, quiet=False)52    with fab.settings(hosts=hosts):...

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.