6593582871

Committed 20 Oct 2023 11:13PM UTC coverage: 97.941%. First build

Build # 6593582871

Build Type

Pull #1

github-actions

Committed by

web-flow

Commit Message

Merge 5142e8044 into 50e4a8880

Pull Request Pull Request #1: feat: Initial operator

Run Details

67 of 71 branches covered (0.0%)

Branch coverage included in aggregate %.

366 of 366 new or added lines in 12 files covered. (100.0%)

361 of 366 relevant lines covered (98.63%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.77

/app/api/client_resources.py

import logging
from datetime import datetime
from typing import Any, Literal

from gql import gql
from gql.transport.exceptions import TransportQueryError
from pydantic import BaseModel, ConfigDict, Field

from app.api.protocol import TwingateClientProtocol
from app.crds import ResourceSpec


class ResourceAddress(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    type: Literal["IP", "DNS"]
    value: str


class ResourceRemoteNetwork(BaseModel):
    id: str


class ResourceSecurityPolicy(BaseModel):
    id: str


class Resource(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    id: str
    name: str
    created_at: datetime = Field(alias="createdAt")
    updated_at: datetime = Field(alias="updatedAt")
    address: ResourceAddress
    alias: str | None = None
    remote_network: ResourceRemoteNetwork = Field(alias="remoteNetwork")
    security_policy: ResourceSecurityPolicy | None = Field(
        alias="securityPolicy", default=None
    )

    @staticmethod
    def get_graphql_fragment():
        return """
            fragment ResourceFields on Resource {
                id
                name
                createdAt
                updatedAt
                address {
                  type
                  value
                }
                alias
                remoteNetwork { id }
                securityPolicy { id }
            }
        """

    def is_matching_spec(self, crd: ResourceSpec) -> bool:
        return (
            self.name == crd.name
            and self.address.value == crd.address
            and self.alias == crd.alias
            and self.remote_network.id == crd.remote_network_id
            and (self.security_policy and self.security_policy.id)
            == crd.security_policy_id
        )

    def to_spec(self, **overrides: Any) -> ResourceSpec:
        data = dict(
            id=self.id,
            name=self.name,
            address=self.address.value,
            alias=self.alias,
            remote_network_id=self.remote_network.id,
            security_policy_id=self.security_policy.id
            if self.security_policy
            else None,
        )
        data.update(overrides)
        return ResourceSpec(**data)  # type: ignore


# fmt:off

_RESOURCE_FRAGMENT = Resource.get_graphql_fragment()

QUERY_GET_RESOURCE = gql(_RESOURCE_FRAGMENT + """
    query GetResource($id: ID!) {
      resource(id: $id) {
        ...ResourceFields
      }
    }
"""
)

MUT_CREATE_RESOURCE = gql(_RESOURCE_FRAGMENT + """
    mutation CreateResource($name: String!, $address: String!, $alias: String, $remoteNetworkId: ID!, $securityPolicyId: ID) {
        resourceCreate(name: $name, address: $address, alias: $alias, remoteNetworkId: $remoteNetworkId, securityPolicyId: $securityPolicyId) {
            ok
            error
            entity {
                ...ResourceFields
            }
        }
    }
"""
)

MUT_UPDATE_RESOURCE = gql(_RESOURCE_FRAGMENT + """
    mutation UpdateResource($id: ID!, $name: String!, $address: String, $alias: String, $remoteNetworkId: ID!, $securityPolicyId: ID) {
        resourceUpdate(id: $id, name: $name, address: $address, alias: $alias, remoteNetworkId: $remoteNetworkId, securityPolicyId: $securityPolicyId) {
            ok
            error
            entity {
                ...ResourceFields
            }
        }
    }
"""
)

MUT_DELETE_RESOURCE = gql("""
mutation DeleteResource($id: ID!) {
    resourceDelete(id: $id) {
        ok
        error
    }
}
"""
)

# fmt:on


class TwingateResourceAPIs:
    def get_resource(self: TwingateClientProtocol, resource_id: str) -> Resource | None:
        try:
            result = self.execute_gql(
                QUERY_GET_RESOURCE, variable_values={"id": resource_id}
            )
            return Resource(**result["resource"]) if result["resource"] else None
        except TransportQueryError:
            logging.exception("Failed to get resource")
            return None

    def resource_create(
        self: TwingateClientProtocol, resource: ResourceSpec
    ) -> Resource | None:
        result = self.execute_mutation(
            "resourceCreate",
            MUT_CREATE_RESOURCE,
            variable_values={
                "name": resource.name,
                "address": resource.address,
                "alias": resource.alias,
                "remoteNetworkId": resource.remote_network_id,
                "securityPolicyId": resource.security_policy_id,
            },
        )

        return Resource(**result["entity"])

    def resource_update(
        self: TwingateClientProtocol, resource: ResourceSpec
    ) -> Resource | None:
        result = self.execute_mutation(
            "resourceUpdate",
            MUT_UPDATE_RESOURCE,
            variable_values={
                "id": resource.id,
                "name": resource.name,
                "address": resource.address,
                "alias": resource.alias,
                "remoteNetworkId": resource.remote_network_id,
                "securityPolicyId": resource.security_policy_id,
            },
        )
        return Resource(**result["entity"])

    def resource_delete(self: TwingateClientProtocol, resource_id: str) -> bool:
        try:
            result = self.execute_mutation(
                "resourceDelete",
                MUT_DELETE_RESOURCE,
                variable_values={"id": resource_id},
            )

            return bool(result["ok"])
        except AttributeError:
            return False

1	import logging	1✔
2	from datetime import datetime	1✔
3	from typing import Any, Literal	1✔
4
5	from gql import gql	1✔
6	from gql.transport.exceptions import TransportQueryError	1✔
7	from pydantic import BaseModel, ConfigDict, Field	1✔
8
9	from app.api.protocol import TwingateClientProtocol	1✔
10	from app.crds import ResourceSpec	1✔
11
12
13	class ResourceAddress(BaseModel):	1✔
14	model_config = ConfigDict(arbitrary_types_allowed=True)	1✔
15
16	type: Literal["IP", "DNS"]	1✔
17	value: str	1✔
18
19
20	class ResourceRemoteNetwork(BaseModel):	1✔
21	id: str	1✔
22
23
24	class ResourceSecurityPolicy(BaseModel):	1✔
25	id: str	1✔
26
27
28	class Resource(BaseModel):	1✔
29	model_config = ConfigDict(populate_by_name=True)	1✔
30
31	id: str	1✔
32	name: str	1✔
33	created_at: datetime = Field(alias="createdAt")	1✔
34	updated_at: datetime = Field(alias="updatedAt")	1✔
35	address: ResourceAddress	1✔
36	alias: str \| None = None	1✔
37	remote_network: ResourceRemoteNetwork = Field(alias="remoteNetwork")	1✔
38	security_policy: ResourceSecurityPolicy \| None = Field(	1✔
39	alias="securityPolicy", default=None
40	)
41
42	@staticmethod	1✔
43	def get_graphql_fragment():	1✔
44	return """	1✔
45	fragment ResourceFields on Resource {
46	id
47	name
48	createdAt
49	updatedAt
50	address {
51	type
52	value
53	}
54	alias
55	remoteNetwork { id }
56	securityPolicy { id }
57	}
58	"""
59
60	def is_matching_spec(self, crd: ResourceSpec) -> bool:	1✔
61	return (	1✔
62	self.name == crd.name
63	and self.address.value == crd.address
64	and self.alias == crd.alias
65	and self.remote_network.id == crd.remote_network_id
66	and (self.security_policy and self.security_policy.id)
67	== crd.security_policy_id
68	)
69
70	def to_spec(self, **overrides: Any) -> ResourceSpec:	1✔
71	data = dict(	1✔
72	id=self.id,
73	name=self.name,
74	address=self.address.value,
75	alias=self.alias,
76	remote_network_id=self.remote_network.id,
77	security_policy_id=self.security_policy.id
78	if self.security_policy
79	else None,
80	)
81	data.update(overrides)	1✔
82	return ResourceSpec(**data) # type: ignore	1✔
83
84
85	# fmt:off
86
87	_RESOURCE_FRAGMENT = Resource.get_graphql_fragment()	1✔
88
89	QUERY_GET_RESOURCE = gql(_RESOURCE_FRAGMENT + """	1✔
90	query GetResource($id: ID!) {
91	resource(id: $id) {
92	...ResourceFields
93	}
94	}
95	"""
96	)
97
98	MUT_CREATE_RESOURCE = gql(_RESOURCE_FRAGMENT + """	1✔
99	mutation CreateResource($name: String!, $address: String!, $alias: String, $remoteNetworkId: ID!, $securityPolicyId: ID) {
100	resourceCreate(name: $name, address: $address, alias: $alias, remoteNetworkId: $remoteNetworkId, securityPolicyId: $securityPolicyId) {
101	ok
102	error
103	entity {
104	...ResourceFields
105	}
106	}
107	}
108	"""
109	)
110
111	MUT_UPDATE_RESOURCE = gql(_RESOURCE_FRAGMENT + """	1✔
112	mutation UpdateResource($id: ID!, $name: String!, $address: String, $alias: String, $remoteNetworkId: ID!, $securityPolicyId: ID) {
113	resourceUpdate(id: $id, name: $name, address: $address, alias: $alias, remoteNetworkId: $remoteNetworkId, securityPolicyId: $securityPolicyId) {
114	ok
115	error
116	entity {
117	...ResourceFields
118	}
119	}
120	}
121	"""
122	)
123
124	MUT_DELETE_RESOURCE = gql("""	1✔
125	mutation DeleteResource($id: ID!) {
126	resourceDelete(id: $id) {
127	ok
128	error
129	}
130	}
131	"""
132	)
133
134	# fmt:on
135
136
137	class TwingateResourceAPIs:	1✔
138	def get_resource(self: TwingateClientProtocol, resource_id: str) -> Resource \| None:	1✔
139	try:	1✔
140	result = self.execute_gql(	1✔
141	QUERY_GET_RESOURCE, variable_values={"id": resource_id}
142	)
143	return Resource(**result["resource"]) if result["resource"] else None	1✔
144	except TransportQueryError:	1✔
145	logging.exception("Failed to get resource")	1✔
146	return None	1✔
147
148	def resource_create(	1✔
149	self: TwingateClientProtocol, resource: ResourceSpec
150	) -> Resource \| None:
151	result = self.execute_mutation(	1✔
152	"resourceCreate",
153	MUT_CREATE_RESOURCE,
154	variable_values={
155	"name": resource.name,
156	"address": resource.address,
157	"alias": resource.alias,
158	"remoteNetworkId": resource.remote_network_id,
159	"securityPolicyId": resource.security_policy_id,
160	},
161	)
162
163	return Resource(**result["entity"])	1✔
164
165	def resource_update(	1✔
166	self: TwingateClientProtocol, resource: ResourceSpec
167	) -> Resource \| None:
168	result = self.execute_mutation(	1✔
169	"resourceUpdate",
170	MUT_UPDATE_RESOURCE,
171	variable_values={
172	"id": resource.id,
173	"name": resource.name,
174	"address": resource.address,
175	"alias": resource.alias,
176	"remoteNetworkId": resource.remote_network_id,
177	"securityPolicyId": resource.security_policy_id,
178	},
179	)
180	return Resource(**result["entity"])	1✔
181
182	def resource_delete(self: TwingateClientProtocol, resource_id: str) -> bool:	1✔
183	try:	1✔
184	result = self.execute_mutation(	1✔
185	"resourceDelete",
186	MUT_DELETE_RESOURCE,
187	variable_values={"id": resource_id},
188	)
189
190	return bool(result["ok"])	1✔
191	except AttributeError:	×
192	return False	×

Twingate / kubernetes-operator / 6593582871

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous