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

How to use update_environment method in green

Best Python code snippet using green

__main__.py

Source:__main__.py

1#!/usr/bin/env python32#3#	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,4#	INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR5#	PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE6#	FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR7#	OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER8#	DEALINGS IN THE SOFTWARE.9#10# Copyright (c) 2020, 2021 Skylark Wireless.11import sys12import argparse13import asyncio14import logging15from typing import Iterable16from pyfaros.updater.updater import do_update, do_update_and_wait17from pyfaros.updater.update_environment import UpdateEnvironment18from pyfaros.discover.discover import Discover, CPERemote, IrisRemote, HubRemote, VgerRemote, Remote19import pkg_resources20async def update_devices(environment: UpdateEnvironment, devices: Iterable[Remote], args) -> None:21    timeout = args.timeout22    store_ssh = not args.no_ssh23    if timeout == 0:24        await do_update(environment, devices, store_ssh=store_ssh)25    else:26        if not await do_update_and_wait(environment, devices, 15, timeout, store_ssh=store_ssh):27            logging.error('Failed to reach devices within {} seconds after reboot'.format(timeout))28if __name__ == '__main__':29    parser = argparse.ArgumentParser(30            prog="python3 -m pyfaros.updater",31            description="Update Skylark Wireless Devices",32            add_help=False)33    parser.add_argument(34        'serial', help="Serials to patch", nargs="*")35    general_options = parser.add_argument_group("General Options")36    device_type_options = parser.add_argument_group("Device Type Override Options")37    advanced_options = parser.add_argument_group("Advanced Options")38    general_options.add_argument(39        '-U', '--user', help="Username", action="store", required=True)40    general_options.add_argument(41        '-P', '--password', help="Password", action="store", required=True)42    general_options.add_argument(43        '-d', '--debug', help="turn on debug messages", action="store_true")44    general_options.add_argument(45        '-u', '--universal',46        help="Path to universal tarball",47        action="store",48        default=None)49    general_options.add_argument(50        '-R', '--recursive', help="Upgrade all connected devices", action="store_true", required=False)51    general_options.add_argument(52        '-n', '--dry-run',53        help="Don't actuall do the update.",54        action="store_true",55        default=False)56    general_options.add_argument(57        "-v", "--version",58        action="version",59        #version="pyfaros-{}".format(dir(pyfaros)),60        version="pyfaros-{}".format(pkg_resources.get_distribution("pyfaros").version),61        help="Displays the version and then exits.",62    )63    general_options.add_argument(64        "-h", "--help",65        action="help",66        default=argparse.SUPPRESS,67        help="Displays this help message and then exits.",68    )69    general_options.add_argument(70        '-t', '--timeout',71        type=int, default=300,72        help='The maximum duration (in seconds) to wait for the devices to come back online.',73    )74    advanced_options.add_argument(75        '--bootbit-only',76        help="Only update bootbit, no other files.",77        action="store_true",78        default=False)79    advanced_options.add_argument(80        '--imageub-only',81        help="Only update imageub, no other files.",82        action="store_true",83        default=False)84    advanced_options.add_argument(85        '--bootbin-only',86        help="Only update bootbin, no other files.",87        action="store_true",88        default=False)89    advanced_options.add_argument(90        '--file',91        help="Individual tarballs to apply.",92        action="append",93        default=[])94    advanced_options.add_argument(95        '--standalone',96        help="Update all standalone iris nodes.",97        action="store_true",98        default=False)99    advanced_options.add_argument(100        '--patch-all',101        help="Patch everything on the network.",102        action="store_true",103        default=False)104    advanced_options.add_argument(105        '--no-ssh',106        help="Do not store ssh keys in /boot/ when updating cards.",107        action='store_true',108        default=False)109    advanced_options.add_argument(110        '--enable-sudo',111        help="Force a password-less sudo.  Only used on firmware older than 2019-07.0.0",112        action='store_true',113        default=False)114    extra_helps = {115        "hub:som6": "  WARNING: Choosing the wrong type will cause the HUB to not boot and the SD will need to be externally re-imaged.",116        "hub:som9": "  WARNING: Choosing the wrong type will cause the HUB to not boot and the SD will need to be externally re-imaged.",117        "iris030_ue:iris030_sdr": "  WARNING: If the Iris is plugged in to 1 GbE fiber this WILL BREAK CONNECTIVITY to the Iris.",118    }119    for device in [IrisRemote, CPERemote, HubRemote, VgerRemote]:120        for v1 in device.Variant:121            support_to = getattr(v1, 'support_to', list(set(device.Variant) - {v1}))122            for v2 in support_to:123                if v1 in getattr (v2, 'support_from', device.Variant):124                    extra_help = extra_helps.get("{}:{}".format(v1.value, v2.value), "")125                    devname = device.__name__.strip("Remote")126                    devname_pl = devname + ("es" if devname.endswith('s') else "s")127                    help_str = "For {} currently on a {} image, apply a {} image.{}".format(128                        devname_pl, v1.value, v2.value, extra_help)129                    # REVISIT: This is a hacky way of having specialized help for the HUB.130                    if not getattr(v1, 'support_from', True):131                        help_str = "Apply the {} image to the {}.{}".format(v2.value, v1.value, extra_help)132                    device_type_options.add_argument(133                        '--treat-{}-as-{}'.format(v1.value, v2.value),134                        help=help_str,135                        action="store_true",136                        default=False)137    args = parser.parse_args()138    if args.debug:139        logging.basicConfig(level=logging.DEBUG)140    else:141        logging.basicConfig(level=logging.INFO)142        logging.getLogger("asyncssh").setLevel(level=logging.WARN)143    logging.debug("Entered main")144    if args.universal:145        mode = UpdateEnvironment.Mode.UNIVERSAL_TARBALL146    elif args.file:147        mode = UpdateEnvironment.Mode.INDIVIDUAL_TARBALLS148    else:149        parser.print_help()150        sys.exit(0)151    try:152        with UpdateEnvironment(153            universal_tarball_path=args.universal,154            individual_tarball_paths=args.file,155            bootbin_path=None,156            imageub_path=None,157            bootbin_only=False,158            imageub_only=False,159            mode=mode,160            family=None,161            variant=None,162            iwantabrick=False,163            interactive=False) as update_environment:164            logging.debug("Entered ue")165            logging.debug(args)166            logging.debug("Looking for remaps, and remapping")167            for device in [IrisRemote, CPERemote, HubRemote, VgerRemote]:168                for v1 in device.Variant:169                    for v2 in device.Variant:170                        # Just use the existing of the attrbiute to determine if the operation is supported171                        remap_name = 'treat_{}_as_{}'.format(v1.value, v2.value)172                        remap_wanted = getattr(args, remap_name, False)173                        if remap_wanted:174                            logging.debug("Did remap for {} to {}".format(v1.value, v2.value))175                            update_environment.mapping[v1] = update_environment.mapping[v2]176            top = Discover()177            discovered = sorted(178                filter(update_environment.availablefilter(),179                       list(top)),180                key=Discover.Sortings.POWER_DEPENDENCY)181            logging.debug("Discovered objects: {}".format(discovered))182            if args.recursive:183                for device in top:184                    if device.serial in args.serial:185                        for child_device in device.walk():186                            if child_device.serial not in args.serial:187                                args.serial.append(child_device.serial)188            if not args.patch_all:189                discovered = list(filter(lambda x: x.serial in args.serial, discovered))190            elif args.standalone:191                discovered = list(192                    filter(193                        lambda x: x.rrh is None,194                        filter(lambda x: isinstance(x, IrisRemote),195                               discovered)))196            logging.debug("Filtered discovered objects: {}".format(discovered))197            logging.info("About to flash devices:")198            for device in discovered:199                device.set_credentials(args.user, args.password)200                if args.enable_sudo:201                    device.enable_sudo()202                device.set_variant()203                logging.info("\t {} - {}\n\t\t{}\n\t\t{}\n\t\t{}".format(204                    device.serial, device.address,205                    update_environment.mapping[device.variant].bootbin,206                    update_environment.mapping[device.variant].bootbit,207                    update_environment.mapping[device.variant].imageub))208            if not args.dry_run:209                loop = asyncio.get_event_loop()210                loop.run_until_complete(update_devices(update_environment, discovered, args))211                loop.close()212    except Exception as e:213        logging.debug(e)...

