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

How to use create_parser method in localstack

Best Python code snippet using localstack_python

db_manage.py

Source:db_manage.py

1#!/usr/bin/env python2# Copyright 2010-2015 OpenStack LLC.3# All Rights Reserved.4#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 or14# implied.15# See the License for the specific language governing permissions and16# limitations under the License.17import argparse18import os19import sys20sys.path.insert(0, os.getcwd())21from barbican.common import config22from barbican.model import clean23from barbican.model.migration import commands24from oslo_log import log25# Import and configure logging.26CONF = config.CONF27log.setup(CONF, 'barbican')28LOG = log.getLogger(__name__)29class DatabaseManager(object):30    """Database Manager class.31    Builds and executes a CLI parser to manage the Barbican database32    This extends the Alembic commands.33    """34    def __init__(self, conf):35        self.conf = conf36        self.parser = self.get_main_parser()37        self.subparsers = self.parser.add_subparsers(38            title='subcommands',39            description='Action to perform')40        self.add_revision_args()41        self.add_upgrade_args()42        self.add_history_args()43        self.add_current_args()44        self.add_clean_args()45    def get_main_parser(self):46        """Create top-level parser and arguments."""47        parser = argparse.ArgumentParser(description='Barbican DB manager.')48        parser.add_argument('--dburl', '-d', default=self.conf.sql_connection,49                            help='URL to the database.')50        return parser51    def add_revision_args(self):52        """Create 'revision' command parser and arguments."""53        create_parser = self.subparsers.add_parser('revision', help='Create a '54                                                   'new DB version file.')55        create_parser.add_argument('--message', '-m', default='DB change',56                                   help='the message for the DB change')57        create_parser.add_argument('--autogenerate',58                                   help='autogenerate from models',59                                   action='store_true')60        create_parser.set_defaults(func=self.revision)61    def add_upgrade_args(self):62        """Create 'upgrade' command parser and arguments."""63        create_parser = self.subparsers.add_parser('upgrade',64                                                   help='Upgrade to a '65                                                   'future version DB '66                                                   'version file')67        create_parser.add_argument('--version', '-v', default='head',68                                   help='the version to upgrade to, or else '69                                        'the latest/head if not specified.')70        create_parser.set_defaults(func=self.upgrade)71    def add_history_args(self):72        """Create 'history' command parser and arguments."""73        create_parser = self.subparsers.add_parser(74            'history',75            help='List changeset scripts in chronological order.')76        create_parser.add_argument('--verbose', '-V', action="store_true",77                                   help='Show full information about the '78                                        'revisions.')79        create_parser.set_defaults(func=self.history)80    def add_current_args(self):81        """Create 'current' command parser and arguments."""82        create_parser = self.subparsers.add_parser(83            'current',84            help='Display the current revision for a database.')85        create_parser.add_argument('--verbose', '-V', action="store_true",86                                   help='Show full information about the '87                                        'revision.')88        create_parser.set_defaults(func=self.current)89    def add_clean_args(self):90        """Create 'clean' command parser and arguments."""91        create_parser = self.subparsers.add_parser(92            'clean',93            help='Clean up soft deletions in the database')94        create_parser.add_argument(95            '--min-days', '-m', type=int, default=90,96            help='minimum number of days to keep soft deletions. default is'97                 ' %(default)s days.')98        create_parser.add_argument('--clean-unassociated-projects', '-p',99                                   action="store_true",100                                   help='Remove projects that have no '101                                        'associated resources.')102        create_parser.add_argument('--soft-delete-expired-secrets', '-e',103                                   action="store_true",104                                   help='Soft delete expired secrets.')105        create_parser.add_argument('--verbose', '-V', action='store_true',106                                   help='Show full information about the'107                                        ' cleanup')108        create_parser.add_argument('--log-file', '-L',109                                   default=CONF.log_file,110                                   type=str,111                                   help='Set log file location. '112                                        'Default value for log_file can be '113                                        'found in barbican.conf')114        create_parser.set_defaults(func=self.clean)115    def revision(self, args):116        """Process the 'revision' Alembic command."""117        commands.generate(autogenerate=args.autogenerate,118                          message=args.message,119                          sql_url=args.dburl)120    def upgrade(self, args):121        """Process the 'upgrade' Alembic command."""122        LOG.debug("Performing database schema migration...")123        commands.upgrade(to_version=args.version, sql_url=args.dburl)124    def history(self, args):125        commands.history(args.verbose, sql_url=args.dburl)126    def current(self, args):127        commands.current(args.verbose, sql_url=args.dburl)128    def clean(self, args):129        clean.clean_command(130            sql_url=args.dburl,131            min_num_days=args.min_days,132            do_clean_unassociated_projects=args.clean_unassociated_projects,133            do_soft_delete_expired_secrets=args.soft_delete_expired_secrets,134            verbose=args.verbose,135            log_file=args.log_file)136    def execute(self):137        """Parse the command line arguments."""138        args = self.parser.parse_args()139        # Perform other setup here...140        args.func(args)141def _exception_is_successful_exit(thrown_exception):142    return (isinstance(thrown_exception, SystemExit) and143            (thrown_exception.code is None or thrown_exception.code == 0))144def main():145    try:146        dm = DatabaseManager(CONF)147        dm.execute()148    except Exception as ex:149        if not _exception_is_successful_exit(ex):150            LOG.exception('Problem seen trying to run barbican db manage')151            sys.stderr.write("ERROR: {0}\n".format(ex))152            sys.exit(1)153if __name__ == '__main__':...

