25
25
from unittest import TestCase
26
26
27
27
import requests
28
- from mock import patch
28
+ import six
29
+
30
+ if six .PY2 :
31
+ from io import BytesIO as StringIO # noqa: F811
32
+ else :
33
+ from io import StringIO
29
34
30
35
31
36
class MockResponse :
@@ -55,70 +60,6 @@ def json(self):
55
60
56
61
57
62
class TestCLI (TestCase ):
58
-
59
- # pylint: disable=no-method-argument
60
- def mocked_requests_post (* args , ** kwargs ):
61
- if args [0 ] == "https://127.0.0.1:8080/api/v1/login" :
62
- return MockResponse (
63
- json_data = {},
64
- status_code = 200 ,
65
- headers = {
66
- "location" : (
67
- "/api/v1/session/df1bfacb-xxxx-xxxx-xxxx-c8f57d8f3c71"
68
- )
69
- },
70
- )
71
- raise RuntimeError ("Unhandle POST request: " + args [0 ])
72
-
73
- # pylint: disable=no-method-argument
74
- def mocked_requests_get (* args , ** kwargs ):
75
- if args [0 ] == "https://127.0.0.1:8080/api/v2/k8scluster" :
76
- return MockResponse (
77
- json_data = {
78
- "_links" : {"self" : {"href" : "/api/v2/k8scluster" }},
79
- "_embedded" : {
80
- "k8sclusters" : [
81
- {
82
- "_links" : {
83
- "self" : {"href" : "/api/v2/k8scluster/20" }
84
- },
85
- "label" : {
86
- "name" : "def" ,
87
- "description" : "my cluster" ,
88
- },
89
- "k8s_version" : "1.17.0" ,
90
- "pod_network_range" : "10.192.0.0/12" ,
91
- "service_network_range" : "10.96.0.0/12" ,
92
- "pod_dns_domain" : "cluster.local" ,
93
- "created_by_user_id" : "/api/v1/user/5" ,
94
- "created_by_user_name" : "admin" ,
95
- "created_time" : 1588260014 ,
96
- "k8shosts_config" : [
97
- {
98
- "node" : "/api/v2/worker/k8shost/4" ,
99
- "role" : "worker" ,
100
- },
101
- {
102
- "node" : "/api/v2/worker/k8shost/5" ,
103
- "role" : "master" ,
104
- },
105
- ],
106
- "status" : "ready" ,
107
- "status_message" : "really ready" ,
108
- "api_endpoint_access" : "api:1234" ,
109
- "dashboard_endpoint_access" : "dashboard:1234" ,
110
- "admin_kube_config" : "xyz==" ,
111
- "dashboard_token" : "abc==" ,
112
- "persistent_storage" : {"nimble_csi" : False },
113
- }
114
- ]
115
- },
116
- },
117
- status_code = 200 ,
118
- headers = {},
119
- )
120
- raise RuntimeError ("Unhandle GET request: " + args [0 ])
121
-
122
63
def setUp (self ):
123
64
file_data = dedent (
124
65
"""[default]
@@ -141,18 +82,31 @@ def setUp(self):
141
82
self .cli = cli
142
83
self .cli .HPECP_CONFIG_FILE = self .tmpFile .name
143
84
85
+ self .saved_stdout = sys .stdout
86
+ self .out = StringIO ()
87
+ sys .stdout = self .out
88
+
144
89
def tearDown (self ):
145
90
self .tmpFile .close ()
91
+ sys .stdout = self .saved_stdout
146
92
147
- def test_autocomplete_bash (self ):
93
+ def test_config_file_missing (self ):
148
94
149
- hpecp = self .cli .CLI ()
150
- hpecp .autocomplete .bash ()
95
+ with self .assertRaises (SystemExit ) as cm :
96
+ self .cli .HPECP_CONFIG_FILE = "this_file_should_not_exist"
97
+ self .cli .get_client ()
151
98
152
- # TODO move this to tests/library/k8s_cluster_test.py
153
- @patch ("requests.post" , side_effect = mocked_requests_post )
154
- @patch ("requests.get" , side_effect = mocked_requests_get )
155
- def test_k8scluster_list (self , mock_post , mock_get ):
99
+ self .assertEqual (cm .exception .code , 1 )
100
+
101
+ self .assertEqual (
102
+ self .out .getvalue (),
103
+ "Could not find configuration file 'this_file_should_not_exist'\n " ,
104
+ )
105
+
106
+ def test_autocomplete_bash (self ):
156
107
157
- hpecp = self .cli .CLI ()
158
- hpecp .k8scluster .list ()
108
+ try :
109
+ hpecp = self .cli .CLI ()
110
+ hpecp .autocomplete .bash ()
111
+ except Exception :
112
+ self .fail ("Unexpected exception." )
0 commit comments