test_integration_basics.py

"""Tests Nighthawk's basic functionality."""

import json
import logging
import math
import os
import pytest
import subprocess
import sys
import time
from threading import Thread

from test.integration.common import IpVersion
from test.integration.integration_test_fixtures import (
    http_test_server_fixture, https_test_server_fixture, https_test_server_fixture,
    multi_http_test_server_fixture, multi_https_test_server_fixture, quic_test_server_fixture,
    server_config, server_config_quic)
from test.integration import asserts
from test.integration import utility

# TODO(oschaaf): we mostly verify stats observed from the client-side. Add expectations
# for the server side as well.


@pytest.mark.skipif(utility.isSanitizerRun(), reason="Unstable and very slow in sanitizer runs")
def test_http_h1(http_test_server_fixture):
  """Test http1 over plain http.

  Runs the CLI configured to use plain HTTP/1 against our test server, and sanity
  checks statistics from both client and server.
  """
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      http_test_server_fixture.getTestServerRootUri(), "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:24"
  ])
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_rx_bytes_total", 3400)
  # It is possible that the # of upstream_cx > # of backend connections for H1
  # as new connections will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_tx_bytes_total", 500)
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "default.total_match_count", 1)

  global_histograms = http_test_server_fixture.getNighthawkGlobalHistogramsbyIdFromJson(parsed_json)
  asserts.assertEqual(int(global_histograms["benchmark_http_client.response_body_size"]["count"]),
                      25)
  asserts.assertEqual(int(global_histograms["benchmark_http_client.response_header_size"]["count"]),
                      25)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_body_size"]["raw_mean"]), 10)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_header_size"]["raw_mean"]), 97)
  asserts.assertEqual(int(global_histograms["benchmark_http_client.response_body_size"]["raw_min"]),
                      10)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_header_size"]["raw_min"]), 97)
  asserts.assertEqual(int(global_histograms["benchmark_http_client.response_body_size"]["raw_max"]),
                      10)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_header_size"]["raw_max"]), 97)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_body_size"]["raw_pstdev"]), 0)
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.response_header_size"]["raw_pstdev"]), 0)

  asserts.assertGreaterEqual(len(counters), 12)


def _mini_stress_test(fixture, args):
  # run a test with more rps then we can handle, and a very small client-side queue.
  # we should observe both lots of successfull requests as well as time spend in blocking mode.,
  parsed_json, _ = fixture.runNighthawkClient(args)
  counters = fixture.getNighthawkCounterMapFromJson(parsed_json)
  # We set a reasonably low expectation of 100 requests. We set it low, because we want this
  # test to succeed on a reasonable share of setups (hopefully practically all).
  MIN_EXPECTED_REQUESTS = 100
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", MIN_EXPECTED_REQUESTS)
  if "--h2" in args:
    asserts.assertCounterEqual(counters, "upstream_cx_http2_total", 1)
  else:
    asserts.assertCounterEqual(counters, "upstream_cx_http1_total", 1)
  global_histograms = fixture.getNighthawkGlobalHistogramsbyIdFromJson(parsed_json)

  if "--open-loop" in args:
    asserts.assertEqual(int(global_histograms["sequencer.blocking"]["count"]), 0)
  else:
    asserts.assertGreaterEqual(int(global_histograms["sequencer.blocking"]["count"]), 1)

  asserts.assertGreaterEqual(
      int(global_histograms["benchmark_http_client.request_to_response"]["count"]), 1)
  asserts.assertGreaterEqual(int(global_histograms["benchmark_http_client.latency_2xx"]["count"]),
                             1)
  return counters


# The mini stress tests below are executing in closed-loop mode. As we guard the pool against
# overflows, we can set fixed expectations with respect to overflows and anticipated pending
# totals.
def test_http_h1_mini_stress_test_with_client_side_queueing(http_test_server_fixture):
  """Run a max rps test with the h1 pool against our test server, using a small client-side queue."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "999999", "--max-pending-requests",
      "10", "--connections", "1", "--duration", "100", "--termination-predicate",
      "benchmark.http_2xx:99", "--simple-warmup"
  ])
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_total", 11)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_overflow", 10)


def test_http_h1_mini_stress_test_without_client_side_queueing(http_test_server_fixture):
  """Run a max rps test with the h1 pool against our test server, with no client-side queueing."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "999999", "--connections", "1",
      "--duration", "100", "--termination-predicate", "benchmark.http_2xx:99"
  ])
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertNotIn("upstream_cx_overflow", counters)


