test_resource.py

# Copyright 2023 The Crossplane Authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import dataclasses
import datetime
import unittest

import pydantic
from google.protobuf import json_format
from google.protobuf import struct_pb2 as structpb

import crossplane.function.proto.v1.run_function_pb2 as fnv1
from crossplane.function import logging, resource

from .testdata.models.io.upbound.aws.s3 import v1beta2


class TestResource(unittest.TestCase):
    def setUp(self) -> None:
        logging.configure(level=logging.Level.DISABLED)

    def test_add(self) -> None:
        @dataclasses.dataclass
        class TestCase:
            reason: str
            r: fnv1.Resource
            source: dict | structpb.Struct | pydantic.BaseModel
            want: fnv1.Resource

        cases = [
            TestCase(
                reason="Updating from a dict should work.",
                r=fnv1.Resource(),
                source={"apiVersion": "example.org", "kind": "Resource"},
                want=fnv1.Resource(
                    resource=resource.dict_to_struct(
                        {"apiVersion": "example.org", "kind": "Resource"}
                    ),
                ),
            ),
            TestCase(
                reason="Updating an existing resource from a dict should work.",
                r=fnv1.Resource(
                    resource=resource.dict_to_struct(
                        {"apiVersion": "example.org", "kind": "Resource"}
                    ),
                ),
                source={
                    "metadata": {"name": "cool"},
                },
                want=fnv1.Resource(
                    resource=resource.dict_to_struct(
                        {
                            "apiVersion": "example.org",
                            "kind": "Resource",
                            "metadata": {"name": "cool"},
                        }
                    ),
                ),
            ),
            TestCase(
                reason="Updating from a struct should work.",
                r=fnv1.Resource(),
                source=resource.dict_to_struct(
                    {"apiVersion": "example.org", "kind": "Resource"}
                ),
                want=fnv1.Resource(
                    resource=resource.dict_to_struct(
                        {"apiVersion": "example.org", "kind": "Resource"}
                    ),
                ),
            ),
            TestCase(
                reason="Updating from a Pydantic model should work.",
                r=fnv1.Resource(),
                source=v1beta2.Bucket(
                    spec=v1beta2.Spec(
                        forProvider=v1beta2.ForProvider(region="us-west-2"),
                    ),
                ),
                want=fnv1.Resource(
                    resource=resource.dict_to_struct(
                        {"spec": {"forProvider": {"region": "us-west-2"}}}
                    ),
                ),
            ),
        ]

        for case in cases:
            resource.update(case.r, case.source)
            self.assertEqual(
                json_format.MessageToDict(case.want),
                json_format.MessageToDict(case.r),
                "-want, +got",
            )

    def test_get_condition(self) -> None:
        @dataclasses.dataclass
        class TestCase:
            reason: str
            res: structpb.Struct
            typ: str
            want: resource.Condition

        cases = [
            TestCase(
                reason="Return an unknown condition if the resource has no status.",
                res=resource.dict_to_struct({}),
                typ="Ready",
                want=resource.Condition(typ="Ready", status="Unknown"),
            ),
            TestCase(
                reason="Return an unknown condition if the resource has no conditions.",
                res=resource.dict_to_struct({"status": {}}),
                typ="Ready",
                want=resource.Condition(typ="Ready", status="Unknown"),
            ),
            TestCase(
                reason="Return an unknown condition if the resource does not have the "
                "requested type of condition.",
                res=resource.dict_to_struct(
                    {
                        "status": {
                            "conditions": [
                                {
                                    "type": "Cool",
                                    "status": "True",
                                }
                            ]
                        }
                    }
                ),
                typ="Ready",
                want=resource.Condition(typ="Ready", status="Unknown"),
            ),
            TestCase(
                reason="Return a minimal condition if it exists.",
                res=resource.dict_to_struct(
                    {
                        "status": {
                            "conditions": [
                                {
                                    "type": "Ready",
                                    "status": "True",
                                }
                            ]
                        }
                    }
                ),
                typ="Ready",
                want=resource.Condition(typ="Ready", status="True"),
            ),
            TestCase(
                reason="Return a full condition if it exists.",
                res=resource.dict_to_struct(
                    {
                        "status": {
                            "conditions": [
                                {
                                    "type": "Ready",
                                    "status": "True",
                                    "reason": "Cool",
                                    "message": "This condition is very cool",
                                    "lastTransitionTime": "2023-10-02T16:30:00Z",
                                }
                            ]
                        }
                    }
                ),
                typ="Ready",
                want=resource.Condition(
                    typ="Ready",
                    status="True",
                    reason="Cool",
                    message="This condition is very cool",
                    last_transition_time=datetime.datetime(
                        year=2023,
                        month=10,
                        day=2,
                        hour=16,
                        minute=30,
                        tzinfo=datetime.UTC,
                    ),
                ),
            ),
        ]

        for case in cases:
            got = resource.get_condition(case.res, case.typ)
            self.assertEqual(
                dataclasses.asdict(case.want), dataclasses.asdict(got), "-want, +got"
            )

    def test_get_credentials(self) -> None:
        @dataclasses.dataclass
        class TestCase:
            reason: str
            req: structpb.Struct
            name: str
            want: resource.Credentials

        cases = [
            TestCase(
                reason="Return the specified credentials if they exist.",
                req=resource.dict_to_struct(
                    {"credentials": {"test": {"type": "data", "data": {"foo": "bar"}}}}
                ),
                name="test",
                want=resource.Credentials(type="data", data={"foo": "bar"}),
            ),
            TestCase(
                reason="Return empty credentials if no credentials section exists.",
                req=resource.dict_to_struct({}),
                name="test",
                want=resource.Credentials(type="data", data={}),
            ),
            TestCase(
                reason="Return empty credentials if the specified name does not exist.",
                req=resource.dict_to_struct(
                    {
                        "credentials": {
                            "nottest": {"type": "data", "data": {"foo": "bar"}}
                        }
                    }
                ),
                name="test",
                want=resource.Credentials(type="data", data={}),
            ),
        ]

        for case in cases:
            got = resource.get_credentials(case.req, case.name)
            self.assertEqual(
                dataclasses.asdict(case.want), dataclasses.asdict(got), "-want, +got"
            )

    def test_struct_to_dict(self) -> None:
        @dataclasses.dataclass
        class TestCase:
            reason: str
            s: structpb.Struct
            want: dict

        cases = [
            TestCase(
                reason="Convert a struct with no fields to an empty dictionary.",
                s=structpb.Struct(),
                want={},
            ),
            TestCase(
                reason="Convert a struct with a single field to a dictionary.",
                s=structpb.Struct(fields={"foo": structpb.Value(string_value="bar")}),
                want={"foo": "bar"},
            ),
            TestCase(
                reason="Convert a nested struct to a dictionary.",
                s=structpb.Struct(
                    fields={
                        "foo": structpb.Value(
                            struct_value=structpb.Struct(
                                fields={"bar": structpb.Value(string_value="baz")}
                            )
                        )
                    }
                ),
                want={"foo": {"bar": "baz"}},
            ),
        ]

        for case in cases:
            got = resource.struct_to_dict(case.s)
            self.assertEqual(case.want, got, "-want, +got")


if __name__ == "__main__":
    unittest.main()