...

Text file src/github.com/datawire/ambassador/v2/python/tests/kat/t_headerrouting.py

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

     1from typing import ClassVar, Generator, Tuple, Union
     2
     3from abstract_tests import HTTP, AmbassadorTest, MappingTest, Node, ServiceType
     4from kat.harness import EDGE_STACK, Query, variants
     5
     6
     7class HeaderRoutingTest(MappingTest):
     8    parent: AmbassadorTest
     9    target: ServiceType
    10    target2: ServiceType
    11    weight: int
    12
    13    @classmethod
    14    def variants(cls) -> Generator[Node, None, None]:
    15        for v in variants(ServiceType):
    16            yield cls(v, v.clone("target2"), name="{self.target.name}")
    17
    18    # XXX This type: ignore is here because we're deliberately overriding the
    19    # parent's init to have a different signature... but it's also intimately
    20    # (nay, incestuously) related to the variant()'s yield() above, and I really
    21    # don't want to deal with that right now. So. We'll deal with it later.
    22    def init(self, target: ServiceType, target2: ServiceType):  # type: ignore
    23        MappingTest.init(self, target)
    24        self.target2 = target2
    25
    26    def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
    27        yield self.target, self.format(
    28            """
    29---
    30apiVersion: getambassador.io/v3alpha1
    31kind: Mapping
    32name:  {self.name}-target1
    33hostname: "*"
    34prefix: /{self.name}/
    35service: http://{self.target.path.fqdn}
    36"""
    37        )
    38        yield self.target2, self.format(
    39            """
    40---
    41apiVersion: getambassador.io/v3alpha1
    42kind: Mapping
    43name:  {self.name}-target2
    44hostname: "*"
    45prefix: /{self.name}/
    46service: http://{self.target2.path.fqdn}
    47headers:
    48    X-Route: target2
    49"""
    50        )
    51
    52    def queries(self):
    53        yield Query(self.parent.url(self.name + "/"))
    54        yield Query(self.parent.url(self.name + "/"), headers={"X-Route": "target2"})
    55
    56    def check(self):
    57        assert (
    58            self.results[0].backend.name == self.target.path.k8s
    59        ), f"r0 wanted {self.target.path.k8s} got {self.results[0].backend.name}"
    60        assert (
    61            self.results[1].backend.name == self.target2.path.k8s
    62        ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
    63
    64
    65class HeaderRoutingAuth(ServiceType):
    66    skip_variant: ClassVar[bool] = True
    67
    68    def __init__(self, *args, **kwargs) -> None:
    69        # Do this unconditionally, since that's part of the point of this class.
    70        kwargs[
    71            "service_manifests"
    72        ] = """
    73---
    74kind: Service
    75apiVersion: v1
    76metadata:
    77  name: {self.path.k8s}
    78spec:
    79  selector:
    80    backend: {self.path.k8s}
    81  ports:
    82  - name: http
    83    protocol: TCP
    84    port: 80
    85    targetPort: 80
    86  - name: https
    87    protocol: TCP
    88    port: 443
    89    targetPort: 443
    90---
    91apiVersion: v1
    92kind: Pod
    93metadata:
    94  name: {self.path.k8s}
    95  labels:
    96    backend: {self.path.k8s}
    97spec:
    98  containers:
    99  - name: backend
   100    image: {images[test-auth]}
   101    ports:
   102    - containerPort: 80
   103    env:
   104    - name: BACKEND
   105      value: {self.path.k8s}
   106"""
   107
   108        super().__init__(*args, **kwargs)
   109
   110    def requirements(self):
   111        yield ("url", Query("http://%s/ambassador/check/" % self.path.fqdn))
   112
   113
   114class AuthenticationHeaderRouting(AmbassadorTest):
   115    target1: ServiceType
   116    target2: ServiceType
   117    auth: ServiceType
   118
   119    def init(self):
   120        if EDGE_STACK:
   121            self.xfail = "XFailing for now, custom AuthServices not supported in Edge Stack"
   122        self.target1 = HTTP(name="target1")
   123        self.target2 = HTTP(name="target2")
   124        self.auth = HeaderRoutingAuth()
   125
   126    def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
   127        # The auth service we're using works like this:
   128        #
   129        # prefix ENDS WITH /good/ -> 200, include X-Auth-Route -> we should hit target2
   130        # prefix ENDS WITH /nohdr/ -> 200, no X-Auth-Route -> we should hit target1
   131        # anything else -> 403 -> we should see the 403
   132
   133        yield self, self.format(
   134            """
   135---
   136apiVersion: getambassador.io/v3alpha1
   137kind: AuthService
   138name:  {self.auth.path.k8s}
   139auth_service: "{self.auth.path.fqdn}"
   140proto: http
   141path_prefix: ""
   142timeout_ms: 5000
   143
   144allowed_authorization_headers:
   145- X-Auth-Route
   146- Extauth
   147"""
   148        )
   149        yield self.target1, self.format(
   150            """
   151---
   152apiVersion: getambassador.io/v3alpha1
   153kind: Mapping
   154name:  {self.name}-target1
   155hostname: "*"
   156prefix: /target/
   157service: http://{self.target1.path.fqdn}
   158"""
   159        )
   160        yield self.target2, self.format(
   161            """
   162---
   163apiVersion: getambassador.io/v3alpha1
   164kind: Mapping
   165name:  {self.name}-target2
   166hostname: "*"
   167prefix: /target/
   168service: http://{self.target2.path.fqdn}
   169headers:
   170    X-Auth-Route: Route
   171"""
   172        )
   173
   174    def queries(self):
   175        # [0]
   176        yield Query(self.url("target/"), expected=403)
   177
   178        # [1]
   179        yield Query(self.url("target/good/"), expected=200)
   180
   181        # [2]
   182        yield Query(self.url("target/nohdr/"), expected=200)
   183
   184        # [3]
   185        yield Query(self.url("target/crap/"), expected=403)
   186
   187    def check(self):
   188        # [0] should be a 403 from auth
   189        assert (
   190            self.results[0].backend.name == self.auth.path.k8s
   191        ), f"r0 wanted {self.auth.path.k8s} got {self.results[0].backend.name}"
   192
   193        # [1] should go to target2
   194        assert (
   195            self.results[1].backend.name == self.target2.path.k8s
   196        ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
   197
   198        # [2] should go to target1
   199        assert (
   200            self.results[2].backend.name == self.target1.path.k8s
   201        ), f"r2 wanted {self.target1.path.k8s} got {self.results[2].backend.name}"
   202
   203        # [3] should be a 403 from auth
   204        assert (
   205            self.results[3].backend.name == self.auth.path.k8s
   206        ), f"r3 wanted {self.auth.path.k8s} got {self.results[3].backend.name}"

View as plain text