def test_http_h2_mini_stress_test_with_client_side_queueing(http_test_server_fixture):
  """Run a max rps test with the h2 pool against our test server, using a small client-side queue."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "999999", "--max-pending-requests",
      "10", "--h2", "--max-active-requests", "1", "--connections", "1", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:99", "--simple-warmup"
  ])
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_overflow", 10)


def test_http_h2_mini_stress_test_without_client_side_queueing(http_test_server_fixture):
  """Run a max rps test with the h2 pool against our test server, with no client-side queueing."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "999999", "--h2",
      "--max-active-requests", "1", "--connections", "1", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:99"
  ])
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertNotIn("upstream_rq_pending_overflow", counters)


@pytest.mark.skipif(not utility.isRunningInAzpCi(),
                    reason="Has very high failure rate in local executions.")
@pytest.mark.skipif(utility.isSanitizerRun(), reason="Unstable and very slow in sanitizer runs")
def test_http_h1_mini_stress_test_open_loop(http_test_server_fixture):
  """Run an H1 open loop stress test. We expect higher pending and overflow counts."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "10000", "--max-pending-requests",
      "1", "--open-loop", "--max-active-requests", "1", "--connections", "1", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:99", "--simple-warmup"
  ])
  # we expect pool overflows
  asserts.assertCounterGreater(counters, "benchmark.pool_overflow", 10)


@pytest.mark.skipif(not utility.isRunningInAzpCi(),
                    reason="Has very high failure rate in local executions.")
@pytest.mark.skipif(utility.isSanitizerRun(), reason="Unstable and very slow in sanitizer runs")
def test_http_h2_mini_stress_test_open_loop(http_test_server_fixture):
  """Run an H2 open loop stress test. We expect higher overflow counts."""
  counters = _mini_stress_test(http_test_server_fixture, [
      http_test_server_fixture.getTestServerRootUri(), "--rps", "10000", "--max-pending-requests",
      "1", "--h2", "--open-loop", "--max-active-requests", "1", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:99", "--simple-warmup"
  ])
  # we expect pool overflows
  asserts.assertCounterGreater(counters, "benchmark.pool_overflow", 10)


def test_http_h2(http_test_server_fixture):
  """Test h2 over plain http.

  Runs the CLI configured to use h2c against our test server, and sanity
  checks statistics from both client and server.
  """
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      "--h2",
      http_test_server_fixture.getTestServerRootUri(), "--max-active-requests", "1", "--duration",
      "100", "--termination-predicate", "benchmark.http_2xx:24", "--rps", "100"
  ])
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_http2_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_rx_bytes_total", 1030)
  asserts.assertCounterEqual(counters, "upstream_cx_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_tx_bytes_total", 403)
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "default.total_match_count", 1)
  asserts.assertGreaterEqual(len(counters), 12)


def test_http_concurrency(http_test_server_fixture):
  """Test that concurrency acts like a multiplier."""
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      "--concurrency 4 --rps 100 --connections 1", "--duration", "100", "--termination-predicate",
      "benchmark.http_2xx:24",
      http_test_server_fixture.getTestServerRootUri()
  ])
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)

  # Quite a loose expectation, but this may fluctuate depending on server load.
  # Ideally we'd see 4 workers * 5 rps * 5s = 100 requests total
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 100)
  # Assert that we at least have 1 connection for each event loop (1*4). It is possible that the # of
  # upstream_cx > # of backend connections for H1 as new connections will spawn if the existing clients
  # cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 4)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h1(https_test_server_fixture):
  """Test h1 over https.

  Runs the CLI configured to use HTTP/1 over https against our test server, and sanity
  checks statistics from both client and server.
  """
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--connections", "1", "--rps", "100",
      "--duration", "100", "--termination-predicate", "benchmark.http_2xx:24"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_rx_bytes_total", 3400)
  # It is possible that the # of upstream_cx > # of backend connections for H1 as new connections
  # will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_tx_bytes_total", 500)
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "ssl.ciphers.ECDHE-RSA-AES128-GCM-SHA256", 1)
  asserts.assertCounterEqual(counters, "ssl.curves.X25519", 1)
  asserts.assertCounterEqual(counters, "ssl.handshake", 1)
  asserts.assertCounterEqual(counters, "ssl.sigalgs.rsa_pss_rsae_sha256", 1)
  asserts.assertCounterEqual(counters, "ssl.versions.TLSv1.2", 1)
  asserts.assertCounterEqual(counters, "default.total_match_count", 1)
  asserts.assertEqual(len(counters), 17)

  server_stats = https_test_server_fixture.getTestServerStatisticsJson()
  asserts.assertEqual(
      https_test_server_fixture.getServerStatFromJson(server_stats,
                                                      "http.ingress_http.downstream_rq_2xx"), 25)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h2(https_test_server_fixture):
  """Test http2 over https.

  Runs the CLI configured to use HTTP/2 (using https) against our test server, and sanity
  checks statistics from both client and server.
  """
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      "--h2",
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:24", "--max-active-requests", "1"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_http2_total", 1)
  # Through emperical observation, 1030 has been determined to be the minimum of bytes
  # we can expect to have received when execution has stopped.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_rx_bytes_total", 1030)
  asserts.assertCounterEqual(counters, "upstream_cx_total", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_tx_bytes_total", 403)
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "ssl.ciphers.ECDHE-RSA-AES128-GCM-SHA256", 1)
  asserts.assertCounterEqual(counters, "ssl.curves.X25519", 1)
  asserts.assertCounterEqual(counters, "ssl.handshake", 1)
  asserts.assertCounterEqual(counters, "ssl.sigalgs.rsa_pss_rsae_sha256", 1)
  asserts.assertCounterEqual(counters, "ssl.versions.TLSv1.2", 1)
  asserts.assertCounterEqual(counters, "default.total_match_count", 1)
  asserts.assertEqual(len(counters), 17)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h2_multiple_connections(https_test_server_fixture):
  """Test that the experimental h2 pool uses multiple connections.

  The burst we send ensures we will need 10 connections right away, as we
  limit max active streams per connection to 1 by setting the experimental
  flag to use multiple h2 connections.
  """
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      "--h2",
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:99", "--max-active-requests", "10",
      "--max-pending-requests", "10", "--max-concurrent-streams", "1", "--burst-size", "10"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 100)
  # Empirical observation shows we may end up creating more then 10 connections.
  # This is stock Envoy h/2 pool behavior.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http2_total", 10)


def test_h3_quic(quic_test_server_fixture):
  """Test http3 quic.

  Runs the CLI configured to use HTTP/3 Quic against our test server, and sanity
  checks statistics from both client and server.
  """
  parsed_json, _ = quic_test_server_fixture.runNighthawkClient([
      "--protocol http3",
      quic_test_server_fixture.getTestServerRootUri(),
      "--rps",
      "100",
      "--duration",
      "100",
      "--termination-predicate",
      "benchmark.http_2xx:24",
      "--max-active-requests",
      "1",
      # Envoy doesn't support disabling certificate verification on Quic
      # connections, so the host in our requests has to match the hostname in
      # the leaf certificate.
      "--request-header",
      "Host:www.lyft.com"
  ])
  counters = quic_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_http3_total", 1)
  asserts.assertCounterEqual(counters, "upstream_cx_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_pending_total", 1)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "default.total_match_count", 1)


def test_h3_quic_with_custom_http3_protocol_options(quic_test_server_fixture):
  """Test http3 quic with custom http3 protocol options.

  Sets the maximum number of concurrent streams to one and verifies that
  Nighthawk uses multiple connections.
  """
  http3_protocol_options = "{quic_protocol_options:{max_concurrent_streams:1}}"

  parsed_json, _ = quic_test_server_fixture.runNighthawkClient([
      "--protocol http3",
      quic_test_server_fixture.getTestServerRootUri(),
      "--rps",
      "100",
      "--duration",
      "100",
      "--termination-predicate",
      "benchmark.http_2xx:99",
      "--max-active-requests",
      "10",
      "--max-pending-requests",
      "10",
      "--burst-size",
      "10",
      # Envoy doesn't support disabling certificate verification on Quic
      # connections, so the host in our requests has to match the hostname in
      # the leaf certificate.
      "--request-header",
      "Host:www.lyft.com",
      "--http3-protocol-options %s" % http3_protocol_options
  ])
  counters = quic_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 100)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http3_total", 10)


def test_h3_quic_with_custom_upstream_bind_configuration(quic_test_server_fixture):
  """Test http3 quic with a custom upstream bind configuration.

  Runs the CLI configured to use HTTP/3 Quic against our test server, and sanity
  checks statistics from both client and server. Sets custom address to bind
  upstream requests to.
  """
  address = quic_test_server_fixture.server_ip
  upstream_bind_config = f"{{source_address:{{address:\"{address}\",port_value:0}}}}"

  parsed_json, _ = quic_test_server_fixture.runNighthawkClient([
      "--protocol http3",
      quic_test_server_fixture.getTestServerRootUri(),
      "--rps",
      "100",
      "--duration",
      "100",
      "--termination-predicate",
      "benchmark.http_2xx:24",
      "--max-active-requests",
      "1",
      # Envoy doesn't support disabling certificate verification on Quic
      # connections, so the host in our requests has to match the hostname in
      # the leaf certificate.
      "--request-header",
      "Host:www.lyft.com",
      "--upstream-bind-config %s" % upstream_bind_config
  ])
  counters = quic_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterEqual(counters, "upstream_cx_http3_total", 1)


def _do_tls_configuration_test(https_test_server_fixture, cli_parameter, use_h2, ciphers):
  """Test with different ciphers.

  For a given choice of (--tls-context, --transport-socket) x (H1, H2),
  run a series of traffic tests with different ciphers.

  Args:
    https_test_server_fixture: pytest.fixture that controls a test server and client
    cli_parameter: string, --tls-context or --transport-socket
    use_h2: boolean, whether to pass --h2
    ciphers: list[string], list of ciphers to use with TLS
  """
  if cli_parameter == "--tls-context":
    json_template = "{common_tls_context:{tls_params:{cipher_suites:[\"-ALL:%s\"]}}}"
  else:
    json_template = "%s%s%s" % (
        "{name:\"envoy.transport_sockets.tls\",typed_config:{",
        "\"@type\":\"type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext\",",
        "common_tls_context:{tls_params:{cipher_suites:[\"-ALL:%s\"]}}}}")

  for cipher in ciphers:
    parsed_json, _ = https_test_server_fixture.runNighthawkClient(
        (["--protocol", "http2"] if use_h2 else []) + [
            "--duration", "10", "--termination-predicate", "benchmark.http_2xx:0", cli_parameter,
            json_template % cipher,
            https_test_server_fixture.getTestServerRootUri()
        ])
    counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
    asserts.assertCounterGreaterEqual(counters, "ssl.ciphers.%s" % cipher, 1)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h1_tls_context_configuration_rsa(https_test_server_fixture):
  """Test that specifying RSA compatible tls cipher suites works with the h1 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--tls-context",
                             use_h2=False,
                             ciphers=["ECDHE-RSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h1_transport_socket_configuration_rsa(https_test_server_fixture):
  """Test that specifying RSA compatible tls cipher suites via transport socket works with the h1 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--transport-socket",
                             use_h2=False,
                             ciphers=["ECDHE-RSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h2_tls_context_configuration_rsa(https_test_server_fixture):
  """Test that specifying RSA compatible tls cipher suites works with the h2 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--tls-context",
                             use_h2=True,
                             ciphers=["ECDHE-RSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_h2_transport_socket_configuration_rsa(https_test_server_fixture):
  """Test that specifying RSA compatible tls cipher suites via transport socket works with the h2 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--transport-socket",
                             use_h2=True,
                             ciphers=["ECDHE-RSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize(
    'server_config', ["nighthawk/test/integration/configurations/nighthawk_https_origin_dsa.yaml"])
def test_https_h1_tls_context_configuration_dsa(https_test_server_fixture):
  """Test that specifying DSA comptible tls cipher suites works with the h1 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--tls-context",
                             use_h2=False,
                             ciphers=["ECDHE-ECDSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize(
    'server_config', ["nighthawk/test/integration/configurations/nighthawk_https_origin_dsa.yaml"])
def test_https_h1_transport_socket_configuration_dsa(https_test_server_fixture):
  """Test that specifying DSA comptible tls cipher suites via transport socket works with the h1 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--transport-socket",
                             use_h2=False,
                             ciphers=["ECDHE-ECDSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize(
    'server_config', ["nighthawk/test/integration/configurations/nighthawk_https_origin_dsa.yaml"])
def test_https_h2_tls_context_configuration_dsa(https_test_server_fixture):
  """Test that specifying DSA comptible tls cipher suites works with the h2 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--tls-context",
                             use_h2=True,
                             ciphers=["ECDHE-ECDSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize(
    'server_config', ["nighthawk/test/integration/configurations/nighthawk_https_origin_dsa.yaml"])
def test_https_h2_transport_socket_configuration_dsa(https_test_server_fixture):
  """Test that specifying DSA comptible tls cipher suites via transport socket works with the h2 pool."""
  _do_tls_configuration_test(https_test_server_fixture,
                             "--transport-socket",
                             use_h2=True,
                             ciphers=["ECDHE-ECDSA-AES256-GCM-SHA384"])


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_prefetching(https_test_server_fixture):
  """Test we prefetch connections.

  We test for 1 second at 1 rps, which should
  result in 1 connection max without prefetching. However, we specify 50 connections
  and the prefetching flag, so we ought to see 50 http1 connections created.
  """
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      "--duration 1", "--rps 1", "--prefetch-connections", "--connections 50",
      https_test_server_fixture.getTestServerRootUri()
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "upstream_cx_http1_total", 50)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_https_log_verbosity(https_test_server_fixture):
  """Test that the specified log verbosity level is respected.

  This tests for a sentinel we know is only right when the level
  is set to 'trace'.
  """
  # TODO(oschaaf): this is kind of fragile. Can we improve?
  trace_level_sentinel = "nighthawk_service_zone"
  _, logs = https_test_server_fixture.runNighthawkClient(
      ["--duration 1", "--rps 1", "-v debug",
       https_test_server_fixture.getTestServerRootUri()])
  asserts.assertNotIn(trace_level_sentinel, logs)

  _, logs = https_test_server_fixture.runNighthawkClient(
      ["--duration 1", "--rps 1", "-v trace",
       https_test_server_fixture.getTestServerRootUri()])
  asserts.assertIn(trace_level_sentinel, logs)


def test_dotted_output_format(http_test_server_fixture):
  """Test that we get the dotted string output format when requested, and ensure we get latency percentiles."""
  output, _ = http_test_server_fixture.runNighthawkClient([
      "--duration 1", "--rps 10", "--output-format dotted",
      http_test_server_fixture.getTestServerRootUri()
  ],
                                                          as_json=False)
  asserts.assertIn("global.benchmark_http_client.request_to_response.permilles-500.microseconds",
                   output)


# TODO(oschaaf): add percentiles to the gold testing in the C++ output formatter
# once the fortio formatter has landed (https://github.com/envoyproxy/nighthawk/pull/168)
def test_cli_output_format(http_test_server_fixture):
  """Test that we observe latency percentiles with CLI output."""
  output, _ = http_test_server_fixture.runNighthawkClient(
      ["--duration 1", "--rps 10",
       http_test_server_fixture.getTestServerRootUri()], as_json=False)
  asserts.assertIn("Initiation to completion", output)
  asserts.assertIn("Percentile", output)


@pytest.mark.parametrize(
    'filter_configs',
    ["{}", "{static_delay: \"0.01s\"}", "{emit_previous_request_delta_in_response_header: \"aa\"}"])
def test_request_body_gets_transmitted(http_test_server_fixture, filter_configs):
  """Test request body transmission handling code for our extensions.

  Ensure that the number of bytes we request for the request body gets reflected in the upstream
  connection transmitted bytes counter for h1 and h2.
  """

  def check_upload_expectations(fixture, parsed_json, expected_transmitted_bytes,
                                expected_received_bytes):
    counters = fixture.getNighthawkCounterMapFromJson(parsed_json)
    asserts.assertCounterGreaterEqual(counters, "upstream_cx_tx_bytes_total",
                                      expected_transmitted_bytes)
    server_stats = fixture.getTestServerStatisticsJson()
    # Server side expectations start failing with larger upload sizes
    asserts.assertGreaterEqual(
        fixture.getServerStatFromJson(server_stats,
                                      "http.ingress_http.downstream_cx_rx_bytes_total"),
        expected_received_bytes)

  # TODO(#531): The dynamic-delay extension hangs unless we lower the request entity body size.
  upload_bytes = 1024 * 1024 if "static_delay" in filter_configs else 1024 * 1024 * 3
  requests = 10
  args = [
      http_test_server_fixture.getTestServerRootUri(), "--duration", "100", "--rps", "100",
      "--request-body-size",
      str(upload_bytes), "--termination-predicate",
      "benchmark.http_2xx:%s" % str(requests), "--connections", "1", "--request-method", "POST",
      "--max-active-requests", "1", "--request-header",
      "x-nighthawk-dynamic-delay-config:%s" % filter_configs
  ]
  # Test we transmit the expected amount of bytes with H1
  parsed_json, _ = http_test_server_fixture.runNighthawkClient(args)
  check_upload_expectations(http_test_server_fixture, parsed_json, upload_bytes * requests,
                            upload_bytes * requests)

  # Test we transmit the expected amount of bytes with H2
  args.append("--h2")
  parsed_json, _ = http_test_server_fixture.runNighthawkClient(args)
  # We didn't reset the server in between, so our expectation for received bytes on the server side is raised.
  check_upload_expectations(http_test_server_fixture, parsed_json, upload_bytes * requests,
                            upload_bytes * requests * 2)


def test_http_h1_termination_predicate(http_test_server_fixture):
  """Test with a termination predicate.

  Should result in successful execution, with 10 successful requests.
  We would expect 25 based on rps and duration.
  """
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      http_test_server_fixture.getTestServerRootUri(), "--duration", "5", "--rps", "500",
      "--connections", "1", "--termination-predicate", "benchmark.http_2xx:9"
  ])
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 10)


