KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_defaults method in tox
Best Python code snippet using tox_python
test_runner.py
Source:test_runner.py 
1# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#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# A copy of the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "LICENSE.txt" file accompanying this file.10# This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, express or implied.11# See the License for the specific language governing permissions and limitations under the License.12import datetime13import logging14import multiprocessing15import os16import re17import sys18import time19import urllib.request20from tempfile import TemporaryDirectory21import argparse22import boto323import pytest24from assertpy import assert_that25from framework.tests_configuration.config_renderer import dump_rendered_config_file, read_config_file26from framework.tests_configuration.config_utils import get_all_regions27from framework.tests_configuration.config_validator import assert_valid_config28from reports_generator import generate_cw_report, generate_json_report, generate_junitxml_merged_report29from retrying import retry30from utils import InstanceTypesData31logger = logging.getLogger()32logging.basicConfig(format="%(asctime)s - %(levelname)s - %(module)s - %(message)s", level=logging.INFO)33START_TIME = time.time()34START_TIME_ISO = datetime.datetime.fromtimestamp(START_TIME).isoformat()35LOGS_DIR = "{0}.logs".format(START_TIME)36OUT_DIR = "{0}.out".format(START_TIME)37TEST_DEFAULTS = {38 "parallelism": None,39 "retry_on_failures": False,40 "features": "", # empty string means all41 "regions": [],42 "oss": [],43 "schedulers": [],44 "instances": [],45 "dry_run": False,46 "reports": [],47 "cw_region": "us-east-1",48 "cw_namespace": "ParallelCluster/IntegrationTests",49 "cw_timestamp_day_start": False,50 "sequential": False,51 "output_dir": "tests_outputs",52 "custom_node_url": None,53 "custom_cookbook_url": None,54 "createami_custom_cookbook_url": None,55 "cookbook_git_ref": None,56 "node_git_ref": None,57 "ami_owner": None,58 "createami_custom_node_url": None,59 "custom_awsbatchcli_url": None,60 "custom_ami": None,61 "pre_install": None,62 "post_install": None,63 "vpc_stack": None,64 "api_uri": None,65 "cluster": None,66 "api_definition_s3_uri": None,67 "api_infrastructure_s3_uri": None,68 "public_ecr_image_uri": None,69 "no_delete": False,70 "benchmarks": False,71 "benchmarks_target_capacity": 200,72 "benchmarks_max_time": 30,73 "stackname_suffix": "",74 "delete_logs_on_success": False,75 "tests_root_dir": "./tests",76 "instance_types_data": None,77 "use_default_iam_credentials": False,78 "iam_user_role_stack_name": None,79 "directory_stack_name": None,80 "ldaps_nlb_stack_name": None,81 "slurm_database_stack_name": None,82}83def _init_argparser():84 parser = argparse.ArgumentParser(85 description="Run integration tests suite.", formatter_class=argparse.ArgumentDefaultsHelpFormatter86 )87 parser.add_argument("--key-name", help="Key to use for EC2 instances", required=True)88 parser.add_argument("--key-path", help="Path to the key to use for SSH connections", required=True, type=_is_file)89 parser.add_argument(90 "-n", "--parallelism", help="Tests parallelism for every region.", default=TEST_DEFAULTS.get("parallelism")91 )92 parser.add_argument(93 "--sequential",94 help="Run tests in a single process. When not specified tests will spawn a process for each region under test.",95 action="store_true",96 default=TEST_DEFAULTS.get("sequential"),97 )98 parser.add_argument(99 "--credential",100 action="append",101 help="STS credential to assume when running tests in a specific region."102 "Credentials need to be in the format <region>,<endpoint>,<ARN>,<externalId> and can"103 " be specified multiple times. <region> represents the region credentials are used for, <endpoint> is the sts "104 " endpoint to contact in order to assume credentials, <account-id> is the id of the account where the role to "105 " assume is defined, <externalId> is the id to use when assuming the role. "106 "(e.g. ap-east-1,https://sts.us-east-1.amazonaws.com,arn:aws:iam::<account-id>:role/role-to-assume,externalId)",107 required=False,108 )109 parser.add_argument(110 "--use-default-iam-credentials",111 help="Use the default IAM creds to run pcluster CLI commands. Skips the creation of pcluster CLI IAM role.",112 action="store_true",113 default=TEST_DEFAULTS.get("use_default_iam_credentials"),114 )115 parser.add_argument(116 "--retry-on-failures",117 help="Retry once more the failed tests after a delay of 60 seconds.",118 action="store_true",119 default=TEST_DEFAULTS.get("retry_on_failures"),120 )121 parser.add_argument(122 "--tests-root-dir",123 help="Root dir where integration tests are defined",124 default=TEST_DEFAULTS.get("tests_root_dir"),125 )126 dimensions_group = parser.add_argument_group("Test dimensions")127 dimensions_group.add_argument(128 "-c",129 "--tests-config",130 help="Config file that specifies the tests to run and the dimensions to enable for each test. "131 "Note that when a config file is used the following flags are ignored: instances, regions, oss, schedulers. "132 "Refer to the docs for further details on the config format: "133 "https://github.com/aws/aws-parallelcluster/blob/develop/tests/integration-tests/README.md",134 )135 dimensions_group.add_argument(136 "-i",137 "--instances",138 help="AWS instances under test. Ignored when tests-config is used.",139 default=TEST_DEFAULTS.get("instances"),140 nargs="*",141 )142 dimensions_group.add_argument(143 "-o",144 "--oss",145 help="OSs under test. Ignored when tests-config is used.",146 default=TEST_DEFAULTS.get("oss"),147 nargs="*",148 )149 dimensions_group.add_argument(150 "-s",151 "--schedulers",152 help="Schedulers under test. Ignored when tests-config is used.",153 default=TEST_DEFAULTS.get("schedulers"),154 nargs="*",155 )156 dimensions_group.add_argument(157 "-r",158 "--regions",159 help="AWS regions where tests are executed. Ignored when tests-config is used.",160 default=TEST_DEFAULTS.get("regions"),161 nargs="*",162 )163 dimensions_group.add_argument(164 "-f",165 "--features",166 help="Run only tests for the listed features. Prepending the not keyword to the feature name causes the "167 "feature to be excluded.",168 default=TEST_DEFAULTS.get("features"),169 nargs="+",170 )171 reports_group = parser.add_argument_group("Test reports")172 reports_group.add_argument(173 "--show-output",174 help="Do not redirect tests stdout to file. Not recommended when running in multiple regions.",175 action="store_true",176 default=TEST_DEFAULTS.get("show_output"),177 )178 reports_group.add_argument(179 "--reports",180 help="create tests report files. junitxml creates a junit-xml style report file. html creates an html "181 "style report file. json creates a summary with details for each dimensions. cw publishes tests metrics into "182 "CloudWatch",183 nargs="+",184 choices=["html", "junitxml", "json", "cw"],185 default=TEST_DEFAULTS.get("reports"),186 )187 reports_group.add_argument(188 "--cw-region", help="Region where to publish CloudWatch metrics", default=TEST_DEFAULTS.get("cw_region")189 )190 reports_group.add_argument(191 "--cw-namespace",192 help="CloudWatch namespace where to publish metrics",193 default=TEST_DEFAULTS.get("cw_namespace"),194 )195 reports_group.add_argument(196 "--cw-timestamp-day-start",197 action="store_true",198 help="CloudWatch metrics pushed with at timestamp equal to the start of the current day (midnight)",199 default=TEST_DEFAULTS.get("cw_timestamp_day_start"),200 )201 reports_group.add_argument(202 "--output-dir", help="Directory where tests outputs are generated", default=TEST_DEFAULTS.get("output_dir")203 )204 custom_group = parser.add_argument_group("Custom packages and templates")205 custom_group.add_argument(206 "--custom-node-url",207 help="URL to a custom node package.",208 default=TEST_DEFAULTS.get("custom_node_url"),209 type=_is_url,210 )211 custom_group.add_argument(212 "--custom-cookbook-url",213 help="URL to a custom cookbook package.",214 default=TEST_DEFAULTS.get("custom_cookbook_url"),215 type=_is_url,216 )217 custom_group.add_argument(218 "--createami-custom-cookbook-url",219 help="URL to a custom cookbook package for the createami command.",220 default=TEST_DEFAULTS.get("createami_custom_cookbook_url"),221 type=_is_url,222 )223 custom_group.add_argument(224 "--createami-custom-node-url",225 help="URL to a custom node package for the createami command.",226 default=TEST_DEFAULTS.get("createami_custom_node_url"),227 type=_is_url,228 )229 custom_group.add_argument(230 "--custom-awsbatchcli-url",231 help="URL to a custom awsbatch cli package.",232 default=TEST_DEFAULTS.get("custom_awsbatchcli_url"),233 type=_is_url,234 )235 custom_group.add_argument(236 "--pre-install", help="URL to a pre install script", default=TEST_DEFAULTS.get("pre_install")237 )238 custom_group.add_argument(239 "--post-install", help="URL to a post install script", default=TEST_DEFAULTS.get("post_install")240 )241 custom_group.add_argument(242 "--instance-types-data",243 help="Additional information about instance types used in the tests. The format is a JSON map "244 "instance_type -> data, where data must respect the same structure returned by ec2 "245 "describe-instance-types",246 default=TEST_DEFAULTS.get("instance_types_data"),247 )248 ami_group = parser.add_argument_group("AMI selection parameters")249 ami_group.add_argument(250 "--custom-ami", help="custom AMI to use for all tests.", default=TEST_DEFAULTS.get("custom_ami")251 )252 ami_group.add_argument(253 "--pcluster-git-ref",254 help="Git ref of the custom cli package used to build the AMI.",255 default=TEST_DEFAULTS.get("pcluster_git_ref"),256 )257 ami_group.add_argument(258 "--cookbook-git-ref",259 help="Git ref of the custom cookbook package used to build the AMI.",260 default=TEST_DEFAULTS.get("cookbook_git_ref"),261 )262 ami_group.add_argument(263 "--node-git-ref",264 help="Git ref of the custom node package used to build the AMI.",265 default=TEST_DEFAULTS.get("node_git_ref"),266 )267 ami_group.add_argument(268 "--ami-owner",269 help="Override the owner value when fetching AMIs to use with cluster. By default pcluster uses amazon.",270 default=TEST_DEFAULTS.get("ami_owner"),271 )272 banchmarks_group = parser.add_argument_group("Benchmarks")273 banchmarks_group.add_argument(274 "--benchmarks",275 help="run benchmarks tests. This disables the execution of all tests defined under the tests directory.",276 action="store_true",277 default=TEST_DEFAULTS.get("benchmarks"),278 )279 banchmarks_group.add_argument(280 "--benchmarks-target-capacity",281 help="set the target capacity for benchmarks tests",282 default=TEST_DEFAULTS.get("benchmarks_target_capacity"),283 type=int,284 )285 banchmarks_group.add_argument(286 "--benchmarks-max-time",287 help="set the max waiting time in minutes for benchmarks tests",288 default=TEST_DEFAULTS.get("benchmarks_max_time"),289 type=int,290 )291 api_group = parser.add_argument_group("API options")292 api_group.add_argument(293 "--api-definition-s3-uri",294 help="URI of the Docker image for the Lambda of the ParallelCluster API",295 default=TEST_DEFAULTS.get("api_definition_s3_uri"),296 )297 api_group.add_argument(298 "--api-infrastructure-s3-uri",299 help="URI of the CloudFormation template for the ParallelCluster API",300 default=TEST_DEFAULTS.get("api_definition_s3_uri"),301 )302 api_group.add_argument(303 "--public-ecr-image-uri",304 help="S3 URI of the ParallelCluster API spec",305 default=TEST_DEFAULTS.get("public_ecr_image_uri"),306 )307 api_group.add_argument(308 "--api-uri",309 help="URI of an existing ParallelCluster API",310 default=TEST_DEFAULTS.get("api_uri"),311 )312 debug_group = parser.add_argument_group("Debugging/Development options")313 debug_group.add_argument(314 "--vpc-stack", help="Name of an existing vpc stack.", default=TEST_DEFAULTS.get("vpc_stack")315 )316 debug_group.add_argument(317 "--cluster", help="Use an existing cluster instead of creating one.", default=TEST_DEFAULTS.get("cluster")318 )319 debug_group.add_argument(320 "--no-delete",321 action="store_true",322 help="Don't delete stacks after tests are complete.",323 default=TEST_DEFAULTS.get("no_delete"),324 )325 debug_group.add_argument(326 "--delete-logs-on-success",327 help="delete CloudWatch logs when a test succeeds",328 action="store_true",329 default=TEST_DEFAULTS.get("delete_logs_on_success"),330 )331 debug_group.add_argument(332 "--stackname-suffix",333 help="set a suffix in the integration tests stack names",334 default=TEST_DEFAULTS.get("stackname_suffix"),335 )336 debug_group.add_argument(337 "--dry-run",338 help="Only show the list of tests that would run with specified options.",339 action="store_true",340 default=TEST_DEFAULTS.get("dry_run"),341 )342 debug_group.add_argument(343 "--iam-user-role-stack-name",344 help="Name of an existing IAM user role stack.",345 default=TEST_DEFAULTS.get("iam_user_role_stack_name"),346 )347 debug_group.add_argument(348 "--directory-stack-name",349 help="Name of CFN stack providing AD domain to be used for testing AD integration feature.",350 default=TEST_DEFAULTS.get("directory_stack_name"),351 )352 debug_group.add_argument(353 "--ldaps-nlb-stack-name",354 help="Name of CFN stack providing NLB to enable use of LDAPS with a Simple AD directory when testing AD "355 "integration feature.",356 default=TEST_DEFAULTS.get("ldaps_nlb_stack_name"),357 )358 debug_group.add_argument(359 "--slurm-database-stack-name",360 help="Name of CFN stack providing database stack to be used for testing Slurm accounting feature.",361 default=TEST_DEFAULTS.get("slurm_database_stack_name"),362 )363 return parser364def _is_file(value):365 if not os.path.isfile(value):366 raise argparse.ArgumentTypeError("'{0}' is not a valid file".format(value))367 return value368@retry(stop_max_attempt_number=10, wait_fixed=5000)369def _is_url(value):370 scheme = urllib.request.urlparse(value).scheme371 if scheme in ["https", "s3", "file"]:372 try:373 if scheme == "s3":374 match = re.match(r"s3://(.*?)/(.*)", value)375 if not match or len(match.groups()) < 2:376 raise argparse.ArgumentTypeError(f"'{value}' is not a valid S3url")377 else:378 bucket_name, object_name = match.group(1), match.group(2)379 boto3.client("s3").head_object(Bucket=bucket_name, Key=object_name)380 else:381 urllib.request.urlopen(value)382 return value383 except Exception as e:384 raise argparse.ArgumentTypeError(f"'{value}' is not a valid url:{e}")385 else:386 raise argparse.ArgumentTypeError("'{0}' is not a valid url".format(value))387def _test_config_file(value):388 _is_file(value)389 try:390 config = read_config_file(value)391 return config392 except Exception:393 raise argparse.ArgumentTypeError("'{0}' is not a valid test config".format(value))394def _join_with_not(args):395 """396 Join 'not' with next token, so they397 can be used together as single pytest marker398 """399 it = iter(args)400 while True:401 try:402 current = next(it)403 except StopIteration:404 break405 if current == "not":406 try:407 current += " " + next(it)408 except StopIteration:409 raise Exception("'not' needs to be always followed by an item")410 yield current411def _get_pytest_args(args, regions, log_file, out_dir): # noqa: C901412 pytest_args = ["-s", "-vv", "-l"]413 pytest_args.append("--tests-log-file={0}/{1}".format(args.output_dir, log_file))414 pytest_args.append("--output-dir={0}/{1}".format(args.output_dir, out_dir))415 pytest_args.append(f"--key-name={args.key_name}")416 pytest_args.append(f"--key-path={args.key_path}")417 pytest_args.extend(["--stackname-suffix", args.stackname_suffix])418 pytest_args.extend(["--rootdir", args.tests_root_dir])419 pytest_args.append("--ignore=./benchmarks")420 if args.benchmarks:421 pytest_args.append("--benchmarks")422 # Show all tests durations423 pytest_args.append("--durations=0")424 # Run only tests with the given markers425 if args.features:426 pytest_args.append("-m")427 pytest_args.append(" or ".join(list(_join_with_not(args.features))))428 if args.tests_config:429 _set_tests_config_args(args, pytest_args, out_dir)430 if args.instance_types_data:431 pytest_args.append("--instance-types-data-file={0}".format(args.instance_types_data))432 if regions:433 pytest_args.append("--regions")434 pytest_args.extend(regions)435 if args.instances:436 pytest_args.append("--instances")437 pytest_args.extend(args.instances)438 if args.oss:439 pytest_args.append("--oss")440 pytest_args.extend(args.oss)441 if args.schedulers:442 pytest_args.append("--schedulers")443 pytest_args.extend(args.schedulers)444 if args.delete_logs_on_success:445 pytest_args.append("--delete-logs-on-success")446 if args.credential:447 pytest_args.append("--credential")448 pytest_args.extend(args.credential)449 if args.use_default_iam_credentials:450 pytest_args.append("--use-default-iam-credentials")451 if args.retry_on_failures:452 # Rerun tests on failures for one more time after 60 seconds delay453 pytest_args.extend(["--reruns", "1", "--reruns-delay", "60"])454 if args.parallelism:455 pytest_args.extend(["-n", args.parallelism])456 if args.dry_run:457 pytest_args.append("--collect-only")458 if any(report in ["junitxml", "json", "cw"] for report in args.reports):459 pytest_args.append("--junit-xml={0}/{1}/results.xml".format(args.output_dir, out_dir))460 if "html" in args.reports:461 pytest_args.append("--html={0}/{1}/results.html".format(args.output_dir, out_dir))462 _set_custom_packages_args(args, pytest_args)463 _set_ami_args(args, pytest_args)464 _set_custom_stack_args(args, pytest_args)465 _set_api_args(args, pytest_args)466 return pytest_args467def _set_custom_packages_args(args, pytest_args): # noqa: C901468 if args.custom_node_url:469 pytest_args.extend(["--custom-node-package", args.custom_node_url])470 if args.custom_cookbook_url:471 pytest_args.extend(["--custom-chef-cookbook", args.custom_cookbook_url])472 if args.createami_custom_cookbook_url:473 pytest_args.extend(["--createami-custom-chef-cookbook", args.createami_custom_cookbook_url])474 if args.createami_custom_node_url:475 pytest_args.extend(["--createami-custom-node-package", args.createami_custom_node_url])476 if args.custom_awsbatchcli_url:477 pytest_args.extend(["--custom-awsbatchcli-package", args.custom_awsbatchcli_url])478 if args.pre_install:479 pytest_args.extend(["--pre-install", args.pre_install])480 if args.post_install:481 pytest_args.extend(["--post-install", args.post_install])482def _set_ami_args(args, pytest_args):483 if args.custom_ami:484 pytest_args.extend(["--custom-ami", args.custom_ami])485 if args.pcluster_git_ref:486 pytest_args.extend(["--pcluster-git-ref", args.pcluster_git_ref])487 if args.cookbook_git_ref:488 pytest_args.extend(["--cookbook-git-ref", args.cookbook_git_ref])489 if args.node_git_ref:490 pytest_args.extend(["--node-git-ref", args.node_git_ref])491 if args.ami_owner:492 pytest_args.extend(["--ami-owner", args.ami_owner])493def _set_custom_stack_args(args, pytest_args):494 if args.vpc_stack:495 pytest_args.extend(["--vpc-stack", args.vpc_stack])496 if args.cluster:497 pytest_args.extend(["--cluster", args.cluster])498 if args.no_delete:499 pytest_args.append("--no-delete")500 if args.iam_user_role_stack_name:501 pytest_args.extend(["--iam-user-role-stack-name", args.iam_user_role_stack_name])502 if args.directory_stack_name:503 pytest_args.extend(["--directory-stack-name", args.directory_stack_name])504 if args.ldaps_nlb_stack_name:505 pytest_args.extend(["--ldaps-nlb-stack-name", args.ldaps_nlb_stack_name])506 if args.slurm_database_stack_name:507 pytest_args.extend(["--slurm-database-stack-name", args.slurm_database_stack_name])508def _set_api_args(args, pytest_args):509 if args.api_definition_s3_uri:510 pytest_args.extend(["--api-definition-s3-uri", args.api_definition_s3_uri])511 if args.public_ecr_image_uri:512 pytest_args.extend(["--public-ecr-image-uri", args.public_ecr_image_uri])513 if args.api_uri:514 pytest_args.extend(["--api-uri", args.api_uri])515 if args.api_infrastructure_s3_uri:516 pytest_args.extend(["--api-infrastructure-s3-uri", args.api_infrastructure_s3_uri])517def _set_tests_config_args(args, pytest_args, out_dir):518 # Dump the rendered file to avoid re-rendering in pytest processes519 rendered_config_file = f"{args.output_dir}/{out_dir}/tests_config.yaml"520 with open(rendered_config_file, "x", encoding="utf-8") as text_file:521 text_file.write(dump_rendered_config_file(args.tests_config))522 pytest_args.append(f"--tests-config-file={rendered_config_file}")523def _get_pytest_regionalized_args(region, args, our_dir, logs_dir):524 return _get_pytest_args(525 args=args,526 regions=[region],527 log_file="{0}/{1}.log".format(logs_dir, region),528 out_dir="{0}/{1}".format(our_dir, region),529 )530def _get_pytest_non_regionalized_args(args, out_dir, logs_dir):531 return _get_pytest_args(532 args=args, regions=args.regions, log_file="{0}/all_regions.log".format(logs_dir), out_dir=out_dir533 )534def _run_test_in_region(region, args, out_dir, logs_dir):535 out_dir_region = "{base_dir}/{out_dir}/{region}".format(base_dir=args.output_dir, out_dir=out_dir, region=region)536 os.makedirs(out_dir_region, exist_ok=True)537 # Redirect stdout to file538 if not args.show_output:539 sys.stdout = open("{0}/pytest.out".format(out_dir_region), "w")540 pytest_args_regionalized = _get_pytest_regionalized_args(region, args, out_dir, logs_dir)541 with TemporaryDirectory() as temp_dir:542 pytest_args_regionalized.extend(["--basetemp", temp_dir])543 logger.info("Starting pytest in region {0} with params {1}".format(region, pytest_args_regionalized))544 pytest.main(pytest_args_regionalized)545def _make_logging_dirs(base_dir):546 logs_dir = "{base_dir}/{logs_dir}".format(base_dir=base_dir, logs_dir=LOGS_DIR)547 os.makedirs(logs_dir, exist_ok=True)548 logger.info("Configured logs dir: {0}".format(logs_dir))549 out_dir = "{base_dir}/{out_dir}".format(base_dir=base_dir, out_dir=OUT_DIR)550 os.makedirs(out_dir, exist_ok=True)551 logger.info("Configured tests output dir: {0}".format(out_dir))552def _run_parallel(args):553 jobs = []554 if args.regions:555 enabled_regions = args.regions556 else:557 enabled_regions = get_all_regions(args.tests_config)558 for region in enabled_regions:559 p = multiprocessing.Process(target=_run_test_in_region, args=(region, args, OUT_DIR, LOGS_DIR))560 jobs.append(p)561 p.start()562 for job in jobs:563 job.join()564def _check_args(args):565 # If --cluster is set only one os, scheduler, instance type and region can be provided566 if args.cluster:567 if len(args.oss) > 1 or len(args.schedulers) > 1 or len(args.instances) > 1 or len(args.regions) > 1:568 logger.error(569 "when cluster option is specified, you can have a single value for oss, regions, instances "570 "and schedulers and you need to make sure they match the cluster specific ones"571 )572 exit(1)573 if not args.tests_config:574 assert_that(args.regions).described_as("--regions cannot be empty").is_not_empty()575 assert_that(args.instances).described_as("--instances cannot be empty").is_not_empty()576 assert_that(args.oss).described_as("--oss cannot be empty").is_not_empty()577 assert_that(args.schedulers).described_as("--schedulers cannot be empty").is_not_empty()578 else:579 try:580 args.tests_config = _test_config_file(args.tests_config)581 assert_valid_config(args.tests_config, args.tests_root_dir)582 logger.info("Found valid config file:\n%s", dump_rendered_config_file(args.tests_config))583 except Exception:584 raise argparse.ArgumentTypeError("'{0}' is not a valid test config".format(args.tests_config))585def _run_sequential(args):586 # Redirect stdout to file587 if not args.show_output:588 sys.stdout = open("{0}/{1}/pytest.out".format(args.output_dir, OUT_DIR), "w")589 pytest_args_non_regionalized = _get_pytest_non_regionalized_args(args, OUT_DIR, LOGS_DIR)590 logger.info("Starting pytest with params {0}".format(pytest_args_non_regionalized))591 pytest.main(pytest_args_non_regionalized)592def main():593 """Entrypoint for tests executor."""594 if sys.version_info < (3, 7):595 logger.error("test_runner requires python >= 3.7")596 exit(1)597 args = _init_argparser().parse_args()598 # Load additional instance types data, if provided.599 # This step must be done before loading test config files in order to resolve instance type placeholders.600 if args.instance_types_data:601 InstanceTypesData.load_additional_instance_types_data(args.instance_types_data)602 _check_args(args)603 logger.info("Parsed test_runner parameters {0}".format(args))604 _make_logging_dirs(args.output_dir)605 if args.sequential:606 _run_sequential(args)607 else:608 _run_parallel(args)609 logger.info("All tests completed!")610 reports_output_dir = "{base_dir}/{out_dir}".format(base_dir=args.output_dir, out_dir=OUT_DIR)611 if "junitxml" in args.reports:612 generate_junitxml_merged_report(reports_output_dir)613 if "json" in args.reports:614 logger.info("Generating tests report")615 generate_json_report(reports_output_dir)616 if "cw" in args.reports:617 logger.info("Publishing CloudWatch metrics")618 generate_cw_report(reports_output_dir, args.cw_namespace, args.cw_region, args.cw_timestamp_day_start)619if __name__ == "__main__":...
test_yamldict.py
Source:test_yamldict.py
1"""Copyright (c) 2015 Kyle James Walker2See the file LICENSE for copying permission.3"""4# -*- coding: utf-8 -*-5from __future__ import print_function6from __future__ import unicode_literals7import mock8import unittest9from mock import mock_open10from yamlsettings import (load, load_all, save_all,11 update_from_env, update_from_file, yamldict)12from . import builtin_module, path_override, open_override, isfile_override13class YamlDictTestCase(unittest.TestCase):14 def setUp(self):15 # Patch open related functions.16 open_patcher = mock.patch('{}.open'.format(builtin_module))17 open_patch = open_patcher.start()18 open_patch.side_effect = open_override19 self.addCleanup(open_patcher.stop)20 path_patcher = mock.patch('os.path.exists')21 path_patch = path_patcher.start()22 path_patch.side_effect = path_override23 self.addCleanup(path_patcher.stop)24 isfile_patcher = mock.patch('os.path.isfile')25 isfile_patch = isfile_patcher.start()26 isfile_patch.side_effect = isfile_override27 self.addCleanup(isfile_patcher.stop)28 def test_load_single_file(self):29 test_defaults = load('defaults.yml')30 self.assertEqual(test_defaults.config.greet, 'Hello')31 self.assertEqual(test_defaults.config.leave, 'Goodbye')32 self.assertEqual(test_defaults.config.secret, 'I have no secrets')33 self.assertEqual(test_defaults.config.meaning, 42)34 # Verify missing raises an AttributeError and not a KeyError35 self.assertRaises(AttributeError, getattr, test_defaults, 'missing')36 # Verify access to class attributes37 self.assertEqual(str(test_defaults.__class__),38 "<class 'yamlsettings.yamldict.YAMLDict'>")39 # Test dir access of keys for tab complete40 self.assertEqual(dir(test_defaults.config),41 ['greet', 'leave', 'meaning', 'secret'])42 # Test representation of value43 self.assertEqual(44 str(test_defaults.config),45 'greet: Hello\nleave: Goodbye\nsecret: I have no secrets\n'46 'meaning: 42\n',47 )48 self.assertEqual(49 repr(test_defaults.config),50 "{'greet': 'Hello', 'leave': 'Goodbye', "51 "'secret': 'I have no secrets', 'meaning': 42}",52 )53 # Test foo is saved in keys54 test_defaults.foo = 'bar'55 self.assertEqual(test_defaults.foo, 'bar')56 self.assertEqual(test_defaults['foo'], 'bar')57 def test_load_first_found(self):58 test_settings = load(['missing.yml', 'defaults.yml', 'settings.yml'])59 self.assertEqual(test_settings.config.greet, 'Hello')60 self.assertEqual(test_settings.config.leave, 'Goodbye')61 self.assertEqual(test_settings.config.secret, 'I have no secrets')62 self.assertEqual(test_settings.config.meaning, 42)63 @mock.patch.dict('os.environ', {64 'CONFIG_MEANING': '42.42',65 'CONFIG_SECRET': '!!python/object/apply:tests.get_secret []'})66 def test_load_with_envs(self):67 test_defaults = load('defaults.yml')68 update_from_env(test_defaults)69 self.assertEqual(test_defaults.config.greet, 'Hello')70 self.assertEqual(test_defaults.config.leave, 'Goodbye')71 self.assertEqual(test_defaults.config.secret, 's3cr3tz')72 self.assertEqual(test_defaults.config.meaning, 42.42)73 def test_update(self):74 test_settings = load('defaults.yml')75 test_settings.update(load('settings.yml'))76 self.assertEqual(test_settings.config.greet, 'Hello')77 self.assertEqual(test_settings.config.leave, 'Goodbye')78 self.assertEqual(test_settings.config.secret, 'I have many secrets')79 self.assertEqual(test_settings.config.meaning, 42)80 self.assertEqual(test_settings.config_excited.greet, "Whazzzzup!")81 def test_rebase(self):82 test_settings = load('settings.yml')83 test_settings.rebase(load('defaults.yml'))84 self.assertEqual(test_settings.config.greet, 'Hello')85 self.assertEqual(test_settings.config.leave, 'Goodbye')86 self.assertEqual(test_settings.config.secret, 'I have many secrets')87 self.assertEqual(test_settings.config.meaning, 42)88 self.assertEqual(test_settings.config_excited.greet, "Whazzzzup!")89 def test_limit(self):90 test_settings = load('settings.yml')91 test_settings.limit(['config'])92 self.assertEqual(list(test_settings), ['config'])93 test_settings2 = load('settings.yml')94 test_settings2.limit('config')95 self.assertEqual(list(test_settings), list(test_settings2))96 def test_clone_changes_isolated(self):97 test_settings = load('defaults.yml')98 test_clone = test_settings.clone()99 test_settings.config.greet = "Hodo"100 self.assertNotEqual(test_settings.config.greet,101 test_clone.config.greet)102 def test_load_all(self):103 for test_input in ['fancy.yml', ['fancy.yml']]:104 section_count = 0105 for c_yml in load_all(test_input):106 if section_count == 0:107 self.assertEqual(c_yml.test.id1.name, 'hi')108 self.assertEqual(c_yml.test.test[2].sub_test.a, 10)109 self.assertEqual(c_yml.test.test[2].sub_test.b.name, 'hi')110 elif section_count == 1:111 self.assertEqual(c_yml.test_2.test2.message, 'same here')112 elif section_count == 2:113 self.assertEqual(c_yml.test_3.test.name, 'Hello')114 section_count += 1115 self.assertEqual(section_count, 3)116 def test_save_all(self):117 cur_ymls = load_all('fancy.yml')118 m = mock_open()119 with mock.patch('{}.open'.format(builtin_module), m, create=True):120 save_all(cur_ymls, 'out.yml')121 m.assert_called_once_with('out.yml', 'w')122 def test_update_from_file(self):123 test_defaults = load('defaults.yml')124 update_from_file(test_defaults, 'settings.yml')125 self.assertEqual(test_defaults.config.secret, 'I have many secrets')126 self.assertEqual(list(test_defaults), ['config'])127 @mock.patch.dict('os.environ', {128 'TEST_GREETING_INTRODUCE': 'The environment says hello!',129 'TEST_DICT_VAR_MIX_B': 'Goodbye Variable',130 })131 def test_variable_override(self):132 test_settings = load("single_fancy.yml")133 update_from_env(test_settings)134 self.assertEqual(test_settings.test.greeting.introduce,135 'The environment says hello!')136 self.assertEqual(test_settings.test.dict_var_mix.b,137 'Goodbye Variable')138 @mock.patch.dict('os.environ', {139 'TEST_CONFIG_DB': 'OurSQL',140 })141 def test_stupid_override(self):142 test_settings = load("stupid.yml")143 update_from_env(test_settings)144 self.assertEqual(test_settings.test.config.db,145 'OurSQL')146 self.assertEqual(test_settings.test.config_db,147 'OurSQL')148 @mock.patch.dict('os.environ', {149 'TEST_GREETING_INTRODUCE': 'The environment says hello!',150 })151 def test_file_writing(self):152 test_settings = load("single_fancy.yml")153 update_from_env(test_settings)154 m = mock_open()155 with mock.patch('{}.open'.format(builtin_module), m, create=True):156 with open('current_file.yml', 'w') as h:157 h.write(str(test_settings))158 m.assert_called_once_with('current_file.yml', 'w')159 handle = m()160 handle.write.assert_called_with(161 'test:\n'162 ' id1: &id001\n'163 ' name: hi\n'164 ' id2: &id002\n'165 ' name: hello\n'166 ' var_list:\n'167 ' - *id001\n'168 ' - *id002\n'169 ' dict_var_mix:\n'170 ' a: 10\n'171 ' b: *id001\n'172 ' dict_with_list:\n'173 ' name: jin\n'174 ' set:\n'175 ' - 1\n'176 ' - 2\n'177 ' - 3\n'178 ' greeting:\n'179 ' introduce: The environment says hello!\n'180 ' part: Till we meet again\n'181 ' crazy:\n'182 ' type: !!python/name:logging.handlers.SysLogHandler \'\'\n'183 ' module: !!python/module:sys \'\'\n'184 ' instance: !!python/object:tests.SoftwareEngineer\n'185 ' name: jin\n'186 )187 def test_yaml_dict_merge(self):188 test_settings = load("merge.yml")189 # Verify the merge was successful190 self.assertEqual(test_settings.base.config.db, "MySQL")191 self.assertEqual(test_settings.merged.config.db, "MySQL")192 # Verify whoami was properly overridden193 self.assertEqual(test_settings.base.whoami, "base")194 self.assertEqual(test_settings.merged.whoami, "merged")195 def test_list_replace_on_update(self):196 test_defaults = load('defaults.yml')197 test_defaults.update({'a': [1, 2, 3]})198 self.assertEqual(test_defaults.a, [1, 2, 3])199 test_defaults.update({'a': (4,)})200 self.assertEqual(test_defaults.a, (4,))201 @mock.patch.dict('os.environ', {'FOO_BAR': 'new-baz'})202 def test_dash_vars_with_env(self):203 """Test items with dashes can be overritten with env"""204 test_settings = yamldict.YAMLDict({'foo-bar': 'baz'})205 assert test_settings['foo-bar'] == 'baz'206 update_from_env(test_settings)207 assert test_settings['foo-bar'] == 'new-baz'208if __name__ == '__main__':...
transformers.py
Source:transformers.py
1# Copyright 2008-2015 Nokia Networks2# Copyright 2016- Robot Framework Foundation3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15from ast import NodeVisitor16from robot.variables import VariableIterator17from ..model import ForLoop, Keyword18from .testsettings import TestSettings19def fixture(node, fixture_type):20 if node.name is None:21 return None22 return Keyword(node.name, args=node.args, type=fixture_type,23 lineno=node.lineno)24class SettingsBuilder(NodeVisitor):25 def __init__(self, suite, test_defaults):26 self.suite = suite27 self.test_defaults = test_defaults28 def visit_Documentation(self, node):29 self.suite.doc = node.value30 def visit_Metadata(self, node):31 self.suite.metadata[node.name] = node.value32 def visit_SuiteSetup(self, node):33 self.suite.keywords.setup = fixture(node, Keyword.SETUP_TYPE)34 def visit_SuiteTeardown(self, node):35 self.suite.keywords.teardown = fixture(node, Keyword.TEARDOWN_TYPE)36 def visit_TestSetup(self, node):37 self.test_defaults.setup = fixture(node, Keyword.SETUP_TYPE)38 def visit_TestTeardown(self, node):39 self.test_defaults.teardown = fixture(node, Keyword.TEARDOWN_TYPE)40 def visit_TestTimeout(self, node):41 self.test_defaults.timeout = node.value42 def visit_DefaultTags(self, node):43 self.test_defaults.default_tags = node.values44 def visit_ForceTags(self, node):45 self.test_defaults.force_tags = node.values46 def visit_TestTemplate(self, node):47 self.test_defaults.template = node.value48 def visit_ResourceImport(self, node):49 self.suite.resource.imports.create(type='Resource', name=node.name,50 lineno=node.lineno)51 def visit_LibraryImport(self, node):52 self.suite.resource.imports.create(type='Library', name=node.name,53 args=node.args, alias=node.alias,54 lineno=node.lineno)55 def visit_VariablesImport(self, node):56 self.suite.resource.imports.create(type='Variables', name=node.name,57 args=node.args, lineno=node.lineno)58 def visit_VariableSection(self, node):59 pass60 def visit_TestCaseSection(self, node):61 pass62 def visit_KeywordSection(self, node):63 pass64class SuiteBuilder(NodeVisitor):65 def __init__(self, suite, test_defaults):66 self.suite = suite67 self.test_defaults = test_defaults68 def visit_SettingSection(self, node):69 pass70 def visit_Variable(self, node):71 self.suite.resource.variables.create(name=node.name, value=node.value,72 lineno=node.lineno, error=node.error)73 def visit_TestCase(self, node):74 TestCaseBuilder(self.suite, self.test_defaults).visit(node)75 def visit_Keyword(self, node):76 KeywordBuilder(self.suite.resource).visit(node)77class ResourceBuilder(NodeVisitor):78 def __init__(self, resource):79 self.resource = resource80 def visit_Documentation(self, node):81 self.resource.doc = node.value82 def visit_LibraryImport(self, node):83 self.resource.imports.create(type='Library', name=node.name,84 args=node.args, alias=node.alias,85 lineno=node.lineno)86 def visit_ResourceImport(self, node):87 self.resource.imports.create(type='Resource', name=node.name,88 lineno=node.lineno)89 def visit_VariablesImport(self, node):90 self.resource.imports.create(type='Variables', name=node.name,91 args=node.args, lineno=node.lineno)92 def visit_Variable(self, node):93 self.resource.variables.create(name=node.name, value=node.value,94 lineno=node.lineno, error=node.error)95 def visit_Keyword(self, node):96 KeywordBuilder(self.resource).visit(node)97class TestCaseBuilder(NodeVisitor):98 def __init__(self, suite, defaults):99 self.suite = suite100 self.settings = TestSettings(defaults)101 self.test = None102 def visit_TestCase(self, node):103 self.test = self.suite.tests.create(name=node.name, lineno=node.lineno)104 self.generic_visit(node)105 self._set_settings(self.test, self.settings)106 def _set_settings(self, test, settings):107 test.keywords.setup = settings.setup108 test.keywords.teardown = settings.teardown109 test.timeout = settings.timeout110 test.tags = settings.tags111 if settings.template:112 test.template = settings.template113 self._set_template(test, settings.template)114 def _set_template(self, parent, template):115 for kw in parent.keywords:116 if kw.type == kw.FOR_LOOP_TYPE:117 self._set_template(kw, template)118 elif kw.type == kw.KEYWORD_TYPE:119 name, args = self._format_template(template, kw.args)120 kw.name = name121 kw.args = args122 def _format_template(self, template, arguments):123 variables = VariableIterator(template, identifiers='$')124 count = len(variables)125 if count == 0 or count != len(arguments):126 return template, arguments127 temp = []128 for (before, _, after), arg in zip(variables, arguments):129 temp.extend([before, arg])130 temp.append(after)131 return ''.join(temp), ()132 def visit_ForLoop(self, node):133 # Header and end used only for deprecation purposes. Remove in RF 3.3!134 loop = ForLoop(node.variables, node.values, node.flavor, node.lineno,135 node._header, node._end)136 ForLoopBuilder(loop).visit(node)137 self.test.keywords.append(loop)138 def visit_TemplateArguments(self, node):139 self.test.keywords.create(args=node.args, lineno=node.lineno)140 def visit_Documentation(self, node):141 self.test.doc = node.value142 def visit_Setup(self, node):143 self.settings.setup = fixture(node, Keyword.SETUP_TYPE)144 def visit_Teardown(self, node):145 self.settings.teardown = fixture(node, Keyword.TEARDOWN_TYPE)146 def visit_Timeout(self, node):147 self.settings.timeout = node.value148 def visit_Tags(self, node):149 self.settings.tags = node.values150 def visit_Template(self, node):151 self.settings.template = node.value152 def visit_KeywordCall(self, node):153 self.test.keywords.create(name=node.keyword, args=node.args,154 assign=node.assign, lineno=node.lineno)155class KeywordBuilder(NodeVisitor):156 def __init__(self, resource):157 self.resource = resource158 self.kw = None159 self.teardown = None160 def visit_Keyword(self, node):161 self.kw = self.resource.keywords.create(name=node.name,162 lineno=node.lineno)163 self.generic_visit(node)164 self.kw.keywords.teardown = self.teardown165 def visit_Documentation(self, node):166 self.kw.doc = node.value167 def visit_Arguments(self, node):168 self.kw.args = node.values169 def visit_Tags(self, node):170 self.kw.tags = node.values171 def visit_Return(self, node):172 self.kw.return_ = node.values173 def visit_Timeout(self, node):174 self.kw.timeout = node.value175 def visit_Teardown(self, node):176 self.teardown = fixture(node, Keyword.TEARDOWN_TYPE)177 def visit_KeywordCall(self, node):178 self.kw.keywords.create(name=node.keyword, args=node.args,179 assign=node.assign, lineno=node.lineno)180 def visit_ForLoop(self, node):181 # Header and end used only for deprecation purposes. Remove in RF 3.3!182 loop = ForLoop(node.variables, node.values, node.flavor, node.lineno,183 node._header, node._end)184 ForLoopBuilder(loop).visit(node)185 self.kw.keywords.append(loop)186class ForLoopBuilder(NodeVisitor):187 def __init__(self, loop):188 self.loop = loop189 def visit_KeywordCall(self, node):190 self.loop.keywords.create(name=node.keyword, args=node.args,191 assign=node.assign, lineno=node.lineno)192 def visit_TemplateArguments(self, node):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tox automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
