<commit_before>from common_fixtures import *  # NOQA
from cattle import ApiError


def _create_stack(client):
    env = client.create_environment(name=random_str())
    env = client.wait_success(env)
    assert env.state == "active"
    return env


def test_create_duplicated_services(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    service_name = random_str()
    service1 = client.create_service(name=service_name,
                                     environmentId=env.id,
                                     launchConfig=launch_config)
    client.wait_success(service1)

    with pytest.raises(ApiError) as e:
        client.create_service(name=service_name,
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    with pytest.raises(ApiError) as e:
        client.create_externalService(name=service_name,
                                      environmentId=env.id,
                                      launchConfig=launch_config,
                                      externalIpAddresses=["172.16.31.10"])
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    with pytest.raises(ApiError) as e:
        client.create_dnsService(name=service_name,
                                 environmentId=env.id)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    # try to update the service with duplicated service name
    service = client.create_service(name=random_str(),
                                    environmentId=env.id,
                                    launchConfig=launch_config)
    service = client.wait_success(service)
    with pytest.raises(ApiError) as e:
        client.update(service, name=service_name)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    # remove the service and try to re-use its name
    client.wait_success(service1.remove())
    client.create_service(name=service_name,
                          environmentId=env.id,
                          launchConfig=launch_config)


def test_external_service_w_hostname(client, context):
    env = _create_stack(client)
    # try to create external service with both hostname externalips
    with pytest.raises(ApiError) as e:
        ips = ["172.16.31.10", '192.168.0.10']
        client.create_externalService(name=random_str(),
                                      environmentId=env.id,
                                      hostname="a.com",
                                      externalIpAddresses=ips)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_circular_refs(client, context):
    env = _create_stack(client)

    # test direct circular ref
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid,
                     "dataVolumesFromLaunchConfigs": ['secondary']}

    secondary_lc = {"imageUuid": image_uuid, "name": "secondary",
                    "dataVolumesFromLaunchConfigs": ['primary']}

    with pytest.raises(ApiError) as e:
        client.create_service(name="primary",
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'

    # test indirect circular ref
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid,
                     "dataVolumesFromLaunchConfigs": ['secondary1']}

    s_lc1 = {"imageUuid": image_uuid, "name": "secondary1",
             "dataVolumesFromLaunchConfigs": ['secondary2']}

    s_lc2 = {"imageUuid": image_uuid, "name": "secondary2",
             "dataVolumesFromLaunchConfigs": ['primary']}

    with pytest.raises(ApiError) as e:
        client.create_service(name="primary",
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[s_lc1, s_lc2])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'


def test_validate_image(client, context):
    env = _create_stack(client)

    # 1. invalide image in primary config
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": "ubuntu:14:04"}

    secondary_lc = {"imageUuid": image_uuid, "name": "secondary"}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'
    assert e.value.error.fieldName == 'imageUuid'

    # 2. invalide image in secondary config
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    secondary_lc = {"imageUuid": "ubuntu:14:04", "name": "secondary"}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'
    assert e.value.error.fieldName == 'imageUuid'


def test_validate_port(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid

    # test invalid format
    launch_config = {"imageUuid": image_uuid, "ports": ["4565tcp"]}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'PortWrongFormat'

    launch_config = {"imageUuid": image_uuid, "ports": ["4565/invalidtcp"]}
    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'PortInvalidProtocol'

    # test reserved port
    launch_config = {"imageUuid": image_uuid, "ports": ["42:43"]}
    with pytest.raises(ApiError) as e:
        client.create_loadBalancerService(name=random_str(),
                                          environmentId=env.id,
                                          launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_vip_requested_ip(client, context):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    # vip out of the range
    with pytest.raises(ApiError) as e:
        vip = "192.168.127.12"
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              vip=vip)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_add_svc_to_removed_stack(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    client.create_service(name=random_str(),
                          environmentId=env.id,
                          launchConfig=launch_config)

    client.create_service(name=random_str(),
                          environmentId=env.id,
                          launchConfig=launch_config)

    env.remove()

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidState'
    assert e.value.error.fieldName == 'environment'

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidState'
    assert e.value.error.fieldName == 'environment'


def test_validate_launch_config_name(client, context):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    svc_name = random_str()
    service = client.create_service(name=svc_name,
                                    environmentId=env.id,
                                    launchConfig=launch_config)

    client.wait_success(service)

    launch_config = {"imageUuid": image_uuid}

    secondary_lc = {"imageUuid": image_uuid,
                    "name": svc_name}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'


def test_validate_service_token(client, context, super_client):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    svc_name = random_str()
    service = client.create_service(name=svc_name,
                                    environmentId=env.id,
                                    launchConfig=launch_config)

    client.wait_success(service)

    service = super_client.reload(service)

    assert service.state == "inactive"
    assert service.data.fields.token is not None

    token = service.data.fields.token

    svc_name = random_str()

    client.update(service, name=svc_name)

    client.wait_success(service)

    service = super_client.reload(service)

    assert service.name == svc_name
    assert service.data.fields.token is not None
    assert service.data.fields.token == token


def test_ip_retain(client, context, super_client):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    svc = client.create_service(name=random_str(),
                                environmentId=env.id,
                                launchConfig=launch_config,
                                scale=1,
                                retainIp=True)
    svc = client.wait_success(svc)
    assert svc.state == "inactive"

    # validate that startFirst can't be used on a service with retainIp = true
    strategy = {"launchConfig": launch_config,
                "intervalMillis": 100,
                "startFirst": True}
    with pytest.raises(ApiError) as e:
        svc.upgrade_action(inServiceStrategy=strategy)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'
<commit_msg>Flake8 error fix<commit_after>from common_fixtures import *  # NOQA
from cattle import ApiError


def _create_stack(client):
    env = client.create_environment(name=random_str())
    env = client.wait_success(env)
    assert env.state == "active"
    return env


def test_create_duplicated_services(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    service_name = random_str()
    service1 = client.create_service(name=service_name,
                                     environmentId=env.id,
                                     launchConfig=launch_config)
    client.wait_success(service1)

    with pytest.raises(ApiError) as e:
        client.create_service(name=service_name,
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    with pytest.raises(ApiError) as e:
        client.create_externalService(name=service_name,
                                      environmentId=env.id,
                                      launchConfig=launch_config,
                                      externalIpAddresses=["172.16.31.10"])
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    with pytest.raises(ApiError) as e:
        client.create_dnsService(name=service_name,
                                 environmentId=env.id)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    # try to update the service with duplicated service name
    service = client.create_service(name=random_str(),
                                    environmentId=env.id,
                                    launchConfig=launch_config)
    service = client.wait_success(service)
    with pytest.raises(ApiError) as e:
        client.update(service, name=service_name)
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'
    assert e.value.error.fieldName == 'name'

    # remove the service and try to re-use its name
    client.wait_success(service1.remove())
    client.create_service(name=service_name,
                          environmentId=env.id,
                          launchConfig=launch_config)


def test_external_service_w_hostname(client, context):
    env = _create_stack(client)
    # try to create external service with both hostname externalips
    with pytest.raises(ApiError) as e:
        ips = ["172.16.31.10", '192.168.0.10']
        client.create_externalService(name=random_str(),
                                      environmentId=env.id,
                                      hostname="a.com",
                                      externalIpAddresses=ips)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_circular_refs(client, context):
    env = _create_stack(client)

    # test direct circular ref
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid,
                     "dataVolumesFromLaunchConfigs": ['secondary']}

    secondary_lc = {"imageUuid": image_uuid, "name": "secondary",
                    "dataVolumesFromLaunchConfigs": ['primary']}

    with pytest.raises(ApiError) as e:
        client.create_service(name="primary",
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'

    # test indirect circular ref
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid,
                     "dataVolumesFromLaunchConfigs": ['secondary1']}

    s_lc1 = {"imageUuid": image_uuid, "name": "secondary1",
             "dataVolumesFromLaunchConfigs": ['secondary2']}

    s_lc2 = {"imageUuid": image_uuid, "name": "secondary2",
             "dataVolumesFromLaunchConfigs": ['primary']}

    with pytest.raises(ApiError) as e:
        client.create_service(name="primary",
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[s_lc1, s_lc2])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'


def test_validate_image(client, context):
    env = _create_stack(client)

    # 1. invalide image in primary config
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": "ubuntu:14:04"}

    secondary_lc = {"imageUuid": image_uuid, "name": "secondary"}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'
    assert e.value.error.fieldName == 'imageUuid'

    # 2. invalide image in secondary config
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    secondary_lc = {"imageUuid": "ubuntu:14:04", "name": "secondary"}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidReference'
    assert e.value.error.fieldName == 'imageUuid'


def test_validate_port(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid

    # test invalid format
    launch_config = {"imageUuid": image_uuid, "ports": ["4565tcp"]}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'PortWrongFormat'

    launch_config = {"imageUuid": image_uuid, "ports": ["4565/invalidtcp"]}
    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'PortInvalidProtocol'

    # test reserved port
    launch_config = {"imageUuid": image_uuid, "ports": ["42:43"]}
    with pytest.raises(ApiError) as e:
        client.create_loadBalancerService(name=random_str(),
                                          environmentId=env.id,
                                          launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_vip_requested_ip(client, context):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    # vip out of the range
    with pytest.raises(ApiError) as e:
        vip = "192.168.127.12"
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              vip=vip)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'


def test_add_svc_to_removed_stack(client, context):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    client.create_service(name=random_str(),
                          environmentId=env.id,
                          launchConfig=launch_config)

    client.create_service(name=random_str(),
                          environmentId=env.id,
                          launchConfig=launch_config)

    env.remove()

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidState'
    assert e.value.error.fieldName == 'environment'

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidState'
    assert e.value.error.fieldName == 'environment'


def test_validate_launch_config_name(client, context):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    svc_name = random_str()
    service = client.create_service(name=svc_name,
                                    environmentId=env.id,
                                    launchConfig=launch_config)

    client.wait_success(service)

    launch_config = {"imageUuid": image_uuid}

    secondary_lc = {"imageUuid": image_uuid,
                    "name": svc_name}

    with pytest.raises(ApiError) as e:
        client.create_service(name=random_str(),
                              environmentId=env.id,
                              launchConfig=launch_config,
                              secondaryLaunchConfigs=[secondary_lc])
    assert e.value.error.status == 422
    assert e.value.error.code == 'NotUnique'


def test_validate_service_token(client, context, super_client):
    env = _create_stack(client)
    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}
    svc_name = random_str()
    service = client.create_service(name=svc_name,
                                    environmentId=env.id,
                                    launchConfig=launch_config)

    client.wait_success(service)

    service = super_client.reload(service)

    assert service.state == "inactive"
    assert service.data.fields.token is not None

    token = service.data.fields.token

    svc_name = random_str()

    client.update(service, name=svc_name)

    client.wait_success(service)

    service = super_client.reload(service)

    assert service.name == svc_name
    assert service.data.fields.token is not None
    assert service.data.fields.token == token


def test_ip_retain(client, context, super_client):
    env = _create_stack(client)

    image_uuid = context.image_uuid
    launch_config = {"imageUuid": image_uuid}

    svc = client.create_service(name=random_str(),
                                environmentId=env.id,
                                launchConfig=launch_config,
                                scale=1,
                                retainIp=True)
    svc = client.wait_success(svc)
    assert svc.state == "inactive"

    # validate that startFirst can't be used on a service with retainIp = true
    strategy = {"launchConfig": launch_config,
                "intervalMillis": 100,
                "startFirst": True}
    with pytest.raises(ApiError) as e:
        svc.upgrade_action(inServiceStrategy=strategy)
    assert e.value.error.status == 422
    assert e.value.error.code == 'InvalidOption'
{'fraction_non_alphanumeric': 0.09589519650655022, 'fraction_numerical': 0.011048034934497816, 'mean_word_length': 2.122579765475866, 'pattern_counts': {'":': 74, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 38, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6270513', 'n_tokens_mistral': 6676, 'n_tokens_neox': 6315, 'n_words': 1186}