def test_http_h1_failure_predicate(http_test_server_fixture):
  """Test with a failure predicate.

  Should result in failing execution, with 10 successfull requests.
  """
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      http_test_server_fixture.getTestServerRootUri(), "--duration", "5", "--rps", "500",
      "--connections", "1", "--failure-predicate", "benchmark.http_2xx:0"
  ],
                                                               expect_failure=True)
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 1)


def test_bad_arg_error_messages(http_test_server_fixture):
  """Test arguments that pass proto validation, but are found to be no good nonetheless, result in reasonable error messages."""
  _, err = http_test_server_fixture.runNighthawkClient(
      [http_test_server_fixture.getTestServerRootUri(), "--termination-predicate ", "a:a"],
      expect_failure=True,
      as_json=False)
  assert "Bad argument: Termination predicate 'a:a' has an out of range threshold." in err


def test_multiple_backends_http_h1(multi_http_test_server_fixture):
  """Test that we can load-test multiple backends on http.

  Runs the CLI configured to use plain HTTP/1 against multiple test servers, and sanity
  checks statistics from both client and server.
  """
  nighthawk_client_args = [
      "--multi-target-path", "/", "--duration", "100", "--termination-predicate",
      "benchmark.http_2xx:24"
  ]
  for uri in multi_http_test_server_fixture.getAllTestServerRootUris():
    nighthawk_client_args.append("--multi-target-endpoint")
    nighthawk_client_args.append(uri.replace("http://", "").replace("/", ""))

  parsed_json, stderr = multi_http_test_server_fixture.runNighthawkClient(nighthawk_client_args)

  counters = multi_http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  # Assert that we at least have 1 connection per backend. It is possible that
  # the # of upstream_cx > # of backend connections for H1 as new connections
  # will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 3)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_total", 3)
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_total", 3)
  asserts.assertCounterGreater(counters, "upstream_cx_rx_bytes_total", 0)
  asserts.assertCounterGreater(counters, "upstream_cx_tx_bytes_total", 0)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "default.total_match_count", 3)
  for parsed_server_json in multi_http_test_server_fixture.getAllTestServerStatisticsJsons():
    single_2xx = multi_http_test_server_fixture.getServerStatFromJson(
        parsed_server_json, "http.ingress_http.downstream_rq_2xx")
    # Confirm that each backend receives some traffic
    asserts.assertGreaterEqual(single_2xx, 1)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_multiple_backends_https_h1(multi_https_test_server_fixture):
  """Test that we can load-test multiple backends on https.

  Runs the CLI configured to use HTTP/1 with TLS against multiple test servers, and sanity
  checks statistics from both client and server.
  """
  nighthawk_client_args = [
      "--multi-target-use-https", "--multi-target-path", "/", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:24"
  ]
  for uri in multi_https_test_server_fixture.getAllTestServerRootUris():
    nighthawk_client_args.append("--multi-target-endpoint")
    nighthawk_client_args.append(uri.replace("https://", "").replace("/", ""))

  parsed_json, stderr = multi_https_test_server_fixture.runNighthawkClient(nighthawk_client_args)

  counters = multi_https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "benchmark.http_2xx", 25)
  asserts.assertCounterGreater(counters, "upstream_cx_rx_bytes_total", 0)
  # Assert that we at least have 1 connection per backend. It is possible that
  # the # of upstream_cx > # of backend connections for H1 as new connections
  # will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 3)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_total", 3)
  asserts.assertCounterGreaterEqual(counters, "upstream_rq_pending_total", 3)
  asserts.assertCounterGreater(counters, "upstream_cx_tx_bytes_total", 0)
  asserts.assertCounterEqual(counters, "upstream_rq_total", 25)
  asserts.assertCounterEqual(counters, "default.total_match_count", 3)
  for parsed_server_json in multi_https_test_server_fixture.getAllTestServerStatisticsJsons():
    single_2xx = multi_https_test_server_fixture.getServerStatFromJson(
        parsed_server_json, "http.ingress_http.downstream_rq_2xx")
    # Confirm that each backend receives some traffic
    asserts.assertGreaterEqual(single_2xx, 1)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/sni_origin.yaml"])
