14
14
15
15
import asyncio
16
16
from threading import Thread
17
+ from typing import Union
17
18
18
19
from mock import patch
19
20
from mocks import FakeAlloyDBClient
20
21
from mocks import FakeCredentials
21
22
import pytest
22
23
23
24
from google .cloud .alloydb .connector import Connector
25
+ from google .cloud .alloydb .connector import IPTypes
24
26
25
27
26
28
def test_Connector_init (credentials : FakeCredentials ) -> None :
@@ -36,6 +38,58 @@ def test_Connector_init(credentials: FakeCredentials) -> None:
36
38
connector .close ()
37
39
38
40
41
+ def test_Connector_init_bad_ip_type (credentials : FakeCredentials ) -> None :
42
+ """Test that Connector errors due to bad ip_type str."""
43
+ bad_ip_type = "BAD-IP-TYPE"
44
+ with pytest .raises (ValueError ) as exc_info :
45
+ Connector (ip_type = bad_ip_type , credentials = credentials )
46
+ assert (
47
+ exc_info .value .args [0 ]
48
+ == f"Incorrect value for ip_type, got '{ bad_ip_type } '. Want one of: 'PUBLIC', 'PRIVATE'."
49
+ )
50
+
51
+
52
+ @pytest .mark .parametrize (
53
+ "ip_type, expected" ,
54
+ [
55
+ (
56
+ "private" ,
57
+ IPTypes .PRIVATE ,
58
+ ),
59
+ (
60
+ "PRIVATE" ,
61
+ IPTypes .PRIVATE ,
62
+ ),
63
+ (
64
+ IPTypes .PRIVATE ,
65
+ IPTypes .PRIVATE ,
66
+ ),
67
+ (
68
+ "public" ,
69
+ IPTypes .PUBLIC ,
70
+ ),
71
+ (
72
+ "PUBLIC" ,
73
+ IPTypes .PUBLIC ,
74
+ ),
75
+ (
76
+ IPTypes .PUBLIC ,
77
+ IPTypes .PUBLIC ,
78
+ ),
79
+ ],
80
+ )
81
+ def test_Connector_init_ip_type (
82
+ ip_type : Union [str , IPTypes ], expected : IPTypes , credentials : FakeCredentials
83
+ ) -> None :
84
+ """
85
+ Test to check whether the __init__ method of Connector
86
+ properly sets ip_type.
87
+ """
88
+ connector = Connector (credentials = credentials , ip_type = ip_type )
89
+ assert connector ._ip_type == expected
90
+ connector .close ()
91
+
92
+
39
93
def test_Connector_context_manager (credentials : FakeCredentials ) -> None :
40
94
"""
41
95
Test to check whether the __init__ method of Connector
@@ -84,6 +138,28 @@ def test_connect(credentials: FakeCredentials, fake_client: FakeAlloyDBClient) -
84
138
assert connection is True
85
139
86
140
141
+ def test_connect_bad_ip_type (
142
+ credentials : FakeCredentials , fake_client : FakeAlloyDBClient
143
+ ) -> None :
144
+ """Test that Connector.connect errors due to bad ip_type str."""
145
+ with Connector (credentials = credentials ) as connector :
146
+ connector ._client = fake_client
147
+ bad_ip_type = "BAD-IP-TYPE"
148
+ with pytest .raises (ValueError ) as exc_info :
149
+ connector .connect (
150
+ "projects/test-project/locations/test-region/clusters/test-cluster/instances/test-instance" ,
151
+ "pg8000" ,
152
+ user = "test-user" ,
153
+ password = "test-password" ,
154
+ db = "test-db" ,
155
+ ip_type = bad_ip_type ,
156
+ )
157
+ assert (
158
+ exc_info .value .args [0 ]
159
+ == f"Incorrect value for ip_type, got '{ bad_ip_type } '. Want one of: 'PUBLIC', 'PRIVATE'."
160
+ )
161
+
162
+
87
163
def test_connect_unsupported_driver (credentials : FakeCredentials ) -> None :
88
164
"""
89
165
Test that connector.connect errors with unsupported database driver.
0 commit comments