fabfile.py

Source:fabfile.py

...80def deploy_to_dev_environment():81    version = "build_" + str(int(time.time()))82    upload_to_s3(BUCKET_NAME, version, FILE_NAME)83    create_version(APPLICATION_NAME, version)84    update_environment(ENVIRONMENT_NAME, version)85@task()86def create_new_version(version):87    upload_to_s3(BUCKET_NAME, version, FILE_NAME)88    create_version(APPLICATION_NAME, version)89    90@task()91def create_new_prod_version(version):92    upload_to_s3(BUCKET_NAME, version, FILE_NAME)93    create_version(APPLICATION_NAME, version)94    95    ENVIRONMENT_NAME = 'gladminds-work-prod'96    update_environment(ENVIRONMENT_NAME, version)97    ENVIRONMENT_NAME = 'gladminds-web-prod-2-1'98    update_environment(ENVIRONMENT_NAME, version)99    ENVIRONMENT_NAME = 'gladminds-work-prod-sms'100    update_environment(ENVIRONMENT_NAME, version)101@task()102def create_new_qa_version(version):103    upload_to_s3(BUCKET_NAME, version, FILE_NAME)104    create_version(APPLICATION_NAME, version)105    ENVIRONMENT_NAME = 'gladminds-worker-qa2-1'106    update_environment(ENVIRONMENT_NAME, version)107    108    ENVIRONMENT_NAME = 'gladminds-webserver-qa2'109    update_environment(ENVIRONMENT_NAME, version)110    111@task()112def create_new_demosfa_version(version):113    upload_to_s3(BUCKET_NAME, version, FILE_NAME)114    create_version('Demos', version)115    ENVIRONMENT_NAME = 'demos-env01'116    update_environment(ENVIRONMENT_NAME, version)117@task()118def create_new_staging_version(version):119    upload_to_s3(BUCKET_NAME, version, FILE_NAME)120    create_version(APPLICATION_NAME, version)121    ENVIRONMENT_NAME = 'gladminds-web-dev'122    update_environment(ENVIRONMENT_NAME, version)123    ENVIRONMENT_NAME = 'gladminds-work-dev'124    update_environment(ENVIRONMENT_NAME, version)  125    ENVIRONMENT_NAME = 'gladminds-work-dev-sms'126    update_environment(ENVIRONMENT_NAME, version)127def upload_to_s3(bucket_name, key, file_name):128    conn = boto.connect_s3(ACCESS_KEY, SECRET_KEY)129    bucket = conn.get_bucket(bucket_name)130    k = Key(bucket)131    k.key = key132    k.set_contents_from_filename(file_name)133def create_version(application, version):134    beanstalk = boto.connect_beanstalk(ACCESS_KEY, SECRET_KEY)135    beanstalk.create_application_version(application, version, s3_bucket=BUCKET_NAME, s3_key=version)136def update_environment(environment, version):137    beanstalk = boto.connect_beanstalk(ACCESS_KEY, SECRET_KEY)138    beanstalk.update_environment(environment_name=environment, version_label=version)139@task()140def check():141    '''Runs all checks and reports as JSON and out/summary.html. Useful for CI.'''142    import xml.etree.ElementTree as ET143    state.output.everything = False144    global CAPTURE145    global NEVER_FAIL146    global COVERAGE_ENABLED147    COVERAGE_ENABLED = True148    CAPTURE = True149    NEVER_FAIL = True150    test_integration()151    js_lint = lint_js().split('\n')152    py_lint = lint_py().split('\n')...