def test_https_h1_sni(https_test_server_fixture):
  """Test that SNI indication works on https/h1."""
  # Verify success when we set the right host
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:2", "--request-header", "host: sni.com"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 1)
  # It is possible that the # of upstream_cx > # of backend connections for H1
  # as new connections will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 1)
  asserts.assertCounterGreaterEqual(counters, "ssl.handshake", 1)

  # Verify failure when we set no host (will get plain http)
  parsed_json, _ = https_test_server_fixture.runNighthawkClient(
      [https_test_server_fixture.getTestServerRootUri(), "--rps", "20", "--duration", "100"],
      expect_failure=True)

  # Verify success when we use plain http and don't request the sni host
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri().replace("https://", "http://"), "--rps",
      "100", "--duration", "20", "--termination-predicate", "benchmark.http_2xx:2"
  ],
                                                                expect_failure=False)

  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 1)
  # It is possible that the # of upstream_cx > # of backend connections for H1
  # as new connections will spawn if the existing clients cannot keep up with the RPS.
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 1)
  asserts.assertNotIn("ssl.handshake", counters)


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/sni_origin.yaml"])
def test_https_h2_sni(https_test_server_fixture):
  """Tests that SNI indication works on https/h1."""
  # Verify success when we set the right host
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:2", "--request-header", ":authority: sni.com",
      "--h2"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http2_total", 1)
  asserts.assertCounterEqual(counters, "ssl.handshake", 1)

  # Verify success when we set the right host
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100",
      "--termination-predicate", "benchmark.http_2xx:2", "--request-header", "host: sni.com", "--h2"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 1)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http2_total", 1)
  asserts.assertCounterEqual(counters, "ssl.handshake", 1)

  # Verify failure when we set no host (will get plain http)
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100", "--h2"
  ],
                                                                expect_failure=True)

  # Verify failure when we provide both host and :authority: (will get plain http)
  parsed_json, _ = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "100", "--h2",
      "--request-header", "host: sni.com", "--request-header", ":authority: sni.com"
  ],
                                                                expect_failure=True)


