def __init__(self, container, network, host, vip=None, ipv6=False):
     self._network = network
     self._container = container
     if network in [marathon.Network.HOST, marathon.Network.BRIDGE]:
         # both of these cases will rely on marathon to assign ports
         self.app, self.uuid = test_helpers.marathon_test_app(
             network=network,
             host_constraint=host,
             vip=vip,
             container_type=container,
             healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP)
     elif network == marathon.Network.USER:
         self.app, self.uuid = test_helpers.marathon_test_app(
             network=marathon.Network.USER,
             network_name='dcos6' if ipv6 else 'dcos',
             host_port=unused_port(),
             host_constraint=host,
             vip=vip,
             container_type=container,
             healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP)
         if vip is not None and container == marathon.Container.DOCKER:
             del self.app['container']['docker']['portMappings'][0]['hostPort']
     # allow this app to run on public slaves
     self.app['acceptedResourceRoles'] = ['*', 'slave_public']
     self.id = self.app['id']

def test_service_discovery_docker_overlay(dcos_api_session):
    app_definition, test_uuid = test_helpers.marathon_test_app(
        container_type=marathon.Container.DOCKER,
        network=marathon.Network.USER,
        host_port=9080)
    del app_definition['container']['docker']['portMappings'][0]['hostPort']
    assert_service_discovery(dcos_api_session, app_definition, [DNSOverlay])

def octarine_runner(dcos_api_session, mode, uuid, uri, bind_port=None):
    log.info("Running octarine(mode={}, uuid={}, uri={}".format(mode, uuid, uri))

    octarine = "/opt/mesosphere/bin/octarine"

    bind_port_str = ""
    if bind_port is not None:
        bind_port_str = "-bindPort {}".format(bind_port)

    server_cmd = "{} -mode {} {} {}".format(octarine, mode, bind_port_str, uuid)
    log.info("Server: {}".format(server_cmd))

    proxy = ('http://127.0.0.1:$({} --client --port {})'.format(octarine, uuid))
    curl_cmd = '''"$(curl --fail --proxy {} {})"'''.format(proxy, uri)
    expected_output = '''"$(printf "{\\n    \\"pong\\": true\\n}")"'''
    check_cmd = """sh -c '[ {} = {} ]'""".format(curl_cmd, expected_output)
    log.info("Check: {}".format(check_cmd))

    app, uuid = test_helpers.marathon_test_app()
    app['requirePorts'] = True
    app['cmd'] = server_cmd
    app['healthChecks'] = [{
        "protocol": "COMMAND",
        "command": {"value": check_cmd},
        'gracePeriodSeconds': 5,
        'intervalSeconds': 10,
        'timeoutSeconds': 10,
        'maxConsecutiveFailures': 30
    }]

    with dcos_api_session.marathon.deploy_and_cleanup(app):
        pass

def test_service_discovery_docker_overlay_port_mapping(dcos_api_session):
    app_definition, test_uuid = test_helpers.marathon_test_app(
        container_type=marathon.Container.DOCKER,
        healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP,
        network=marathon.Network.USER,
        host_port=9080)
    assert_service_discovery(dcos_api_session, app_definition, [DNSOverlay, DNSPortMap])

def test_service_discovery_mesos_overlay(dcos_api_session):
    app_definition, test_uuid = test_helpers.marathon_test_app(
        container_type=marathon.Container.MESOS,
        healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP,
        network=marathon.Network.USER)

    assert_service_discovery(dcos_api_session, app_definition, [DNSOverlay])

def test_service_discovery_docker_bridge(dcos_api_session):
    app_definition, test_uuid = test_helpers.marathon_test_app(
        container_type=marathon.Container.DOCKER,
        network=marathon.Network.BRIDGE,
        container_port=2020,
        host_port=9080)
    assert_service_discovery(dcos_api_session, app_definition, [DNSPortMap])

def vip_app(container: marathon.Container, network: marathon.Network, host: str, vip: str):
    # user_net_port is only actually used for USER network because this cannot be assigned
    # by marathon
    if network in [marathon.Network.HOST, marathon.Network.BRIDGE]:
        # both of these cases will rely on marathon to assign ports
        return test_helpers.marathon_test_app(
            network=network,
            host_constraint=host,
            vip=vip,
            container_type=container)
    elif network == marathon.Network.USER:
        return test_helpers.marathon_test_app(
            network=network,
            host_port=unused_port(marathon.Network.USER),
            host_constraint=host,
            vip=vip,
            container_type=container)
    else:
        raise AssertionError('Unexpected network: {}'.format(network.value))

