PYTHON-2800 Add Atlas connectivity tests for MongoDB Serverless (#669)
This commit is contained in:
parent
834500de56
commit
948ebb27f4
@ -450,11 +450,13 @@ functions:
|
||||
export ATLAS_SHRD='${atlas_shrd}'
|
||||
export ATLAS_TLS11='${atlas_tls11}'
|
||||
export ATLAS_TLS12='${atlas_tls12}'
|
||||
export ATLAS_SERVERLESS='${atlas_serverless}'
|
||||
export ATLAS_SRV_FREE='${atlas_srv_free}'
|
||||
export ATLAS_SRV_REPL='${atlas_srv_repl}'
|
||||
export ATLAS_SRV_SHRD='${atlas_srv_shrd}'
|
||||
export ATLAS_SRV_TLS11='${atlas_srv_tls11}'
|
||||
export ATLAS_SRV_TLS12='${atlas_srv_tls12}'
|
||||
export ATLAS_SRV_SERVERLESS='${atlas_srv_serverless}'
|
||||
EOT
|
||||
- command: shell.exec
|
||||
type: test
|
||||
|
||||
@ -38,11 +38,13 @@ URIS = {
|
||||
"ATLAS_FREE": os.environ.get("ATLAS_FREE"),
|
||||
"ATLAS_TLS11": os.environ.get("ATLAS_TLS11"),
|
||||
"ATLAS_TLS12": os.environ.get("ATLAS_TLS12"),
|
||||
"ATLAS_SERVERLESS": os.environ.get("ATLAS_SERVERLESS"),
|
||||
"ATLAS_SRV_REPL": os.environ.get("ATLAS_SRV_REPL"),
|
||||
"ATLAS_SRV_SHRD": os.environ.get("ATLAS_SRV_SHRD"),
|
||||
"ATLAS_SRV_FREE": os.environ.get("ATLAS_SRV_FREE"),
|
||||
"ATLAS_SRV_TLS11": os.environ.get("ATLAS_SRV_TLS11"),
|
||||
"ATLAS_SRV_TLS12": os.environ.get("ATLAS_SRV_TLS12"),
|
||||
"ATLAS_SRV_SERVERLESS": os.environ.get("ATLAS_SRV_SERVERLESS"),
|
||||
}
|
||||
|
||||
# Set this variable to true to run the SRV tests even when dnspython is not
|
||||
@ -77,6 +79,9 @@ class TestAtlasConnect(unittest.TestCase):
|
||||
def test_tls_12(self):
|
||||
connect(URIS['ATLAS_TLS12'])
|
||||
|
||||
def test_serverless(self):
|
||||
connect(URIS['ATLAS_SERVERLESS'])
|
||||
|
||||
def connect_srv(self, uri):
|
||||
connect(uri)
|
||||
self.assertIn('mongodb+srv://', uri)
|
||||
@ -102,6 +107,10 @@ class TestAtlasConnect(unittest.TestCase):
|
||||
def test_srv_tls_12(self):
|
||||
self.connect_srv(URIS['ATLAS_SRV_TLS12'])
|
||||
|
||||
@unittest.skipUnless(HAS_DNS or MUST_TEST_SRV, 'SRV requires dnspython')
|
||||
def test_srv_serverless(self):
|
||||
self.connect_srv(URIS['ATLAS_SRV_SERVERLESS'])
|
||||
|
||||
def test_uniqueness(self):
|
||||
"""Ensure that we don't accidentally duplicate the test URIs."""
|
||||
uri_to_names = defaultdict(list)
|
||||
@ -114,6 +123,5 @@ class TestAtlasConnect(unittest.TestCase):
|
||||
'duplicate values: %s' % (duplicates,))
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user