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
|
"""
pykube.http unittests
"""
import copy
import logging
import pykube
from . import TestCase
BASE_CONFIG = {
"clusters": [
{"name": "test-cluster", "cluster": {"server": "http://localhost:8080"}}
],
"contexts": [
{
"name": "test-cluster",
"context": {"cluster": "test-cluster", "user": "test-user"},
}
],
"users": [{"name": "test-user", "user": {}}],
"current-context": "test-cluster",
}
_log = logging.getLogger(__name__)
class TestHTTPClient(TestCase):
def setUp(self):
self.config = copy.deepcopy(BASE_CONFIG)
def ensure_no_auth(self, client):
self.assertIsNone(
client.session.cert, msg="Should not send certs when not configured"
)
self.assertNotIn(
"Authorization",
client.session.headers,
msg="Should not send basic auth when not configured",
)
def test_no_auth_with_empty_user(self):
"""
Cluster does not require any authentication--so no credentials are provided in the user info
"""
config = {
"clusters": [
{
"name": "no-auth-cluster",
"cluster": {"server": "http://localhost:8080"},
}
],
"users": [{"name": "no-auth-cluster", "user": {}}],
"contexts": [
{
"name": "no-auth-cluster",
"context": {
"cluster": "no-auth-cluster",
"user": "no-auth-cluster",
},
}
],
"current-context": "no-auth-cluster",
}
client = pykube.HTTPClient(pykube.KubeConfig(doc=config))
self.ensure_no_auth(client)
def test_no_auth_with_no_user(self):
config = {
"clusters": [
{
"name": "no-auth-cluster",
"cluster": {"server": "http://localhost:8080"},
}
],
"contexts": [
{"name": "no-auth-cluster", "context": {"cluster": "no-auth-cluster"}}
],
"current-context": "no-auth-cluster",
}
client = pykube.HTTPClient(pykube.KubeConfig(doc=config))
self.ensure_no_auth(client)
def test_build_session_bearer_token(self):
"""Test that HTTPClient correctly parses the token"""
self.config.update(
{"users": [{"name": "test-user", "user": {"token": "test"}}]}
)
_log.info("Built config: %s", self.config)
client = pykube.HTTPClient(pykube.KubeConfig(doc=self.config))
_log.debug("Checking headers %s", client.session.headers)
# TODO: session.headers is no long filled due to KubernetesHTTPAdapter!
# self.assertIn('Authorization', client.session.headers)
# self.assertEqual(client.session.headers['Authorization'], 'Bearer test')
|