test_cli.py

Source:test_cli.py

1import boto2import mock3import moto4import tempfile5import unittest6from click.testing import CliRunner7from rubberjackcli.click import rubberjack8class CLITests(unittest.TestCase):9    @moto.mock_s3_deprecated10    @mock.patch('boto.beanstalk.layer1.Layer1.create_application_version')11    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')12    def test_deploy(self, cav, ue):13        s3 = boto.connect_s3()14        s3.create_bucket("laterpay-rubberjack-ebdeploy")  # FIXME Remove hardcoded bucket name15        with tempfile.NamedTemporaryFile() as tmp:16            result = CliRunner().invoke(rubberjack, ['deploy', tmp.name], catch_exceptions=False)17            self.assertEquals(result.exit_code, 0, result.output)18    @moto.mock_s3_deprecated19    @mock.patch('boto.beanstalk.layer1.Layer1.describe_environments')20    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')21    def test_promote(self, ue, de):22        de.return_value = {23            'DescribeEnvironmentsResponse': {24                'DescribeEnvironmentsResult': {25                    'Environments': [26                        {27                            'EnvironmentName': 'laterpay-devnull-live',  # FIXME Remove hardcoded EnvName28                            'VersionLabel': 'old',29                        },30                        {31                            'EnvironmentName': 'laterpay-devnull-dev',  # FIXME Remove hardcoded EnvName32                            'VersionLabel': 'new',33                        },34                    ],35                },36            },37        }38        CliRunner().invoke(rubberjack, ['promote'], catch_exceptions=False)39    @moto.mock_s3_deprecated40    @mock.patch('sys.exit')41    @mock.patch('boto.beanstalk.layer1.Layer1.describe_environments')42    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')43    def test_promoting_same_version(self, ue, de, se):44        de.return_value = {45            'DescribeEnvironmentsResponse': {46                'DescribeEnvironmentsResult': {47                    'Environments': [48                        {49                            'EnvironmentName': 'laterpay-devnull-live',  # FIXME Remove hardcoded EnvName50                            'VersionLabel': 'same',51                        },52                        {53                            'EnvironmentName': 'laterpay-devnull-dev',  # FIXME Remove hardcoded EnvName54                            'VersionLabel': 'same',55                        },56                    ],57                },58            },59        }60        CliRunner().invoke(rubberjack, ['promote'], catch_exceptions=False)61        self.assertTrue(se.called)62    @moto.mock_s3_deprecated63    def test_sigv4(self):64        CliRunner().invoke(rubberjack, ['--sigv4-host', 'foo', 'deploy'], catch_exceptions=False)65    @moto.mock_s3_deprecated66    @mock.patch('boto.beanstalk.layer1.Layer1.create_application_version')67    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')68    def test_deploy_to_custom_environment(self, ue, cav):69        s3 = boto.connect_s3()70        s3.create_bucket("laterpay-rubberjack-ebdeploy")  # FIXME Remove hardcoded bucket name71        with tempfile.NamedTemporaryFile() as tmp:72            result = CliRunner().invoke(rubberjack, ['deploy', '--environment', 'wibble', tmp.name], catch_exceptions=False)73            self.assertEquals(result.exit_code, 0, result.output)74        self.assertEqual(cav.call_count, 1, "create_application_version wasn't called, but it should")75        self.assertEqual(ue.call_count, 1, "update_environment wasn't called, but it should")76    @moto.mock_s3_deprecated77    @mock.patch('boto.beanstalk.layer1.Layer1.create_application_version')78    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')79    def test_deploy_without_updating_the_environment(self, ue, cav):80        s3 = boto.connect_s3()81        s3.create_bucket("laterpay-rubberjack-ebdeploy")  # FIXME Remove hardcoded bucket name82        with tempfile.NamedTemporaryFile() as tmp:83            result = CliRunner().invoke(rubberjack, ['deploy', '--no-update-environment', tmp.name], catch_exceptions=False)84            self.assertEquals(result.exit_code, 0, result.output)85        self.assertEqual(cav.call_count, 1, "create_application_version wasn't called, but it should")86        self.assertEqual(ue.call_count, 0, "update_environment was called, but it shouldn't")87    @moto.mock_s3_deprecated88    @mock.patch('boto.beanstalk.layer1.Layer1.create_application_version')89    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')90    def test_deploy_to_custom_bucket(self, ue, cav):91        bucket_name = 'rbbrjck-test'92        s3 = boto.connect_s3()93        s3.create_bucket(bucket_name)94        with tempfile.NamedTemporaryFile() as tmp:95            result = CliRunner().invoke(rubberjack, ['--bucket', bucket_name, 'deploy', tmp.name], catch_exceptions=False)96            self.assertEquals(result.exit_code, 0, result.output)97        self.assertEqual(cav.call_count, 1, "create_application_version wasn't called, but it should")98        self.assertEqual(ue.call_count, 1, "update_environment wasn't called, but it should")99        _, cav_kwargs = cav.call_args100        self.assertEqual(bucket_name, cav_kwargs['s3_bucket'])101    @moto.mock_s3_deprecated102    @mock.patch('boto.beanstalk.layer1.Layer1.update_environment')103    @mock.patch('boto.beanstalk.layer1.Layer1.describe_environments')104    def test_promote_to_custom_environment(self, de, ue):105        CUSTOM_TO_ENVIRONMENT = "loremipsum"106        de.return_value = {107            'DescribeEnvironmentsResponse': {108                'DescribeEnvironmentsResult': {109                    'Environments': [110                        {111                            'EnvironmentName': CUSTOM_TO_ENVIRONMENT,112                            'VersionLabel': 'old',113                        },114                        {115                            'EnvironmentName': 'laterpay-devnull-dev',  # FIXME Remove hardcoded EnvName116                            'VersionLabel': 'new',117                        },118                    ],119                },120            },121        }122        result = CliRunner().invoke(rubberjack, ['promote', '--to-environment', CUSTOM_TO_ENVIRONMENT], catch_exceptions=False)...

env.py

Source:env.py

...6@contextmanager7def set_env(*args, **kwargs):8    """Context Mgr to set an environment variable9    """10    def update_environment(env):11        for k, v in env.items():12            if v is None:13                if k in os.environ:14                    del os.environ[k]15            else:16                os.environ[k] = str(v)17    # Backward compatibility with the old interface which only allowed to18    # update a single environment variable.19    new_values = dict([(args[0], args[1])]) if len(args) == 2 else {}20    new_values.update((k, v) for k, v in kwargs.items())21    # Save variables that are going to be updated.22    saved_values = dict((k, os.environ.get(k)) for k in new_values.keys())23    # Update variables to their temporary values24    try:25        update_environment(new_values)26        yield27    finally:28        # Restore original environment29        update_environment(saved_values)30@contextmanager31def unset_env(env_var_skiplist):32    """Context Mgr to unset an environment variable temporarily."""33    def update_environment(env):34        os.environ.clear()35        os.environ.update(env)36    # Save variables that are going to be updated.37    saved_values = dict(os.environ)38    new_values = dict((k, v) for k, v in os.environ.items() if k not in env_var_skiplist)39    # Update variables to their temporary values40    update_environment(new_values)41    (yield)42    # Restore original environment43    update_environment(saved_values)44@contextmanager45def no_env(key):46    """47    Context Mgr to asserting no environment variable of the given name exists48    (sto enable the testing of the case where no env var of this name exists)49    """50    try:51        orig_value = os.environ[key]52        del os.environ[key]53        env_has_key = True54    except KeyError:55        env_has_key = False56    yield57    if env_has_key:...

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.