controller/rootfs/scheduler/tests/test_kube_http_client.py at e7cebb2f231f9fa74e8df614d07fb8a14013d0c3 · drycc/controller · GitHub

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
"""
Unit tests for the Deis scheduler module.

Run the tests with "./manage.py test scheduler"
"""
import requests
import requests_mock
from unittest import mock
from packaging.version import parse

from django.test import TestCase

import scheduler


def mock_session_for_version(blah=None):
    return requests.Session()


def connection_refused_matcher(request):
    raise requests.ConnectionError("connection refused")


@mock.patch('scheduler.get_k8s_session', mock_session_for_version)
class KubeHTTPClientTest(TestCase):
    """Tests kubernetes HTTP client version calls"""

    def setUp(self):
        self.adapter = requests_mock.Adapter()
        self.url = 'http://versiontest.example.com'
        self.path = '/version'

        # use the real scheduler client.
        self.scheduler = scheduler.KubeHTTPClient(self.url)
        self.scheduler.session.mount(self.url, self.adapter)

    def test_version_for_gke(self):
        """
        Ensure that version() sanitizes info from GKE clusters
        """

        cases = {
                "1.12": {"major": "1", "minor": "12-gke"},
                "1.10": {"major": "1", "minor": "10-gke"},
                "1.9": {"major": "1", "minor": "9-gke"},
                "1.8": {"major": "1", "minor": "8-gke"},
                }

        for canonical in cases:
            resp = cases[canonical]
            self.adapter.register_uri('GET', self.url + self.path, json=resp)

            expected = parse(canonical)
            actual = self.scheduler.version()

            self.assertEqual(expected, actual, "{} breaks".format(resp))

    def test_version_for_eks(self):
        """
        Ensure that version() sanitizes info from EKS clusters
        """

        cases = {
                "1.12": {"major": "1", "minor": "12+"},
                "1.10": {"major": "1", "minor": "10+"},
                "1.9": {"major": "1", "minor": "9+"},
                "1.8": {"major": "1", "minor": "8+"},
                }

        for canonical in cases:
            resp = cases[canonical]
            self.adapter.register_uri('GET', self.url + self.path, json=resp)

            expected = parse(canonical)
            actual = self.scheduler.version()

            self.assertEqual(expected, actual, "{} breaks".format(resp))

    def test_version_vanilla(self):
        """
        Ensure that version() sanitizes info from vanilla k8s clusters
        """

        cases = {
                "1.12": {"major": "1", "minor": "12"},
                "1.10": {"major": "1", "minor": "10"},
                "1.9": {"major": "1", "minor": "9"},
                "1.8": {"major": "1", "minor": "8"},
                }

        for canonical in cases:
            resp = cases[canonical]
            self.adapter.register_uri('GET', self.url + self.path, json=resp)

            expected = parse(canonical)
            actual = self.scheduler.version()

            self.assertEqual(expected, actual, "{} breaks".format(resp))