Best Python code snippet using molecule_python
test_healthcheck_url.py
Source:test_healthcheck_url.py
1#!/usr/bin/env python2# -*- coding: utf-8 -*-3"""4Tests for `django_auto_healthchecks.healthchecks.HealthcheckUrl` class.5"""6from __future__ import unicode_literals7try:8 import mock9except ImportError:10 from unittest import mock11import django_auto_healthchecks.healthchecks as healthchecks12from . import MockSettings13def test_hostname_from_healthcheck_settings():14 test_hostname = 'cronitor.io'15 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname})16 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)17 assert test_hostname in HcUrl.url, "URL does not contain expected hostname"18def test_hostname_detection_fallback_to_settings_hostname():19 test_hostname = 'cronitor.io'20 healthchecks.settings = MockSettings(HOSTNAME=test_hostname)21 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)22 assert test_hostname in HcUrl.url, "URL does not contain expected hostname"23def test_hostname_detection_fallback_to_allowed_hosts():24 test_hostname = 'cronitor.io'25 healthchecks.settings = MockSettings(ALLOWED_HOSTS=(test_hostname,))26 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)27 assert test_hostname in HcUrl.url, "URL does not contain expected hostname"28def test_raise_exception_when_hostname_cannot_be_detected():29 healthchecks.settings = MockSettings()30 raised = False31 try:32 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)33 except healthchecks.HealthcheckError:34 raised = True35 finally:36 assert raised, "Expected HealthcheckError exception not raised"37def test_url_contains_all_expected_parts():38 test_hostname = 'cronitor.io'39 test_path = '/path/to/endpoint'40 test_querystring = {41 'foo': 'bar',42 'bar': 'foo'43 }44 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname, 'HTTPS': False})45 HcUrl = healthchecks.HealthcheckUrl(test_path, test_querystring)46 assert 'foo=bar' in HcUrl.url, "URL does not expected contain querystring part"47 assert 'bar=foo' in HcUrl.url, "URL does not expected contain querystring part"48 assert 'http' in HcUrl.url, "URL does not contain scheme"49 assert test_hostname in HcUrl.url, "URL does not contain hostname"50 assert test_path in HcUrl.url, "URL does not contain path"51def test_https_enabled_healthcheck_settings():52 test_hostname = 'cronitor.io'53 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname, 'HTTPS': True})54 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)55 assert test_hostname in HcUrl.url, "URL does not contain expected hostname"56 assert 'https://' in HcUrl.url, "URL does not contain https://"57def test_https_disabled_healthcheck_settings():58 test_hostname = 'cronitor.io'59 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname, 'HTTPS': False})60 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)61 assert test_hostname in HcUrl.url, "URL does not contain expected hostname"62 assert 'http://' in HcUrl.url, "URL does not contains http://"63def test_https_disabled_by_default():64 test_hostname = 'cronitor.io'65 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname})66 HcUrl = healthchecks.HealthcheckUrl('/path/to/endpoint', None)67 assert 'http://' in HcUrl.url, "URL does not contains http://"68def test_querystring_in_url_not_display_url():69 test_hostname = 'cronitor.io'70 test_path = '/path/to/endpoint'71 test_querystring = {72 'foo': 'bar',73 'bar': 'foo'74 }75 healthchecks.settings = MockSettings(HEALTHCHECKS={'HOSTNAME': test_hostname, 'HTTPS': False})76 HcUrl = healthchecks.HealthcheckUrl(test_path, test_querystring)77 assert 'foo=bar' in HcUrl.url, "URL does not contains querystring"78 assert 'bar=foo' in HcUrl.url, "URL does not contains querystring"79 assert 'foo=bar' not in HcUrl.display, "Display URL unexpectedly containss querystring"...
main.py
Source:main.py
1#!/usr/bin/python2"""3Benchmark the performance / throttling of AWS Lambda,4specifically DNS queries in a VPC.5"""6import json7import os8import time9import socket10import uuid11import boto312def handler(event, context):13 """14 Sample input event:15 {16 "fanout": 8,17 "test_hostname": "*.somewhere.com",18 "duration_secs": 3019 }20 Total worker lambdas are (2 ** `fanout`), e.g.:21 - 1 fanout = 2 total children22 - 4 fanout = 16 total children23 - 8 fanout = 256 total children24 The `test_hostname` is used for DNS lookups.25 Any '*' character is replaced with a random UUID, to break caching.26 The test runs for `duration_secs` seconds, default 30 seconds.27 To query the logs, run this in cloudwatch logs insights:28 stats sum(num_ok)/5 as ok_per_second, avg(num_ok)/5 as avg_ok,29 sum(num_errors) as errors by bin(5sec)30 """31 print("event: ", event)32 fanout = event.get("fanout", 0)33 if fanout > 0:34 event["fanout"] -= 135 launch_fanout(2, event)36 elif event.get("debug", 0):37 """38 Lambda in VPC seems to use a 169.254.x resolver, instead of directly39 querying the VPC / ENI.40 """41 config = open("/etc/resolv.conf", "r").read()42 print("resolv.conf: ", config)43 else:44 duration_secs = event.get("duration_secs", 30)45 test_hostname = event.get("test_hostname", "www.amazon.com")46 run_benchmark(duration_secs, test_hostname)47def run_benchmark(duration_secs, test_hostname):48 end_time = time.time() + duration_secs49 while time.time() < end_time:50 run_benchmark_chunk(5, test_hostname)51def run_benchmark_chunk(duration_secs, test_hostname):52 """53 Run a benchmark for `duration_secs` seconds, then print stats in JSON54 format for cloudwatch logs insights.55 """56 run_start_time = time.time()57 run_end_time = run_start_time + duration_secs58 num_ok = 059 num_errors = 060 while True:61 start_time = time.time()62 if start_time > run_end_time:63 break64 try:65 # NB: Always generating the UUID here, to keep performance consistent66 actual_hostname = test_hostname.replace("*", str(uuid.uuid4()))67 response = socket.gethostbyname(actual_hostname)68 num_ok += 169 except Exception as e:70 error_secs = time.time() - start_time71 data = dict(benchmark_type="error", error_secs=error_secs, error=str(e))72 print(json.dumps(data))73 num_errors += 174 data = dict(benchmark_type="stats",75 num_ok=num_ok,76 num_errors=num_errors)77 print(json.dumps(data))78def launch_fanout(num_children, event):79 """80 Invoke `num_children` children asynchronously81 """82 client = boto3.client('lambda')83 for i in range(num_children):84 response = client.invoke(85 FunctionName=os.environ['AWS_LAMBDA_FUNCTION_NAME'],86 InvocationType='Event',87 Payload=json.dumps(event).encode("utf-8"),88 )...
test_log.py
Source:test_log.py
1import pytest2from botocore.stub import Stubber3import boto34from models.log import Log5import test_util6session = boto3.Session()7dynamodb = session.resource("dynamodb", region_name='us-west-2')8Table = dynamodb.Table(name="logging_aggregator")9@pytest.fixture10def ddb_stub():11 ddb_stub = Stubber(Table.meta.client)12 ddb_stub.activate()13 yield ddb_stub14 ddb_stub.deactivate()15def test_get_logs(mocker, ddb_stub):16 ddb_stub.add_response(17 "query", {18 "Items": [19 test_util.create_item(filename="test_filename",20 hostname="test_hostname",21 last_modified="123",22 logging_data="LOGGING DATA")23 ]24 })25 result = Log(Table).get_with_args()26 assert result[1] == 20027 assert result[0]['data'][0]['host'] == 'test_hostname'28 assert result[0]['data'][0]['filename'] == 'test_filename'29 assert result[0]['data'][0]['last_modified'] == '123'30 assert result[0]['data'][0]['logline'] == 'LOGGING DATA'31def test_get_logs_shows_newest_first(mocker, ddb_stub):32 ddb_stub.add_response(33 "query", {34 "Items": [35 test_util.create_item("test_filename", "test_hostname", "123",36 "LOGGING DATA"),37 test_util.create_item("test_filename", "test_hostname", "456",38 "LATER LINE")39 ]40 })41 result = Log(Table).get_with_args()42 assert result[0]['data'][0]['last_modified'] == '456'43 assert result[0]['data'][0]['logline'] == 'LATER LINE'44 assert result[0]['data'][1]['last_modified'] == '123'45 assert result[0]['data'][1]['logline'] == 'LOGGING DATA'46def test_get_logs_count(mocker, ddb_stub):47 ddb_stub.add_response(48 "query", {49 "Items": [50 test_util.create_item("test_filename", "test_hostname", "123",51 "LOGGING DATA"),52 test_util.create_item("test_filename", "test_hostname", "989",53 "LATEST_LINE"),54 test_util.create_item("test_filename", "test_hostname", "456",55 "LATER LINE")56 ]57 })58 result = Log(Table).get_with_args(args={'count': 2})59 assert result[0]['data'][0]['last_modified'] == '989'60 assert result[0]['data'][0]['logline'] == 'LATEST_LINE'...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!