commands.py

Source:commands.py

1import argparse2import sys3from os.path import commonprefix4from shutil import copyfileobj5from smart_open import open6from s3mothball.helpers import exists7from s3mothball.s3mothball import write_tar, validate_tar, delete_files, open_archived_file8def do_validate(args):9        print("Validating %s against %s" % (args.tar_path, args.manifest_path))10        validate_tar(args.manifest_path, args.tar_path, progress_bar=args.progress_bar)11def do_delete(args):12    print("Deleting objects listed in %s" % args.manifest_path)13    if not args.force_delete:14        buckets = delete_files(args.manifest_path, dry_run=True)15        for bucket, keys in buckets.items():16            print(" * To delete: %s items from s3://%s/%s" % (len(keys['keys']), bucket, commonprefix(keys['keys'])))17            if keys['mismatched']:18                print("   * WARNING: %s keys skipped because ETag doesn't match TarMD5" % len(keys['mismatched']))19            if input("Delete objects? [y/N] ").lower() != 'y':20                print("Canceled.")21                return22    buckets = delete_files(args.manifest_path, dry_run=False)23    for bucket, keys in buckets.items():24        print(" * Deleted %s items from s3://%s/%s" % (len(keys['deleted']), bucket, commonprefix(keys['keys'])))25        if keys['mismatched']:26            print("   * WARNING: %s keys skipped because ETag doesn't match TarMD5" % len(keys['mismatched']))27        if keys['errors']:28            print("   * WARNING: %s keys returned an error message (permissions issue or key not found)" % len(keys['errors']))29def archive_command(args, parser):30    print("Writing %s and %s" % (args.tar_path, args.manifest_path))31    # handle existing output files32    if not args.overwrite:33        for path in (args.tar_path, args.manifest_path):34            if exists(path):35                if input("%s already exists. Overwrite? [y/N] " % path).lower() != 'y':36                    print("Canceled.")37                    return38                args.overwrite = True39    write_tar(args.archive_url, args.manifest_path, args.tar_path, args.strip_prefix, progress_bar=args.progress_bar, overwrite=args.overwrite)40    if args.validate:41        do_validate(args)42    if args.delete:43        do_delete(args)44def validate_command(args, parser):45        do_validate(args)46def delete_command(args, parser):47    if args.validate:48        if not args.tar_path:49            parser.error("tar_path is required unless --no-validate is set.")50        do_validate(args)51    do_delete(args)52def extract_command(args, parser):53    with open_archived_file(args.manifest_path, args.tar_path, args.file_path) as f:54        if args.out:55            with open(args.out, 'wb') as out:56                copyfileobj(f, out)57        else:58            copyfileobj(f, sys.stdout.buffer)59def main(args=None):60    parser = argparse.ArgumentParser(description='Archive files on S3.')61    parser.add_argument('--no-progress', dest='progress_bar', action='store_false', help="Don't show progress bar when archiving and validating")62    parser.set_defaults(progress_bar=True)63    subparsers = parser.add_subparsers(help='Use s3mothball <command> --help for help')64    # archive65    create_parser = subparsers.add_parser('archive', help='Create a new tar archive and manifest.')66    create_parser.add_argument('archive_url', help='S3 prefix to archive, e.g. s3://bucket/prefix/')67    create_parser.add_argument('manifest_path', help='Path or S3 URL for output manifest file')68    create_parser.add_argument('tar_path', help='Path or S3 URL for output tar file')69    create_parser.add_argument('--strip-prefix', help='optional prefix to strip from inventory file when writing tar', default='')70    create_parser.add_argument('--no-validate', dest='validate', action='store_false', help="Don't validate tar against manifest after creating")71    create_parser.add_argument('--delete', dest='delete', action='store_true', help="Delete files from archive_url after archiving")72    create_parser.add_argument('--force-delete', dest='force_delete', action='store_true', help="Delete files from archive_url without asking")73    create_parser.add_argument('--overwrite', dest='overwrite', action='store_true', help="Overwrite existing manifest_path and tar_path without asking")74    create_parser.set_defaults(func=archive_command, validate=True, delete=False, force_delete=False, overwrite=False)75    # validate76    create_parser = subparsers.add_parser('validate', help='Validate an existing tar archive and manifest.')77    create_parser.add_argument('manifest_path', help='Path or URL for manifest file')78    create_parser.add_argument('tar_path', help='Path or URL for tar file')79    create_parser.set_defaults(func=validate_command)80    # delete81    create_parser = subparsers.add_parser('delete', help='Delete original files listed in manifest.')82    create_parser.add_argument('manifest_path', help='Path or URL for manifest file')83    create_parser.add_argument('tar_path', nargs='?', help='Path or URL for tar file')84    create_parser.add_argument('--no-validate', dest='validate', action='store_false', help="Don't validate tar against manifest before deleting")85    create_parser.add_argument('--force-delete', dest='force_delete', action='store_true', help="Delete without asking")86    create_parser.set_defaults(func=delete_command, validate=True, force_delete=False)87    # extract88    create_parser = subparsers.add_parser('extract', help='Extract a file from an archive.')89    create_parser.add_argument('manifest_path', help='Path or URL for manifest file')90    create_parser.add_argument('tar_path', help='Path or URL for tar file')91    create_parser.add_argument('file_path', help='URL of file to extract from manifest, e.g. s3://<Bucket>/<Key>')92    create_parser.add_argument('--out', help='optional output path; default stdout')93    create_parser.set_defaults(func=extract_command)94    args = parser.parse_args(args)95    if hasattr(args, 'func'):96        args.func(args, parser)97    else:98        parser.print_help()...

test_parse.py

Source:test_parse.py

...6from network import Network7from monitors import Monitors8#9@pytest.fixture10def create_parser(input_file):11    """Return a new parser instance."""12    path = "./test_specfiles/test_parse/" + input_file13    # Initialise instances of the four inner simulator classes14    names = Names()15    devices = Devices(names)16    network = Network(names, devices)17    monitors = Monitors(names, devices, network)18    scanner = Scanner(path, names)19    return Parser(names, devices, network, monitors, scanner)20@pytest.mark.parametrize("input_file, errors,  success", [21    ("circuit1_specfile.txt", [], True),22    ("circuit2_specfile.txt", ["create_parser.NO_DEVICES"], False),23    ("circuit3_specfile.txt", ['create_parser.NO_CONNECTIONS'], False),24    ("circuit4_specfile.txt", ['create_parser.NO_MONITORS'], False),...

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.