Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_flow_logs method in localstack

Best Python code snippet using localstack_python

test_flow_logs.py

Source:test_flow_logs.py

...22        Bucket="test-flow-logs",23        CreateBucketConfiguration={"LocationConstraint": "us-west-1"},24    )25    with pytest.raises(ClientError) as ex:26        client.create_flow_logs(27            ResourceType="VPC",28            ResourceIds=[vpc["VpcId"]],29            TrafficType="ALL",30            LogDestinationType="s3",31            LogDestination="arn:aws:s3:::" + bucket.name,32            DryRun=True,33        )34    ex.value.response["Error"]["Code"].should.equal("DryRunOperation")35    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)36    ex.value.response["Error"]["Message"].should.equal(37        "An error occurred (DryRunOperation) when calling the CreateFlowLogs operation: Request would have succeeded, but DryRun flag is set"38    )39    response = client.create_flow_logs(40        ResourceType="VPC",41        ResourceIds=[vpc["VpcId"]],42        TrafficType="ALL",43        LogDestinationType="s3",44        LogDestination="arn:aws:s3:::" + bucket.name,45    )["FlowLogIds"]46    response.should.have.length_of(1)47    flow_logs = client.describe_flow_logs()["FlowLogs"]48    flow_logs.should.have.length_of(1)49    flow_log = flow_logs[0]50    flow_log["FlowLogId"].should.equal(response[0])51    flow_log["DeliverLogsStatus"].should.equal("SUCCESS")52    flow_log["FlowLogStatus"].should.equal("ACTIVE")53    flow_log["ResourceId"].should.equal(vpc["VpcId"])54    flow_log["TrafficType"].should.equal("ALL")55    flow_log["LogDestinationType"].should.equal("s3")56    flow_log["LogDestination"].should.equal("arn:aws:s3:::" + bucket.name)57    flow_log["LogFormat"].should.equal(58        "${version} ${account-id} ${interface-id} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${protocol} ${packets} ${bytes} ${start} ${end} ${action} ${log-status}"59    )60    flow_log["MaxAggregationInterval"].should.equal(600)61@mock_logs62@mock_ec263def test_create_flow_logs_cloud_watch():64    client = boto3.client("ec2", region_name="us-west-1")65    logs_client = boto3.client("logs", region_name="us-west-1")66    vpc = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]67    logs_client.create_log_group(logGroupName="test-group")68    with pytest.raises(ClientError) as ex:69        client.create_flow_logs(70            ResourceType="VPC",71            ResourceIds=[vpc["VpcId"]],72            TrafficType="ALL",73            LogDestinationType="cloud-watch-logs",74            LogGroupName="test-group",75            DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",76            DryRun=True,77        )78    ex.value.response["Error"]["Code"].should.equal("DryRunOperation")79    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)80    ex.value.response["Error"]["Message"].should.equal(81        "An error occurred (DryRunOperation) when calling the CreateFlowLogs operation: Request would have succeeded, but DryRun flag is set"82    )83    response = client.create_flow_logs(84        ResourceType="VPC",85        ResourceIds=[vpc["VpcId"]],86        TrafficType="ALL",87        LogDestinationType="cloud-watch-logs",88        LogGroupName="test-group",89        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",90    )["FlowLogIds"]91    response.should.have.length_of(1)92    flow_logs = client.describe_flow_logs()["FlowLogs"]93    flow_logs.should.have.length_of(1)94    flow_log = flow_logs[0]95    flow_log["FlowLogId"].should.equal(response[0])96    flow_log["DeliverLogsStatus"].should.equal("SUCCESS")97    flow_log["FlowLogStatus"].should.equal("ACTIVE")98    flow_log["ResourceId"].should.equal(vpc["VpcId"])99    flow_log["TrafficType"].should.equal("ALL")100    flow_log["LogDestinationType"].should.equal("cloud-watch-logs")101    flow_log["LogGroupName"].should.equal("test-group")102    flow_log["DeliverLogsPermissionArn"].should.equal(103        "arn:aws:iam::" + ACCOUNT_ID + ":role/test-role"104    )105    flow_log["LogFormat"].should.equal(106        "${version} ${account-id} ${interface-id} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${protocol} ${packets} ${bytes} ${start} ${end} ${action} ${log-status}"107    )108    flow_log["MaxAggregationInterval"].should.equal(600)109@mock_s3110@mock_ec2111def test_create_flow_log_create():112    s3 = boto3.resource("s3", region_name="us-west-1")113    client = boto3.client("ec2", region_name="us-west-1")114    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]115    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]116    bucket = s3.create_bucket(117        Bucket="test-flow-logs",118        CreateBucketConfiguration={"LocationConstraint": "us-west-1",},119    )120    response = client.create_flow_logs(121        ResourceType="VPC",122        ResourceIds=[vpc1["VpcId"], vpc2["VpcId"]],123        TrafficType="ALL",124        LogDestinationType="s3",125        LogDestination="arn:aws:s3:::" + bucket.name,126        LogFormat="${version} ${vpc-id} ${subnet-id} ${instance-id} ${interface-id} ${account-id} ${type} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${pkt-srcaddr} ${pkt-dstaddr} ${protocol} ${bytes} ${packets} ${start} ${end} ${action} ${tcp-flags} ${log-status}",127    )["FlowLogIds"]128    response.should.have.length_of(2)129    flow_logs = client.describe_flow_logs()["FlowLogs"]130    flow_logs.should.have.length_of(2)131    flow_logs[0]["LogFormat"].should.equal(132        "${version} ${vpc-id} ${subnet-id} ${instance-id} ${interface-id} ${account-id} ${type} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${pkt-srcaddr} ${pkt-dstaddr} ${protocol} ${bytes} ${packets} ${start} ${end} ${action} ${tcp-flags} ${log-status}"133    )134    flow_logs[1]["LogFormat"].should.equal(135        "${version} ${vpc-id} ${subnet-id} ${instance-id} ${interface-id} ${account-id} ${type} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${pkt-srcaddr} ${pkt-dstaddr} ${protocol} ${bytes} ${packets} ${start} ${end} ${action} ${tcp-flags} ${log-status}"136    )137@mock_s3138@mock_ec2139def test_delete_flow_logs():140    s3 = boto3.resource("s3", region_name="us-west-1")141    client = boto3.client("ec2", region_name="us-west-1")142    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]143    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]144    bucket = s3.create_bucket(145        Bucket="test-flow-logs",146        CreateBucketConfiguration={"LocationConstraint": "us-west-1"},147    )148    response = client.create_flow_logs(149        ResourceType="VPC",150        ResourceIds=[vpc1["VpcId"], vpc2["VpcId"]],151        TrafficType="ALL",152        LogDestinationType="s3",153        LogDestination="arn:aws:s3:::" + bucket.name,154    )["FlowLogIds"]155    response.should.have.length_of(2)156    flow_logs = client.describe_flow_logs()["FlowLogs"]157    flow_logs.should.have.length_of(2)158    client.delete_flow_logs(FlowLogIds=[response[0]])159    flow_logs = client.describe_flow_logs()["FlowLogs"]160    flow_logs.should.have.length_of(1)161    flow_logs[0]["FlowLogId"].should.equal(response[1])162    client.delete_flow_logs(FlowLogIds=[response[1]])163    flow_logs = client.describe_flow_logs()["FlowLogs"]164    flow_logs.should.have.length_of(0)165@mock_s3166@mock_ec2167def test_delete_flow_logs_delete_many():168    s3 = boto3.resource("s3", region_name="us-west-1")169    client = boto3.client("ec2", region_name="us-west-1")170    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]171    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]172    bucket = s3.create_bucket(173        Bucket="test-flow-logs",174        CreateBucketConfiguration={"LocationConstraint": "us-west-1"},175    )176    response = client.create_flow_logs(177        ResourceType="VPC",178        ResourceIds=[vpc1["VpcId"], vpc2["VpcId"]],179        TrafficType="ALL",180        LogDestinationType="s3",181        LogDestination="arn:aws:s3:::" + bucket.name,182    )["FlowLogIds"]183    response.should.have.length_of(2)184    flow_logs = client.describe_flow_logs()["FlowLogs"]185    flow_logs.should.have.length_of(2)186    client.delete_flow_logs(FlowLogIds=response)187    flow_logs = client.describe_flow_logs()["FlowLogs"]188    flow_logs.should.have.length_of(0)189@mock_ec2190def test_delete_flow_logs_non_existing():191    client = boto3.client("ec2", region_name="us-west-1")192    with pytest.raises(ClientError) as ex:193        client.delete_flow_logs(FlowLogIds=["fl-1a2b3c4d"])194    ex.value.response["Error"]["Code"].should.equal("InvalidFlowLogId.NotFound")195    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)196    ex.value.response["Error"]["Message"].should.equal(197        "These flow log ids in the input list are not found: [TotalCount: 1] fl-1a2b3c4d"198    )199    with pytest.raises(ClientError) as ex:200        client.delete_flow_logs(FlowLogIds=["fl-1a2b3c4d", "fl-2b3c4d5e"])201    ex.value.response["Error"]["Code"].should.equal("InvalidFlowLogId.NotFound")202    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)203    ex.value.response["Error"]["Message"].should.equal(204        "These flow log ids in the input list are not found: [TotalCount: 2] fl-1a2b3c4d fl-2b3c4d5e"205    )206@mock_ec2207def test_create_flow_logs_unsuccessful():208    s3 = boto3.resource("s3", region_name="us-west-1")209    client = boto3.client("ec2", region_name="us-west-1")210    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]211    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]212    response = client.create_flow_logs(213        ResourceType="VPC",214        ResourceIds=[vpc1["VpcId"], vpc2["VpcId"]],215        TrafficType="ALL",216        LogDestinationType="s3",217        LogDestination="arn:aws:s3:::non-existing-bucket",218    )219    response["FlowLogIds"].should.have.length_of(0)220    response["Unsuccessful"].should.have.length_of(2)221    error1 = response["Unsuccessful"][0]["Error"]222    error2 = response["Unsuccessful"][1]["Error"]223    error1["Code"].should.equal("400")224    error1["Message"].should.equal(225        "LogDestination: non-existing-bucket does not exist."226    )227    error2["Code"].should.equal("400")228    error2["Message"].should.equal(229        "LogDestination: non-existing-bucket does not exist."230    )231@mock_s3232@mock_ec2233def test_create_flow_logs_invalid_parameters():234    s3 = boto3.resource("s3", region_name="us-west-1")235    client = boto3.client("ec2", region_name="us-west-1")236    vpc = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]237    bucket = s3.create_bucket(238        Bucket="test-flow-logs",239        CreateBucketConfiguration={"LocationConstraint": "us-west-1"},240    )241    with pytest.raises(ClientError) as ex:242        client.create_flow_logs(243            ResourceType="VPC",244            ResourceIds=[vpc["VpcId"]],245            TrafficType="ALL",246            LogDestinationType="s3",247            LogDestination="arn:aws:s3:::" + bucket.name,248            MaxAggregationInterval=10,249        )250    ex.value.response["Error"]["Code"].should.equal("InvalidParameter")251    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)252    ex.value.response["Error"]["Message"].should.equal(253        "Invalid Flow Log Max Aggregation Interval"254    )255    with pytest.raises(ClientError) as ex:256        client.create_flow_logs(257            ResourceType="VPC",258            ResourceIds=[vpc["VpcId"]],259            TrafficType="ALL",260            LogDestinationType="s3",261        )262    ex.value.response["Error"]["Code"].should.equal("InvalidParameter")263    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)264    ex.value.response["Error"]["Message"].should.equal(265        "LogDestination can't be empty if LogGroupName is not provided."266    )267    with pytest.raises(ClientError) as ex:268        client.create_flow_logs(269            ResourceType="VPC",270            ResourceIds=[vpc["VpcId"]],271            TrafficType="ALL",272            LogDestinationType="s3",273            LogGroupName="test",274        )275    ex.value.response["Error"]["Code"].should.equal("InvalidParameter")276    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)277    ex.value.response["Error"]["Message"].should.equal(278        "LogDestination type must be cloud-watch-logs if LogGroupName is provided."279    )280    with pytest.raises(ClientError) as ex:281        client.create_flow_logs(282            ResourceType="VPC",283            ResourceIds=[vpc["VpcId"]],284            TrafficType="ALL",285            LogGroupName="test",286        )287    ex.value.response["Error"]["Code"].should.equal("InvalidParameter")288    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)289    ex.value.response["Error"]["Message"].should.equal(290        "DeliverLogsPermissionArn can't be empty if LogDestinationType is cloud-watch-logs."291    )292    response = client.create_flow_logs(293        ResourceType="VPC",294        ResourceIds=[vpc["VpcId"]],295        TrafficType="ALL",296        LogDestinationType="s3",297        LogDestination="arn:aws:s3:::" + bucket.name,298    )["FlowLogIds"]299    response.should.have.length_of(1)300    with pytest.raises(ClientError) as ex:301        client.create_flow_logs(302            ResourceType="VPC",303            ResourceIds=[vpc["VpcId"]],304            TrafficType="ALL",305            LogDestinationType="s3",306            LogDestination="arn:aws:s3:::" + bucket.name,307        )308    ex.value.response["Error"]["Code"].should.equal("FlowLogAlreadyExists")309    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)310    ex.value.response["Error"]["Message"].should.equal(311        "Error. There is an existing Flow Log with the same configuration and log destination."312    )313    response = client.create_flow_logs(314        ResourceType="VPC",315        ResourceIds=[vpc["VpcId"]],316        TrafficType="ALL",317        LogGroupName="test-group",318        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",319    )["FlowLogIds"]320    response.should.have.length_of(1)321    with pytest.raises(ClientError) as ex:322        client.create_flow_logs(323            ResourceType="VPC",324            ResourceIds=[vpc["VpcId"]],325            TrafficType="ALL",326            LogGroupName="test-group",327            DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",328        )329    ex.value.response["Error"]["Code"].should.equal("FlowLogAlreadyExists")330    ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)331    ex.value.response["Error"]["Message"].should.equal(332        "Error. There is an existing Flow Log with the same configuration and log destination."333    )334    flow_logs = client.describe_flow_logs()["FlowLogs"]335    flow_logs.should.have.length_of(2)336@mock_s3337@mock_ec2338@mock_logs339def test_describe_flow_logs_filtering():340    s3 = boto3.resource("s3", region_name="us-west-1")341    client = boto3.client("ec2", region_name="us-west-1")342    logs_client = boto3.client("logs", region_name="us-west-1")343    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]344    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]345    vpc3 = client.create_vpc(CidrBlock="10.2.0.0/16")["Vpc"]346    subnet1 = client.create_subnet(VpcId=vpc1["VpcId"], CidrBlock="10.0.0.0/18")[347        "Subnet"348    ]349    bucket1 = s3.create_bucket(350        Bucket="test-flow-logs-1",351        CreateBucketConfiguration={"LocationConstraint": "us-west-1"},352    )353    logs_client.create_log_group(logGroupName="test-group")354    fl1 = client.create_flow_logs(355        ResourceType="Subnet",356        ResourceIds=[subnet1["SubnetId"]],357        TrafficType="ALL",358        LogGroupName="test-group",359        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",360    )["FlowLogIds"][0]361    fl2 = client.create_flow_logs(362        ResourceType="VPC",363        ResourceIds=[vpc2["VpcId"]],364        TrafficType="Accept",365        LogDestinationType="s3",366        LogDestination="arn:aws:s3:::" + bucket1.name,367        TagSpecifications=[368            {"ResourceType": "vpc-flow-log", "Tags": [{"Key": "foo", "Value": "bar"}]}369        ],370    )["FlowLogIds"][0]371    fl3 = client.create_flow_logs(372        ResourceType="VPC",373        ResourceIds=[vpc3["VpcId"]],374        TrafficType="Reject",375        LogGroupName="non-existing-group",376        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role",377    )["FlowLogIds"][0]378    all_flow_logs = client.describe_flow_logs()["FlowLogs"]379    all_flow_logs.should.have.length_of(3)380    fl_by_deliver_status = client.describe_flow_logs(381        Filters=[{"Name": "deliver-log-status", "Values": ["SUCCESS"]}],382    )["FlowLogs"]383    fl_by_deliver_status.should.have.length_of(3)384    fl_by_s3_bucket = client.describe_flow_logs(385        Filters=[{"Name": "log-destination-type", "Values": ["s3"]}],386    )["FlowLogs"]387    fl_by_s3_bucket.should.have.length_of(1)388    fl_by_s3_bucket[0]["FlowLogId"].should.equal(fl2)389    fl_by_s3_bucket[0]["ResourceId"].should.equal(vpc2["VpcId"])390    fl_by_cloud_watch = client.describe_flow_logs(391        Filters=[{"Name": "log-destination-type", "Values": ["cloud-watch-logs"]}],392    )["FlowLogs"]393    fl_by_cloud_watch.should.have.length_of(2)394    flow_logs_ids = tuple(map(lambda fl: fl["FlowLogId"], fl_by_cloud_watch))395    fl1.should.be.within(flow_logs_ids)396    fl3.should.be.within(flow_logs_ids)397    flow_logs_resource_ids = tuple(map(lambda fl: fl["ResourceId"], fl_by_cloud_watch))398    subnet1["SubnetId"].should.be.within(flow_logs_resource_ids)399    vpc3["VpcId"].should.be.within(flow_logs_resource_ids)400    test_fl3 = next(fl for fl in fl_by_cloud_watch if fl["FlowLogId"] == fl3)401    test_fl3["DeliverLogsStatus"].should.equal("FAILED")402    test_fl3["DeliverLogsErrorMessage"].should.equal("Access error")403    fl_by_both = client.describe_flow_logs(404        Filters=[405            {"Name": "log-destination-type", "Values": ["cloud-watch-logs", "s3"]}406        ],407    )["FlowLogs"]408    fl_by_both.should.have.length_of(3)409    fl_by_flow_log_ids = client.describe_flow_logs(410        Filters=[{"Name": "flow-log-id", "Values": [fl1, fl3]}],411    )["FlowLogs"]412    fl_by_flow_log_ids.should.have.length_of(2)413    flow_logs_ids = tuple(map(lambda fl: fl["FlowLogId"], fl_by_flow_log_ids))414    fl1.should.be.within(flow_logs_ids)415    fl3.should.be.within(flow_logs_ids)416    flow_logs_resource_ids = tuple(map(lambda fl: fl["ResourceId"], fl_by_flow_log_ids))417    subnet1["SubnetId"].should.be.within(flow_logs_resource_ids)418    vpc3["VpcId"].should.be.within(flow_logs_resource_ids)419    fl_by_group_name = client.describe_flow_logs(420        Filters=[{"Name": "log-group-name", "Values": ["test-group"]}],421    )["FlowLogs"]422    fl_by_group_name.should.have.length_of(1)423    fl_by_group_name[0]["FlowLogId"].should.equal(fl1)424    fl_by_group_name[0]["ResourceId"].should.equal(subnet1["SubnetId"])425    fl_by_group_name = client.describe_flow_logs(426        Filters=[{"Name": "log-group-name", "Values": ["non-existing-group"]}],427    )["FlowLogs"]428    fl_by_group_name.should.have.length_of(1)429    fl_by_group_name[0]["FlowLogId"].should.equal(fl3)430    fl_by_group_name[0]["ResourceId"].should.equal(vpc3["VpcId"])431    fl_by_resource_id = client.describe_flow_logs(432        Filters=[{"Name": "resource-id", "Values": [vpc2["VpcId"]]}],433    )["FlowLogs"]434    fl_by_resource_id.should.have.length_of(1)435    fl_by_resource_id[0]["FlowLogId"].should.equal(fl2)436    fl_by_resource_id[0]["ResourceId"].should.equal(vpc2["VpcId"])437    fl_by_traffic_type = client.describe_flow_logs(438        Filters=[{"Name": "traffic-type", "Values": ["ALL"]}],439    )["FlowLogs"]440    fl_by_traffic_type.should.have.length_of(1)441    fl_by_traffic_type[0]["FlowLogId"].should.equal(fl1)442    fl_by_traffic_type[0]["ResourceId"].should.equal(subnet1["SubnetId"])443    fl_by_traffic_type = client.describe_flow_logs(444        Filters=[{"Name": "traffic-type", "Values": ["Reject"]}],445    )["FlowLogs"]446    fl_by_traffic_type.should.have.length_of(1)447    fl_by_traffic_type[0]["FlowLogId"].should.equal(fl3)448    fl_by_traffic_type[0]["ResourceId"].should.equal(vpc3["VpcId"])449    fl_by_traffic_type = client.describe_flow_logs(450        Filters=[{"Name": "traffic-type", "Values": ["Accept"]}],451    )["FlowLogs"]452    fl_by_traffic_type.should.have.length_of(1)453    fl_by_traffic_type[0]["FlowLogId"].should.equal(fl2)454    fl_by_traffic_type[0]["ResourceId"].should.equal(vpc2["VpcId"])455    fl_by_tag_key = client.describe_flow_logs(456        Filters=[{"Name": "tag-key", "Values": ["foo"]}],457    )["FlowLogs"]458    fl_by_tag_key.should.have.length_of(1)459    fl_by_tag_key[0]["FlowLogId"].should.equal(fl2)460    fl_by_tag_key[0]["ResourceId"].should.equal(vpc2["VpcId"])461    fl_by_tag_key = client.describe_flow_logs(462        Filters=[{"Name": "tag-key", "Values": ["non-existing"]}],463    )["FlowLogs"]464    fl_by_tag_key.should.have.length_of(0)465    if not settings.TEST_SERVER_MODE:466        client.describe_flow_logs.when.called_with(467            Filters=[{"Name": "not-implemented-filter", "Values": ["foobar"]}],468        ).should.throw(FilterNotImplementedError)469    else:470        client.describe_flow_logs.when.called_with(471            Filters=[{"Name": "not-implemented-filter", "Values": ["foobar"]}],472        ).should.throw(ResponseParserError)473@mock_s3474@mock_ec2475def test_flow_logs_by_ids():476    s3 = boto3.resource("s3", region_name="us-west-1")477    client = boto3.client("ec2", region_name="us-west-1")478    vpc1 = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]479    vpc2 = client.create_vpc(CidrBlock="10.1.0.0/16")["Vpc"]480    vpc3 = client.create_vpc(CidrBlock="10.2.0.0/16")["Vpc"]481    fl1 = client.create_flow_logs(482        ResourceType="VPC",483        ResourceIds=[vpc1["VpcId"]],484        TrafficType="Reject",485        LogGroupName="test-group-1",486        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role-1",487    )["FlowLogIds"][0]488    fl2 = client.create_flow_logs(489        ResourceType="VPC",490        ResourceIds=[vpc2["VpcId"]],491        TrafficType="Reject",492        LogGroupName="test-group-3",493        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role-3",494    )["FlowLogIds"][0]495    fl3 = client.create_flow_logs(496        ResourceType="VPC",497        ResourceIds=[vpc3["VpcId"]],498        TrafficType="Reject",499        LogGroupName="test-group-3",500        DeliverLogsPermissionArn="arn:aws:iam::" + ACCOUNT_ID + ":role/test-role-3",501    )["FlowLogIds"][0]502    flow_logs = client.describe_flow_logs(FlowLogIds=[fl1, fl3])["FlowLogs"]503    flow_logs.should.have.length_of(2)504    flow_logs_ids = tuple(map(lambda fl: fl["FlowLogId"], flow_logs))505    fl1.should.be.within(flow_logs_ids)506    fl3.should.be.within(flow_logs_ids)507    flow_logs_resource_ids = tuple(map(lambda fl: fl["ResourceId"], flow_logs))508    vpc1["VpcId"].should.be.within(flow_logs_resource_ids)509    vpc3["VpcId"].should.be.within(flow_logs_resource_ids)...

app.py

Source:app.py

...23            raise error24        flow_logs = flow_logs_res.get("FlowLogs", [])25        if len(flow_logs) == 0:26            try:27                response = ec2_client.create_flow_logs(28                    ResourceIds=[vpc.get("VpcId")],29                    ResourceType="VPC",30                    TrafficType="ALL",31                    LogDestinationType="s3",32                    LogDestination=destination,33                )34            except ClientError as error:35                raise error36            if response.get("Unsuccessful", []) != []:37                raise Exception(f"failed to create_flow_logs {response}")38            print(f"Created flow logs for {vpc.get('VpcId')}")39        elif len(flow_logs) >= 1:40            exists = False41            for flow_log in flow_logs:42                if flow_log.get("LogDestination", "") == destination:43                    exists = True44            if not exists:45                try:46                    ec2_client.create_flow_logs(47                        ResourceIds=[vpc.get("VpcId")],48                        ResourceType="VPC",49                        TrafficType="ALL",50                        LogDestinationType="s3",51                        LogDestination=destination,52                    )53                except ClientError as error:54                    raise error55                print(f"Created flow logs for {vpc.get('VpcId')}")...

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.