...

Text file src/github.com/emissary-ingress/emissary/v3/python/tests/unit/test_irmapping.py

Documentation: github.com/emissary-ingress/emissary/v3/python/tests/unit

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

View as plain text