9
9
import requests
10
10
from requests .exceptions import RequestException
11
11
12
- sys .path .insert (0 , os .path .abspath (' ../../' ))
12
+ sys .path .insert (0 , os .path .abspath (" ../../" ))
13
13
from bin import cli
14
14
15
+
15
16
class MockResponse :
16
- def __init__ (self , json_data , status_code , headers , raise_for_status_flag = False , text_data = '' , ):
17
+ def __init__ (
18
+ self ,
19
+ json_data ,
20
+ status_code ,
21
+ headers ,
22
+ raise_for_status_flag = False ,
23
+ text_data = "" ,
24
+ ):
17
25
self .json_data = json_data
18
26
self .text = text_data
19
27
self .status_code = status_code
20
28
self .raise_for_status_flag = raise_for_status_flag
21
29
self .headers = headers
30
+
22
31
def raise_for_status (self ):
23
32
if self .raise_for_status_flag :
24
- self .text = ' some error occurred'
33
+ self .text = " some error occurred"
25
34
raise requests .exceptions .HTTPError ()
26
35
else :
27
36
return
37
+
28
38
def json (self ):
29
39
return self .json_data
30
40
41
+
31
42
class TestCLI (TestCase ):
32
43
33
- # pylint: disable=no-method-argument
44
+ # pylint: disable=no-method-argument
34
45
def mocked_requests_post (* args , ** kwargs ):
35
- if args [0 ] == 'https://127.0.0.1:8080/api/v1/login' :
36
- return MockResponse (
37
- json_data = { },
38
- status_code = 200 ,
39
- headers = { "location" : "/api/v1/session/df1bfacb-xxxx-xxxx-xxxx-c8f57d8f3c71" }
40
- )
41
- raise RuntimeError ("Unhandle POST request: " + args [0 ])
42
-
43
- # pylint: disable=no-method-argument
44
- def mocked_requests_get (* args , ** kwargs ):
45
- if args [0 ] == 'https://127.0.0.1:8080/api/v2/k8smanifest' :
46
- return MockResponse (
47
- json_data = {
48
- "_version" :"1.0" ,
49
- "supported_versions" :[
50
- "1.14.10" ,"1.15.7" ,"1.16.4" ,"1.17.0" ,"1.18.0"
51
- ],
52
- "version_info" :{
53
- "1.14.10" :{"_version" :"1.0" ,"min_upgrade_version" :"1.13.0" ,"relnote_url" :"https://v1-14.docs.kubernetes.io/docs/setup/release/notes/" ,"hpecsi" :"1.14" },
54
- "1.15.7" :{"_version" :"1.0" ,"min_upgrade_version" :"1.14.0" ,"relnote_url" :"https://v1-15.docs.kubernetes.io/docs/setup/release/notes/" ,"hpecsi" :"1.15" },
55
- "1.16.4" :{"_version" :"1.0" ,"min_upgrade_version" :"1.15.0" ,"relnote_url" :"https://v1-16.docs.kubernetes.io/docs/setup/release/notes/" ,"hpecsi" :"1.16" },
56
- "1.17.0" :{"_version" :"1.0" ,"min_upgrade_version" :"1.16.0" ,"relnote_url" :"https://v1-17.docs.kubernetes.io/docs/setup/release/notes/" ,"hpecsi" :"1.17" },
57
- "1.18.0" :{"_version" :"1.0" ,"min_upgrade_version" :"1.17.0" ,"relnote_url" :"https://kubernetes.io/docs/setup/release/notes/" ,"hpecsi" :"1.18" }
58
- }
46
+ if args [0 ] == "https://127.0.0.1:8080/api/v1/login" :
47
+ return MockResponse (
48
+ json_data = {},
49
+ status_code = 200 ,
50
+ headers = {
51
+ "location" : "/api/v1/session/df1bfacb-xxxx-xxxx-xxxx-c8f57d8f3c71"
52
+ },
53
+ )
54
+ raise RuntimeError ("Unhandle POST request: " + args [0 ])
55
+
56
+ # pylint: disable=no-method-argument
57
+ def mocked_requests_get (* args , ** kwargs ):
58
+ if args [0 ] == "https://127.0.0.1:8080/api/v2/k8smanifest" :
59
+ return MockResponse (
60
+ json_data = {
61
+ "_version" : "1.0" ,
62
+ "supported_versions" : [
63
+ "1.14.10" ,
64
+ "1.15.7" ,
65
+ "1.16.4" ,
66
+ "1.17.0" ,
67
+ "1.18.0" ,
68
+ ],
69
+ "version_info" : {
70
+ "1.14.10" : {
71
+ "_version" : "1.0" ,
72
+ "min_upgrade_version" : "1.13.0" ,
73
+ "relnote_url" : "https://v1-14.docs.kubernetes.io/docs/setup/release/notes/" ,
74
+ "hpecsi" : "1.14" ,
75
+ },
76
+ "1.15.7" : {
77
+ "_version" : "1.0" ,
78
+ "min_upgrade_version" : "1.14.0" ,
79
+ "relnote_url" : "https://v1-15.docs.kubernetes.io/docs/setup/release/notes/" ,
80
+ "hpecsi" : "1.15" ,
81
+ },
82
+ "1.16.4" : {
83
+ "_version" : "1.0" ,
84
+ "min_upgrade_version" : "1.15.0" ,
85
+ "relnote_url" : "https://v1-16.docs.kubernetes.io/docs/setup/release/notes/" ,
86
+ "hpecsi" : "1.16" ,
87
+ },
88
+ "1.17.0" : {
89
+ "_version" : "1.0" ,
90
+ "min_upgrade_version" : "1.16.0" ,
91
+ "relnote_url" : "https://v1-17.docs.kubernetes.io/docs/setup/release/notes/" ,
92
+ "hpecsi" : "1.17" ,
93
+ },
94
+ "1.18.0" : {
95
+ "_version" : "1.0" ,
96
+ "min_upgrade_version" : "1.17.0" ,
97
+ "relnote_url" : "https://kubernetes.io/docs/setup/release/notes/" ,
98
+ "hpecsi" : "1.18" ,
99
+ },
59
100
},
60
- status_code = 200 ,
61
- headers = { }
101
+ },
102
+ status_code = 200 ,
103
+ headers = {},
62
104
)
63
- raise RuntimeError ("Unhandle GET request: " + args [0 ])
105
+ raise RuntimeError ("Unhandle GET request: " + args [0 ])
64
106
65
- @patch (' requests.post' , side_effect = mocked_requests_post )
66
- @patch (' requests.get' , side_effect = mocked_requests_get )
107
+ @patch (" requests.post" , side_effect = mocked_requests_post )
108
+ @patch (" requests.get" , side_effect = mocked_requests_get )
67
109
def test_k8s_supported_versions (self , mock_post , mock_get ):
68
-
69
- file_data = dedent ("""[default]
110
+
111
+ file_data = dedent (
112
+ """[default]
70
113
api_host = 127.0.0.1
71
114
api_port = 8080
72
115
use_ssl = True
73
116
verify_ssl = False
74
117
warn_ssl = True
75
118
username = admin
76
- password = admin123""" )
119
+ password = admin123"""
120
+ )
77
121
78
122
tmp = tempfile .NamedTemporaryFile (delete = True )
79
123
try :
80
- tmp .write (file_data .encode (' utf-8' ))
124
+ tmp .write (file_data .encode (" utf-8" ))
81
125
tmp .flush ()
82
126
83
127
cli .HPECP_CONFIG_FILE = tmp .name
@@ -89,4 +133,3 @@ def test_k8s_supported_versions(self, mock_post, mock_get):
89
133
90
134
finally :
91
135
tmp .close ()
92
-
0 commit comments