def test_dcos_diagnostics_bundle_create_download_delete(dcos_api_session):
    """
    test bundle create, read, delete workflow
    """
    app, test_uuid = test_helpers.marathon_test_app()
    with dcos_api_session.marathon.deploy_and_cleanup(app):
        bundle = _create_bundle(dcos_api_session)
        _check_diagnostics_bundle_status(dcos_api_session)
        _download_and_extract_bundle(dcos_api_session, bundle)
        _download_and_extract_bundle_from_another_master(dcos_api_session, bundle)
        _delete_bundle(dcos_api_session, bundle)

def test_if_ucr_app_can_be_deployed(dcos_api_session, healthcheck):
    """Marathon app inside ucr deployment integration test.

    Verifies that a marathon docker app inside of a ucr container can be
    deployed and accessed as expected.
    """
    deploy_test_app_and_check(
        dcos_api_session,
        *test_helpers.marathon_test_app(
            container_type=marathon.Container.MESOS,
            healthcheck_protocol=healthcheck))

def test_files_api(dcos_api_session):
    app, test_uuid = test_helpers.marathon_test_app()

    with dcos_api_session.marathon.deploy_and_cleanup(app):
        marathon_framework_id = dcos_api_session.marathon.get('/v2/info').json()['frameworkId']
        app_task = dcos_api_session.marathon.get('/v2/apps/{}/tasks'.format(app['id'])).json()['tasks'][0]

        for required_sandbox_file in ('stdout', 'stderr'):
            content = dcos_api_session.mesos_sandbox_file(
                app_task['slaveId'], marathon_framework_id, app_task['id'], required_sandbox_file)

            assert content, 'File {} should not be empty'.format(required_sandbox_file)

def test_if_docker_app_can_be_deployed(dcos_api_session):
    """Marathon app inside docker deployment integration test.

    Verifies that a marathon app inside of a docker daemon container can be
    deployed and accessed as expected.
    """
    deploy_test_app_and_check(
        dcos_api_session,
        *test_helpers.marathon_test_app(
            network=marathon.Network.BRIDGE,
            container_type=marathon.Container.DOCKER,
            container_port=9080))

