<commit_before>import sys

import base64
import os
import pytest
import shutil
import subprocess
import yaml

yaml_loader = yaml.SafeLoader
yaml_dumper = yaml.SafeDumper

try:
    yaml_loader = yaml.CSafeLoader
    yaml_dumper = yaml.CSafeDumper
except AttributeError:
    pass

from typing import Any, ClassVar, Dict, List, Optional, Sequence
from typing import cast as typecast

from kat.harness import abstract_test, sanitize, Name, Node, Test, Query
from kat import manifests
from kat.utils import ShellCommand

AMBASSADOR_LOCAL = """
---
apiVersion: v1
kind: Secret
metadata:
  name: {self.path.k8s}
  annotations:
    kubernetes.io/service-account.name: {self.path.k8s}
type: kubernetes.io/service-account-token
"""


def assert_default_errors(errors, include_ingress_errors=True):
    default_errors = [
        ["",
         "Ambassador could not find core CRD definitions. Please visit https://www.getambassador.io/reference/core/crds/ for more information. You can continue using Ambassador via Kubernetes annotations, any configuration via CRDs will be ignored..."],
        ["",
         "Ambassador could not find Resolver type CRD definitions. Please visit https://www.getambassador.io/reference/core/crds/ for more information. You can continue using Ambassador via Kubernetes annotations, any configuration via CRDs will be ignored..."]
    ]

    if include_ingress_errors:
        default_errors.append(
            ["",
             "Ambassador is not permitted to read Ingress resources. Please visit https://www.getambassador.io/user-guide/ingress-controller/ for more information. You can continue using Ambassador, but Ingress resources will be ignored..."
            ]
        )

    number_of_default_errors = len(default_errors)
    assert errors[:number_of_default_errors] == default_errors

    for error in errors[number_of_default_errors:]:
        assert 'found invalid port' in error[1], "Could not find 'found invalid port' in the error {}".format(error[1])


DEV = os.environ.get("AMBASSADOR_DEV", "0").lower() in ("1", "yes", "true")


@abstract_test
class AmbassadorTest(Test):

    """
    AmbassadorTest is a top level ambassador test.
    """

    OFFSET: ClassVar[int] = 0
    IMAGE_BUILT: ClassVar[bool] = False

    _index: Optional[int] = None
    _ambassador_id: Optional[str] = None
    single_namespace: bool = False
    disable_endpoints: bool = False
    name: Name
    path: Name
    extra_ports: Optional[List[int]] = None
    debug_diagd: bool = False
    manifest_envs = ""

    env = []

    def manifests(self) -> str:
        rbac = manifests.RBAC_CLUSTER_SCOPE

        if self.debug_diagd:
            self.manifest_envs += """
    - name: AMBASSADOR_DEBUG<commit_msg>Change default value of debug_diagd to true for AmbassadorTests<commit_after>import sys

import base64
import os
import pytest
import shutil
import subprocess
import yaml

yaml_loader = yaml.SafeLoader
yaml_dumper = yaml.SafeDumper

try:
    yaml_loader = yaml.CSafeLoader
    yaml_dumper = yaml.CSafeDumper
except AttributeError:
    pass

from typing import Any, ClassVar, Dict, List, Optional, Sequence
from typing import cast as typecast

from kat.harness import abstract_test, sanitize, Name, Node, Test, Query
from kat import manifests
from kat.utils import ShellCommand

AMBASSADOR_LOCAL = """
---
apiVersion: v1
kind: Secret
metadata:
  name: {self.path.k8s}
  annotations:
    kubernetes.io/service-account.name: {self.path.k8s}
type: kubernetes.io/service-account-token
"""


def assert_default_errors(errors, include_ingress_errors=True):
    default_errors = [
        ["",
         "Ambassador could not find core CRD definitions. Please visit https://www.getambassador.io/reference/core/crds/ for more information. You can continue using Ambassador via Kubernetes annotations, any configuration via CRDs will be ignored..."],
        ["",
         "Ambassador could not find Resolver type CRD definitions. Please visit https://www.getambassador.io/reference/core/crds/ for more information. You can continue using Ambassador via Kubernetes annotations, any configuration via CRDs will be ignored..."]
    ]

    if include_ingress_errors:
        default_errors.append(
            ["",
             "Ambassador is not permitted to read Ingress resources. Please visit https://www.getambassador.io/user-guide/ingress-controller/ for more information. You can continue using Ambassador, but Ingress resources will be ignored..."
            ]
        )

    number_of_default_errors = len(default_errors)
    assert errors[:number_of_default_errors] == default_errors

    for error in errors[number_of_default_errors:]:
        assert 'found invalid port' in error[1], "Could not find 'found invalid port' in the error {}".format(error[1])


DEV = os.environ.get("AMBASSADOR_DEV", "0").lower() in ("1", "yes", "true")


@abstract_test
class AmbassadorTest(Test):

    """
    AmbassadorTest is a top level ambassador test.
    """

    OFFSET: ClassVar[int] = 0
    IMAGE_BUILT: ClassVar[bool] = False

    _index: Optional[int] = None
    _ambassador_id: Optional[str] = None
    single_namespace: bool = False
    disable_endpoints: bool = False
    name: Name
    path: Name
    extra_ports: Optional[List[int]] = None
    debug_diagd: bool = True
    manifest_envs = ""

    env = []

    def manifests(self) -> str:
        rbac = manifests.RBAC_CLUSTER_SCOPE

        if self.debug_diagd:
            self.manifest_envs += """
    - name: AMBASSADOR_DEBUG
{'fraction_non_alphanumeric': 0.09198327576804217, 'fraction_numerical': 0.0036357025995273585, 'mean_word_length': 3.498773507767784, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 6, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3848391', 'n_tokens_mistral': 1674, 'n_tokens_neox': 1580, 'n_words': 491}