@pytest.fixture(scope="function", params=[1, 25])
def qps_parameterization_fixture(request):
  """Yield queries per second values to iterate test parameterization on."""
  param = request.param
  yield param


@pytest.fixture(scope="function", params=[5, 10])
def duration_parameterization_fixture(request):
  """Yield duration values to iterate test parameterization on."""
  param = request.param
  yield param


@pytest.mark.skipif(utility.isSanitizerRun(), reason="Unstable in sanitizer runs.")
def test_http_request_release_timing(http_test_server_fixture, qps_parameterization_fixture,
                                     duration_parameterization_fixture):
  """Test latency-sample-, query- and reply- counts in various configurations."""
  for concurrency in [1, 2]:
    parsed_json, _ = http_test_server_fixture.runNighthawkClient([
        http_test_server_fixture.getTestServerRootUri(), "--duration",
        str(duration_parameterization_fixture), "--rps",
        str(qps_parameterization_fixture), "--concurrency",
        str(concurrency)
    ])

    global_histograms = http_test_server_fixture.getNighthawkGlobalHistogramsbyIdFromJson(
        parsed_json)
    counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)

    global_result = http_test_server_fixture.getGlobalResults(parsed_json)
    actual_duration = utility.get_execution_duration_from_global_result_json(global_result)
    # Ensure Nighthawk managed to execute for at least some time.
    assert actual_duration >= 1

    # The actual duration is a float, flooring if here allows us to use
    # the GreaterEqual matchers below.
    total_requests = qps_parameterization_fixture * concurrency * math.floor(actual_duration)
    asserts.assertGreaterEqual(
        int(global_histograms["benchmark_http_client.request_to_response"]["count"]),
        total_requests)
    asserts.assertGreaterEqual(
        int(global_histograms["benchmark_http_client.queue_to_connect"]["count"]), total_requests)
    asserts.assertGreaterEqual(int(global_histograms["benchmark_http_client.latency_2xx"]["count"]),
                               total_requests)

    asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", (total_requests))
    # Give system resources some time to recover after the last execution.
    time.sleep(2)