def test_l4lb(dcos_api_session):
    '''Test l4lb is load balancing between all the backends
       * create 5 apps using the same VIP
       * get uuid from the VIP in parallel from many threads
       * verify that 5 uuids have been returned
       * only testing if all 5 are hit at least once
    '''
    if not lb_enabled():
        pytest.skip('Load Balancer disabled')
    numapps = 5
    numthreads = numapps * 4
    apps = []
    rvs = deque()
    backends = []
    dnsname = 'l4lbtest.marathon.l4lb.thisdcos.directory:5000'
    with contextlib.ExitStack() as stack:
        for _ in range(numapps):
            origin_app, origin_uuid = \
                test_helpers.marathon_test_app(
                    healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP)
            # same vip for all the apps
            origin_app['portDefinitions'][0]['labels'] = {'VIP_0': '/l4lbtest:5000'}
            apps.append(origin_app)
            stack.enter_context(dcos_api_session.marathon.deploy_and_cleanup(origin_app))
            sp = dcos_api_session.marathon.get_app_service_endpoints(origin_app['id'])
            backends.append({'port': sp[0].port, 'ip': sp[0].host})
            # make sure that the service point responds
            geturl('http://{}:{}/ping'.format(sp[0].host, sp[0].port))
            # make sure that the VIP is responding too
            geturl('http://{}/ping'.format(dnsname))
        vips = geturl("http://localhost:62080/v1/vips")
        [vip] = [vip for vip in vips if vip['vip'] == dnsname and vip['protocol'] == 'tcp']
        for backend in vip['backend']:
            backends.remove(backend)
        assert backends == []

        # do many requests in parallel.
        def thread_request():
            # deque is thread safe
            rvs.append(geturl('http://l4lbtest.marathon.l4lb.thisdcos.directory:5000/test_uuid'))

        threads = [threading.Thread(target=thread_request) for i in range(0, numthreads)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

    expected_uuids = [a['id'].split('-')[2] for a in apps]
    received_uuids = [r['test_uuid'] for r in rvs if r is not None]
    assert len(set(expected_uuids)) == numapps
    assert len(set(received_uuids)) == numapps
    assert set(expected_uuids) == set(received_uuids)

 def __init__(self, container, network, host, vip=None):
     self._network = network
     self._container = container
     if network in [marathon.Network.HOST, marathon.Network.BRIDGE]:
         # both of these cases will rely on marathon to assign ports
         self.app, self.uuid = test_helpers.marathon_test_app(
             network=network,
             host_constraint=host,
             vip=vip,
             container_type=container,
             healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP)
     elif network == marathon.Network.USER:
         self.app, self.uuid = test_helpers.marathon_test_app(
             network=network,
             host_port=unused_port(),
             host_constraint=host,
             vip=vip,
             container_type=container,
             healthcheck_protocol=marathon.Healthcheck.MESOS_HTTP)
     # allow this app to run on public slaves
     self.app['acceptedResourceRoles'] = ['*', 'slave_public']
     self.id = self.app['id']

def test_if_marathon_app_can_be_deployed(dcos_api_session):
    """Marathon app deployment integration test

    This test verifies that marathon app can be deployed, and that service points
    returned by Marathon indeed point to the app that was deployed.

    The application being deployed is a simple http server written in python.
    Please test_server.py for more details.

    This is done by assigning an unique UUID to each app and passing it to the
    docker container as an env variable. After successful deployment, the
    "GET /test_uuid" request is issued to the app. If the returned UUID matches
    the one assigned to test - test succeeds.
    """
    deploy_test_app_and_check(dcos_api_session, *test_helpers.marathon_test_app())

def test_if_marathon_app_can_be_deployed_with_mesos_containerizer(dcos_api_session):
    """Marathon app deployment integration test using the Mesos Containerizer

    This test verifies that a Marathon app using the Mesos containerizer with
    a Docker image can be deployed.

    This is done by assigning an unique UUID to each app and passing it to the
    docker container as an env variable. After successfull deployment, the
    "GET /test_uuid" request is issued to the app. If the returned UUID matches
    the one assigned to test - test succeds.

    When port mapping is available (MESOS-4777), this test should be updated to
    reflect that.
    """
    deploy_test_app_and_check(
        dcos_api_session,
        *test_helpers.marathon_test_app(container_type=marathon.Container.MESOS))

def test_octarine(dcos_api_session, timeout=30):
    expanded_config = test_helpers.get_expanded_config()
    if expanded_config.get('security') == 'strict':
        pytest.skip('See: https://jira.mesosphere.com/browse/DCOS-14760')
    # This app binds to port 80. This is only required by the http (not srv)
    # transparent mode test. In transparent mode, we use ".mydcos.directory"
    # to go to localhost, the port attached there is only used to
    # determine which port to send traffic to on localhost. When it
    # reaches the proxy, the port is not used, and a request is made
    # to port 80.
    app, uuid = test_helpers.marathon_test_app(host_port=80)
    app['acceptedResourceRoles'] = ["slave_public"]
    app['requirePorts'] = True

    with dcos_api_session.marathon.deploy_and_cleanup(app):
        service_points = dcos_api_session.marathon.get_app_service_endpoints(app['id'])
        port_number = service_points[0].port
        # It didn't actually grab port 80 when requirePorts was unset
        assert port_number == app['portDefinitions'][0]["port"]

        app_name = app["id"].strip("/")
        port_name = app['portDefinitions'][0]["name"]
        port_protocol = app['portDefinitions'][0]["protocol"]

        srv = "_{}._{}._{}.marathon.mesos".format(port_name, app_name, port_protocol)
        addr = "{}.marathon.mesos".format(app_name)
        transparent_suffix = ".mydcos.directory"

        standard_mode = "standard"
        transparent_mode = "transparent"

        t_addr_bind = 2508
        t_srv_bind = 2509

        standard_addr = "{}:{}/ping".format(addr, port_number)
        standard_srv = "{}/ping".format(srv)
        transparent_addr = "{}{}:{}/ping".format(addr, transparent_suffix, t_addr_bind)
        transparent_srv = "{}{}:{}/ping".format(srv, transparent_suffix, t_srv_bind)

        # The uuids are different between runs so that they don't have a
        # chance of colliding. They shouldn't anyways, but just to be safe.
        octarine_runner(dcos_api_session, standard_mode, uuid + "1", standard_addr)
        octarine_runner(dcos_api_session, standard_mode, uuid + "2", standard_srv)
        octarine_runner(dcos_api_session, transparent_mode, uuid + "3", transparent_addr, bind_port=t_addr_bind)
        octarine_runner(dcos_api_session, transparent_mode, uuid + "4", transparent_srv, bind_port=t_srv_bind)

def test_ip_per_container(dcos_api_session):
    '''Test if we are able to connect to a task with ip-per-container mode
    '''
    # Launch the test_server in ip-per-container mode (user network)
    app_definition, test_uuid = test_helpers.marathon_test_app(
        container_type=marathon.Container.DOCKER,
        network=marathon.Network.USER,
        host_port=9080)

    assert len(dcos_api_session.slaves) >= 2, 'IP Per Container tests require 2 private agents to work'

    app_definition['instances'] = 2
    app_definition['constraints'] = [['hostname', 'UNIQUE']]

    with dcos_api_session.marathon.deploy_and_cleanup(app_definition, check_health=True):
        service_points = dcos_api_session.marathon.get_app_service_endpoints(app_definition['id'])
        app_port = app_definition['container']['docker']['portMappings'][0]['containerPort']
        cmd = '/opt/mesosphere/bin/curl -s -f -m 5 http://{}:{}/ping'.format(service_points[1].ip, app_port)
        ensure_routable(cmd, service_points[0].host, service_points[0].port)

def test_files_api(dcos_api_session):
    '''
    This test verifies that the standard output and error of a Mesos task can be
    read. We check that neither standard output nor error are empty files. Since
    the default `marathon_test_app()` does not write to its standard output the
    task definition is modified to output something there.
    '''
    app, test_uuid = test_helpers.marathon_test_app()
    app['cmd'] = 'echo $DCOS_TEST_UUID && ' + app['cmd']

    with dcos_api_session.marathon.deploy_and_cleanup(app):
        marathon_framework_id = dcos_api_session.marathon.get('/v2/info').json()['frameworkId']
        app_task = dcos_api_session.marathon.get('/v2/apps/{}/tasks'.format(app['id'])).json()['tasks'][0]

        for required_sandbox_file in ('stdout', 'stderr'):
            content = dcos_api_session.mesos_sandbox_file(
                app_task['slaveId'], marathon_framework_id, app_task['id'], required_sandbox_file)

            assert content, 'File {} should not be empty'.format(required_sandbox_file)

 def __init__(self, container, network, host, vip=None, ipv6=False, app_name_fmt=None):
     args = {
         'app_name_fmt': app_name_fmt,
         'network': network,
         'host_port': unused_port(),
         'host_constraint': host,
         'vip': vip,
         'container_type': container,
         'healthcheck_protocol': marathon.Healthcheck.MESOS_HTTP
     }
     if network == marathon.Network.USER:
         args['container_port'] = unused_port()
         if ipv6:
             args['network_name'] = 'dcos6'
         if vip is not None:
             del args['host_port']
     self.app, self.uuid = test_helpers.marathon_test_app(**args)
     # allow this app to run on public slaves
     self.app['acceptedResourceRoles'] = ['*', 'slave_public']
     self.id = self.app['id']

def test_if_search_is_working(dcos_api_session):
    """Test if custom set search is working.

    Verifies that a marathon app running on the dcos_api_session can resolve names using
    searching the "search" the dcos_api_session was launched with (if any). It also tests
    that absolute searches still work, and search + things that aren't
    sub-domains fails properly.

    The application being deployed is a simple http server written in python.
    Please check test_server.py for more details.
    """
    # Launch the app
    app_definition, test_uuid = test_helpers.marathon_test_app()
    with dcos_api_session.marathon.deploy_and_cleanup(app_definition):
        service_points = dcos_api_session.marathon.get_app_service_endpoints(app_definition['id'])
        # Get the status
        r = requests.get('http://{}:{}/dns_search'.format(service_points[0].host,
                                                          service_points[0].port))
        if r.status_code != 200:
            msg = "Test server replied with non-200 reply: '{0} {1}. "
            msg += "Detailed explanation of the problem: {2}"
            pytest.fail(msg.format(r.status_code, r.reason, r.text))

        r_data = r.json()

        # Make sure we hit the app we expected
        assert r_data['test_uuid'] == test_uuid

        expected_error = {'error': '[Errno -2] Name or service not known'}

        # Check that result matches expectations for this dcos_api_session
        expanded_config = test_helpers.get_expanded_config()
        if expanded_config['dns_search']:
            assert r_data['search_hit_leader'] in dcos_api_session.masters
            assert r_data['always_hit_leader'] in dcos_api_session.masters
            assert r_data['always_miss'] == expected_error
        else:  # No dns search, search hit should miss.
            assert r_data['search_hit_leader'] == expected_error
            assert r_data['always_hit_leader'] in dcos_api_session.masters
            assert r_data['always_miss'] == expected_error