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

How to use enable_enhanced_monitoring method in localstack

Best Python code snippet using localstack_python

kinesis_monitoring.py

Source:kinesis_monitoring.py

...88    if not isinstance(stream, basestring):89        stream = stream.id90    role = get_iam_role_for_stream(stream)91    kinesis_client = aws_common.connect_kinesis(role=role)92    return kinesis_client.enable_enhanced_monitoring(StreamName=stream, ShardLevelMetrics=metrics)93def disable_shard_monitoring(stream, metrics=['ALL']):94    if not isinstance(stream, basestring):95        stream = stream.id96    role = get_iam_role_for_stream(stream)97    kinesis_client = aws_common.connect_kinesis(role=role)98    return kinesis_client.disable_enhanced_monitoring(StreamName=stream, ShardLevelMetrics=metrics)99def get_kinesis_cloudwatch_metrics(stream, metric, shard=None):100    dimensions = [{'Name': 'StreamName', 'Value': stream.id}]101    if shard:102        shard = shard if isinstance(shard, basestring) else shard['id'] if isinstance(shard, dict) else shard.id103        dimensions.append({'Name': 'ShardId', 'Value': shard})104    role = get_iam_role_for_stream(stream)105    return get_cloudwatch_metrics(metric=metric, namespace='AWS/Kinesis', dimensions=dimensions, role=role)106def replace_nan(value, default=0):...

test_kinesis_monitoring.py

Source:test_kinesis_monitoring.py

...4def test_enable_enhanced_monitoring_all():5    client = boto3.client("kinesis", region_name="us-east-1")6    stream_name = "my_stream_summary"7    client.create_stream(StreamName=stream_name, ShardCount=4)8    resp = client.enable_enhanced_monitoring(9        StreamName=stream_name, ShardLevelMetrics=["ALL"]10    )11    resp.should.have.key("StreamName").equals(stream_name)12    resp.should.have.key("CurrentShardLevelMetrics").equals([])13    resp.should.have.key("DesiredShardLevelMetrics").equals(["ALL"])14@mock_kinesis15def test_enable_enhanced_monitoring_is_persisted():16    client = boto3.client("kinesis", region_name="us-east-1")17    stream_name = "my_stream_summary"18    client.create_stream(StreamName=stream_name, ShardCount=4)19    client.enable_enhanced_monitoring(20        StreamName=stream_name, ShardLevelMetrics=["IncomingBytes", "OutgoingBytes"]21    )22    stream = client.describe_stream(StreamName=stream_name)["StreamDescription"]23    metrics = stream["EnhancedMonitoring"][0]["ShardLevelMetrics"]24    set(metrics).should.equal({"IncomingBytes", "OutgoingBytes"})25@mock_kinesis26def test_enable_enhanced_monitoring_in_steps():27    client = boto3.client("kinesis", region_name="us-east-1")28    stream_name = "my_stream_summary"29    client.create_stream(StreamName=stream_name, ShardCount=4)30    client.enable_enhanced_monitoring(31        StreamName=stream_name, ShardLevelMetrics=["IncomingBytes", "OutgoingBytes"]32    )33    resp = client.enable_enhanced_monitoring(34        StreamName=stream_name, ShardLevelMetrics=["WriteProvisionedThroughputExceeded"]35    )36    resp.should.have.key("CurrentShardLevelMetrics").should.have.length_of(2)37    resp["CurrentShardLevelMetrics"].should.contain("IncomingBytes")38    resp["CurrentShardLevelMetrics"].should.contain("OutgoingBytes")39    resp.should.have.key("DesiredShardLevelMetrics").should.have.length_of(3)40    resp["DesiredShardLevelMetrics"].should.contain("IncomingBytes")41    resp["DesiredShardLevelMetrics"].should.contain("OutgoingBytes")42    resp["DesiredShardLevelMetrics"].should.contain(43        "WriteProvisionedThroughputExceeded"44    )45    stream = client.describe_stream(StreamName=stream_name)["StreamDescription"]46    metrics = stream["EnhancedMonitoring"][0]["ShardLevelMetrics"]47    metrics.should.have.length_of(3)48    metrics.should.contain("IncomingBytes")49    metrics.should.contain("OutgoingBytes")50    metrics.should.contain("WriteProvisionedThroughputExceeded")51@mock_kinesis52def test_disable_enhanced_monitoring():53    client = boto3.client("kinesis", region_name="us-east-1")54    stream_name = "my_stream_summary"55    client.create_stream(StreamName=stream_name, ShardCount=4)56    client.enable_enhanced_monitoring(57        StreamName=stream_name,58        ShardLevelMetrics=[59            "IncomingBytes",60            "OutgoingBytes",61            "WriteProvisionedThroughputExceeded",62        ],63    )64    resp = client.disable_enhanced_monitoring(65        StreamName=stream_name, ShardLevelMetrics=["OutgoingBytes"]66    )67    resp.should.have.key("CurrentShardLevelMetrics").should.have.length_of(3)68    resp["CurrentShardLevelMetrics"].should.contain("IncomingBytes")69    resp["CurrentShardLevelMetrics"].should.contain("OutgoingBytes")70    resp["CurrentShardLevelMetrics"].should.contain(71        "WriteProvisionedThroughputExceeded"72    )73    resp.should.have.key("DesiredShardLevelMetrics").should.have.length_of(2)74    resp["DesiredShardLevelMetrics"].should.contain("IncomingBytes")75    resp["DesiredShardLevelMetrics"].should.contain(76        "WriteProvisionedThroughputExceeded"77    )78    stream = client.describe_stream(StreamName=stream_name)["StreamDescription"]79    metrics = stream["EnhancedMonitoring"][0]["ShardLevelMetrics"]80    metrics.should.have.length_of(2)81    metrics.should.contain("IncomingBytes")82    metrics.should.contain("WriteProvisionedThroughputExceeded")83@mock_kinesis84def test_disable_enhanced_monitoring_all():85    client = boto3.client("kinesis", region_name="us-east-1")86    stream_name = "my_stream_summary"87    client.create_stream(StreamName=stream_name, ShardCount=4)88    client.enable_enhanced_monitoring(89        StreamName=stream_name,90        ShardLevelMetrics=[91            "IncomingBytes",92            "OutgoingBytes",93            "WriteProvisionedThroughputExceeded",94        ],95    )96    client.disable_enhanced_monitoring(97        StreamName=stream_name, ShardLevelMetrics=["ALL"]98    )99    stream = client.describe_stream(StreamName=stream_name)["StreamDescription"]100    metrics = stream["EnhancedMonitoring"][0]["ShardLevelMetrics"]...

lambda_function.py

Source:lambda_function.py

...16helper = CfnResource(json_logging=True)17def _enhance_monitoring(stream_name, enable):18    shard_level_metrics = ['ALL']19    if enable == 'true':20        response = client.enable_enhanced_monitoring(21            StreamName=stream_name,22            ShardLevelMetrics=shard_level_metrics23        )24    else:25        response = client.disable_enhanced_monitoring(26            StreamName=stream_name,27            ShardLevelMetrics=shard_level_metrics28        )29    return response['DesiredShardLevelMetrics']30@helper.create31@helper.update32def manage_enhanced_monitoring(event, _):33    _enhance_monitoring(34        event['ResourceProperties']['StreamName'],...

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.