def _send_sigterm(process):
  # Sleep for a while, under tsan the client needs a lot of time
  # to start up. 10 seconds has been determined to work through
  # emperical observation.
  time.sleep(10)
  process.terminate()


def test_cancellation_with_infinite_duration(http_test_server_fixture):
  """Test that we can use signals to cancel execution."""
  args = [
      http_test_server_fixture.nighthawk_client_path, "--concurrency", "2",
      http_test_server_fixture.getTestServerRootUri(), "--no-duration", "--output-format", "json"
  ]
  client_process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  Thread(target=(lambda: _send_sigterm(client_process))).start()
  stdout, stderr = client_process.communicate()
  client_process.wait()
  output = stdout.decode('utf-8')
  asserts.assertEqual(client_process.returncode, 0)
  parsed_json = json.loads(output)
  counters = http_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterEqual(counters, "graceful_stop_requested", 2)
  asserts.assertCounterGreaterEqual(counters, "benchmark.http_2xx", 1)


@pytest.mark.parametrize('server_config', [
    "nighthawk/test/integration/configurations/nighthawk_http_origin.yaml",
    "nighthawk/test/integration/configurations/nighthawk_track_timings.yaml"
])
def test_http_h1_response_header_latency_tracking(http_test_server_fixture, server_config):
  """Test emission and tracking of response header latencies.

  Run the CLI configured to track latencies delivered by response header from the test-server.
  Ensure that the origin_latency_statistic histogram receives the correct number of inputs.
  """
  parsed_json, _ = http_test_server_fixture.runNighthawkClient([
      http_test_server_fixture.getTestServerRootUri(), "--connections", "1", "--rps", "100",
      "--duration", "100", "--termination-predicate", "benchmark.http_2xx:99",
      "--latency-response-header-name", "x-origin-request-receipt-delta"
  ])
  global_histograms = http_test_server_fixture.getNighthawkGlobalHistogramsbyIdFromJson(parsed_json)
  asserts.assertEqual(int(global_histograms["benchmark_http_client.latency_2xx"]["count"]), 100)
  # Verify behavior is correct both with and without the timing filter enabled.
  expected_histogram_count = 99 if "nighthawk_track_timings.yaml" in server_config else 0
  asserts.assertEqual(
      int(global_histograms["benchmark_http_client.origin_latency_statistic"]["count"]),
      expected_histogram_count)


