1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
from typing import Any
from typing import Dict
from typing import Optional
from unittest.mock import MagicMock
import pytest
from pytest_mock import MockFixture
import pykube
from pykube.objects import NamespacedAPIObject
from pykube.objects import Pod
from pykube.utils import obj_merge
def test_api_object():
pod = Pod(None, {"metadata": {"name": "myname"}})
assert repr(pod) == "<Pod myname>"
assert str(pod) == "myname"
assert pod.metadata == {"name": "myname"}
assert pod.labels == {}
assert pod.annotations == {}
def test_object_factory_succeeds():
api = MagicMock()
api.resource_list.return_value = {
"resources": [
{"kind": "ExampleObject", "namespaced": True, "name": "exampleobjects"}
]
}
ExampleObject = pykube.object_factory(api, "example.org/v1", "ExampleObject")
assert ExampleObject.kind == "ExampleObject"
assert ExampleObject.endpoint == "exampleobjects"
assert ExampleObject.version == "example.org/v1"
assert NamespacedAPIObject in ExampleObject.mro()
def test_object_factory_raises_for_unknown_kind():
api = MagicMock()
api.resource_list.return_value = {
"resources": [
{"kind": "ExampleObject", "namespaced": True, "name": "exampleobjects"}
]
}
with pytest.raises(ValueError):
pykube.object_factory(api, "example.org/v1", "OtherObject")
def test_set_annotation():
pod = Pod(None, {"metadata": {"name": "myname"}})
pod.annotations["foo"] = "bar"
assert pod.annotations["foo"] == "bar"
def test_set_label():
pod = Pod(None, {"metadata": {"name": "myname"}})
pod.labels["foo"] = "bar"
assert pod.labels["foo"] == "bar"
def test_update():
pod = Pod(
None,
{"metadata": {"name": "john", "kind": "test"}, "annotations": "a long string"},
)
pod.obj = {"metadata": {"name": "john"}}
pod.api = MagicMock()
pod.api.patch.return_value.json.return_value = obj_merge(
pod.obj, pod._original_obj, False
)
pod.update(is_strategic=False)
assert pod.metadata == {"name": "john"}
assert pod.annotations == {}
@pytest.mark.parametrize(
"port,expected_request_kwargs",
[
# empty port - default service port should be used
(
None,
{
"url": "services/test_service:9090/proxy//",
"namespace": "test_namespace",
"version": "v1",
},
),
# override the default port, should be included in url
(
8080,
{
"url": "services/test_service:8080/proxy//",
"namespace": "test_namespace",
"version": "v1",
},
),
],
)
def test_service_proxy_http_request(
mocker: MockFixture, port: Optional[int], expected_request_kwargs: Dict[str, Any]
) -> None:
mock_client = mocker.MagicMock(spec=pykube.http.HTTPClient)
service = pykube.Service(
mock_client,
{
"apiVersion": "core/v1",
"kind": "Service",
"metadata": {
"name": "test_service",
"namespace": "test_namespace",
},
"spec": {"ports": [{"port": 9090}]},
},
)
service.proxy_http_request("GET", "/", port)
request = mock_client.request
assert request.called
assert request.call_args_list[0][0] == ("GET",)
assert request.call_args_list[0][1] == expected_request_kwargs
|
