...

Text file src/github.com/emissary-ingress/emissary/v3/python/tests/integration/manifests.py

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

     1import json
     2import os
     3import subprocess
     4from base64 import b64encode
     5from typing import Dict, Optional
     6
     7
     8def _get_images() -> Dict[str, str]:
     9    ret: Dict[str, str] = {}
    10
    11    # Keep this list in-sync with the 'push-pytest-images' Makefile target.
    12    image_names = [
    13        "test-auth",
    14        "test-shadow",
    15        "test-stats",
    16        "kat-client",
    17        "kat-server",
    18    ]
    19
    20    if image := os.environ.get("AMBASSADOR_DOCKER_IMAGE"):
    21        ret["emissary"] = image
    22    else:
    23        image_names.append("emissary")
    24
    25    try:
    26        subprocess.run(
    27            ["make"] + [f"docker/{name}.docker.push.remote" for name in image_names],
    28            check=True,
    29            stdout=subprocess.PIPE,
    30            stderr=subprocess.STDOUT,
    31            text=True,
    32        )
    33    except subprocess.CalledProcessError as err:
    34        raise Exception(f"{err.stdout}{err}") from err
    35
    36    for name in image_names:
    37        with open(f"docker/{name}.docker.push.remote", "r") as fh:
    38            # file contents:
    39            #   line 1: image ID
    40            #   line 2: tag 1
    41            #   line 3: tag 2
    42            #   ...
    43            tag = fh.readlines()[1].strip()
    44            ret[name] = tag
    45
    46    return ret
    47
    48
    49_image_cache: Optional[Dict[str, str]] = None
    50
    51
    52def get_images() -> Dict[str, str]:
    53    global _image_cache
    54    if not _image_cache:
    55        _image_cache = _get_images()
    56    return _image_cache
    57
    58
    59_file_cache: Dict[str, str] = {}
    60
    61
    62def load(manifest_name: str) -> str:
    63    if manifest_name in _file_cache:
    64        return _file_cache[manifest_name]
    65    manifest_dir = __file__[: -len(".py")]
    66    manifest_file = os.path.join(manifest_dir, manifest_name + ".yaml")
    67    manifest_content = open(manifest_file, "r").read()
    68    _file_cache[manifest_name] = manifest_content
    69    return manifest_content
    70
    71
    72def format(st: str, /, **kwargs):
    73    serviceAccountExtra = ""
    74    if os.environ.get("DEV_USE_IMAGEPULLSECRET", False):
    75        serviceAccountExtra = """
    76imagePullSecrets:
    77- name: dev-image-pull-secret
    78"""
    79    return st.format(serviceAccountExtra=serviceAccountExtra, images=get_images(), **kwargs)
    80
    81
    82def namespace_manifest(namespace: str) -> str:
    83    ret = f"""
    84---
    85apiVersion: v1
    86kind: Namespace
    87metadata:
    88  name: {namespace}
    89"""
    90
    91    if os.environ.get("DEV_USE_IMAGEPULLSECRET", None):
    92        dockercfg = {
    93            "auths": {
    94                os.path.dirname(os.environ["DEV_REGISTRY"]): {
    95                    "auth": b64encode(
    96                        (
    97                            os.environ["DOCKER_BUILD_USERNAME"]
    98                            + ":"
    99                            + os.environ["DOCKER_BUILD_PASSWORD"]
   100                        ).encode("utf-8")
   101                    ).decode("utf-8")
   102                }
   103            }
   104        }
   105        ret += f"""
   106---
   107apiVersion: v1
   108kind: Secret
   109metadata:
   110  name: dev-image-pull-secret
   111  namespace: {namespace}
   112type: kubernetes.io/dockerconfigjson
   113data:
   114  ".dockerconfigjson": "{b64encode(json.dumps(dockercfg).encode("utf-8")).decode("utf-8")}"
   115---
   116apiVersion: v1
   117kind: ServiceAccount
   118metadata:
   119  name: default
   120  namespace: {namespace}
   121imagePullSecrets:
   122- name: dev-image-pull-secret
   123"""
   124
   125    return ret
   126
   127
   128def crd_manifests() -> str:
   129    ret = ""
   130
   131    ret += namespace_manifest("emissary-system")
   132
   133    # Use .replace instead of .format because there are other '{word}' things in 'description' fields
   134    # that would cause KeyErrors when .format erroneously tries to evaluate them.
   135    ret += (
   136        load("crds")
   137        .replace("{images[emissary]}", get_images()["emissary"])
   138        .replace("{serviceAccountExtra}", format("{serviceAccountExtra}"))
   139    )
   140
   141    return ret

View as plain text