...

Text file src/github.com/datawire/ambassador/v2/python/tests/unit/test_irmapping.py

Documentation: github.com/datawire/ambassador/v2/python/tests/unit

     1import logging
     2
     3import pytest
     4
     5logging.basicConfig(
     6    level=logging.INFO,
     7    format="%(asctime)s test %(levelname)s: %(message)s",
     8    datefmt="%Y-%m-%d %H:%M:%S",
     9)
    10
    11logger = logging.getLogger("ambassador")
    12
    13from ambassador import IR, Config
    14from ambassador.fetch import ResourceFetcher
    15from ambassador.utils import NullSecretHandler
    16
    17
    18def _get_ir_config(yaml):
    19    aconf = Config()
    20    fetcher = ResourceFetcher(logger, aconf)
    21    fetcher.parse_yaml(yaml)
    22    aconf.load_all(fetcher.sorted())
    23
    24    secret_handler = NullSecretHandler(logger, None, None, "0")
    25    ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
    26
    27    assert ir
    28    return ir
    29
    30
    31@pytest.mark.compilertest
    32def test_ir_mapping():
    33    yaml = """
    34apiVersion: getambassador.io/v3alpha1
    35kind: Mapping
    36name: slowsvc-slow
    37namespace: ambassador
    38prefix: /slow/
    39service: slowsvc
    40timeout_ms: 1000
    41docs:
    42  path: /endpoint
    43  display_name: "slow service"
    44  timeout_ms: 8000
    45"""
    46
    47    conf = _get_ir_config(yaml)
    48    all_mappings = []
    49    for i in conf.groups.values():
    50        all_mappings = all_mappings + i.mappings
    51
    52    slowsvc_mappings = [x for x in all_mappings if x["name"] == "slowsvc-slow"]
    53    assert len(slowsvc_mappings) == 1
    54    print(slowsvc_mappings[0].as_dict())
    55    assert slowsvc_mappings[0].docs["timeout_ms"] == 8000

View as plain text