def _run_client_with_args(args):
  return utility.run_binary_with_args("nighthawk_client", args)


def test_client_help():
  """Test that passing --help behaves as expected."""
  (exit_code, output) = _run_client_with_args("--help")
  asserts.assertEqual(exit_code, 0)
  asserts.assertIn("USAGE", output)


def test_client_bad_arg():
  """Test that passing bad arguments behaves as expected."""
  (exit_code, output) = _run_client_with_args("127.0.0.1 --foo")
  asserts.assertEqual(exit_code, 1)
  asserts.assertIn("PARSE ERROR: Argument: --foo", output)


def test_client_cli_bad_uri(http_test_server_fixture):
  """Test that passing a bad URI to the client results in nice behavior."""
  _, err = http_test_server_fixture.runNighthawkClient(["http://http://foo"],
                                                       expect_failure=True,
                                                       as_json=False)
  assert "Invalid target URI" in err


@pytest.mark.parametrize('server_config',
                         ["nighthawk/test/integration/configurations/nighthawk_https_origin.yaml"])
def test_drain(https_test_server_fixture):
  """Test that the pool drain timeout is effective, and we terminate in a timely fashion.

  Sets up the test server to delay replies 100 seconds. Our execution will only last 30 seconds, so we
  expect to observe no replies. Termination should be cut short by the drain timeout, which means
  that we should have results in approximately execution duration + drain timeout = 35 seconds.
  (the pool drain timeout is hard coded to 5 seconds as of writing this).
  If drain timeout is reached, a message will be logged to the user.
  """
  parsed_json, logs = https_test_server_fixture.runNighthawkClient([
      https_test_server_fixture.getTestServerRootUri(), "--rps", "100", "--duration", "20",
      "--request-header", "x-nighthawk-test-server-config: {static_delay: \"100s\"}"
  ])
  counters = https_test_server_fixture.getNighthawkCounterMapFromJson(parsed_json)
  asserts.assertCounterGreaterEqual(counters, "upstream_cx_http1_total", 1)
  asserts.assertNotIn("benchmark.http_2xx", counters)
  asserts.assertIn("Wait for the connection pool drain timed out, proceeding to hard shutdown",
                   logs)