SNOW-115762: skip key pair auth tests against prod.

Author: sfc-gh-stakeda

test/conftest.py

@@ -56,8 +56,6 @@
     'port': '443',
 }
 
-IS_PUBLIC_CI = os.getenv('TRAVIS') == 'true' or os.getenv('APPVEYOR') == 'True'
-
 
 def help():
     print("""Connection parameter must be specified in parameters.py,
@@ -72,6 +70,18 @@ def help():
 """)
 
 
+@pytest.fixture(scope='session')
+def is_public_test():
+    return is_public_testaccount()
+
+
+def is_public_testaccount():
+    db_parameters = get_db_parameters()
+    return os.getenv('TRAVIS') == 'true' or \
+           os.getenv('APPVEYOR') == 'True' or \
+           db_parameters.get('account').startswith('sfctest0')
+
+
 @pytest.fixture(scope='session')
 def db_parameters():
     return get_db_parameters()
@@ -305,7 +315,7 @@ def negative_db(**kwargs):
     if not kwargs.get(u'converter_class'):
         kwargs[u'converter_class'] = DefaultConverterClass()
     cnx = create_connection(**kwargs)
-    if not IS_PUBLIC_CI:
+    if not is_public_testaccount():
         cnx.cursor().execute("alter session set SUPPRESS_INCIDENT_DUMPS=true")
     try:
         yield cnx
@@ -344,5 +354,5 @@ def test_files():
 
 def pytest_runtest_setup(item):
     for _ in item.iter_markers(name="internal"):
-        if IS_PUBLIC_CI:
+        if is_public_testaccount():
             pytest.skip("cannot run on public CI")

test/test_connection.py

@@ -414,20 +414,6 @@ def test_invalid_account_timeout():
         )
 
 
-@pytest.mark.timeout(15)
-def test_invalid_port(db_parameters):
-    with pytest.raises(OperationalError):
-        snowflake.connector.connect(
-            protocol='http',
-            account='testaccount',
-            user=db_parameters['user'],
-            password=db_parameters['password'],
-            host=db_parameters['host'],
-            port=12345,
-            login_timeout=5,
-        )
-
-
 @pytest.mark.timeout(15)
 def test_invalid_proxy(db_parameters):
     with pytest.raises(OperationalError):

test/test_key_pair_authentication.py

@@ -4,8 +4,6 @@
 # Copyright (c) 2012-2019 Snowflake Computing Inc. All right reserved.
 #
 
-import os
-
 import pytest
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
@@ -14,16 +12,11 @@
 
 import snowflake.connector
 
-NO_ACCOUNTADMIN_PRIV = os.getenv('TRAVIS') == 'true' or \
-                       os.getenv('APPVEYOR') == 'True' or \
-                       os.getenv('sf_account') == 'testaccount5'
 
+def test_different_key_length(is_public_test, request, conn_cnx, db_parameters):
+    if is_public_test:
+        pytest.skip('This test requires ACCOUNTADMIN privilege to set the public key')
 
-@pytest.mark.skipif(
-    NO_ACCOUNTADMIN_PRIV,
-    reason="Change user's public key requires accountadmin privilege"
-)
-def test_different_key_length(request, conn_cnx, db_parameters):
     db_config = {
         'protocol': db_parameters['protocol'],
         'account': db_parameters['account'],
@@ -63,11 +56,10 @@ def fin():
         snowflake.connector.connect(**db_config)
 
 
-@pytest.mark.skipif(
-    NO_ACCOUNTADMIN_PRIV,
-    reason="Change user's public key requires accountadmin privilege"
-)
-def test_multiple_key_pair(request, conn_cnx, db_parameters):
+def test_multiple_key_pair(is_public_test, request, conn_cnx, db_parameters):
+    if is_public_test:
+        pytest.skip('This test requires ACCOUNTADMIN privilege to set the public key')
+
     db_config = {
         'protocol': db_parameters['protocol'],
         'account': db_parameters['account'],

test/test_put_get_user_stage.py

@@ -12,28 +12,24 @@
 import pytest
 
 
-@pytest.mark.skipif(
-    'AWS_ACCESS_KEY_ID' not in os.environ,
-    reason="Snowflake admin account is not accessible."
-)
 def test_put_get_small_data_via_user_stage(
-        tmpdir, test_files, conn_cnx, db_parameters):
+        is_public_test, tmpdir, test_files, conn_cnx, db_parameters):
     """
     [s3] Put and Get Small Data via User Stage
     """
+    if is_public_test or 'AWS_ACCESS_KEY_ID' not in os.environ:
+        pytest.skip('This test requires to change the internal parameter')
     _put_get_user_stage(tmpdir, test_files, conn_cnx, db_parameters,
                         number_of_files=5, number_of_lines=10)
 
 
-@pytest.mark.skipif(
-    'AWS_ACCESS_KEY_ID' not in os.environ,
-    reason="Snowflake admin account is not accessible."
-)
-def test_put_get_large_data_via_user_stage(tmpdir, test_files, conn_cnx,
-                                           db_parameters):
+def test_put_get_large_data_via_user_stage(
+        is_public_test, tmpdir, test_files, conn_cnx, db_parameters):
     """
     [s3] Put and Get Large Data via User Stage
     """
+    if is_public_test or 'AWS_ACCESS_KEY_ID' not in os.environ:
+        pytest.skip('This test requires to change the internal parameter')
     _put_get_user_stage(tmpdir, test_files, conn_cnx, db_parameters,
                         number_of_files=2,
                         number_of_lines=200000)
@@ -142,17 +138,16 @@ def _put_get_user_stage(tmpdir, test_files, conn_cnx, db_parameters,
                     name=db_parameters['name']))
 
 
-@pytest.mark.skipif(
-    'AWS_ACCESS_KEY_ID' not in os.environ,
-    reason="Snowflake admin account is not accessible."
-)
-def test_put_get_duplicated_data_user_stage(tmpdir, test_files, conn_cnx,
+def test_put_get_duplicated_data_user_stage(is_public_test, tmpdir, test_files, conn_cnx,
                                             db_parameters,
                                             number_of_files=5,
                                             number_of_lines=100):
     """
     [s3] Put and Get Duplicated Data using User Stage
     """
+    if is_public_test or 'AWS_ACCESS_KEY_ID' not in os.environ:
+        pytest.skip('This test requires to change the internal parameter')
+
     logger = getLogger(__name__)
     assert 'AWS_ACCESS_KEY_ID' in os.environ, 'AWS_ACCESS_KEY_ID is missing'
     assert 'AWS_SECRET_ACCESS_KEY' in os.environ, \
@@ -302,17 +297,12 @@ def test_put_get_duplicated_data_user_stage(tmpdir, test_files, conn_cnx,
                     name=db_parameters['name']))
 
 
-@pytest.mark.skipif(
-    'AWS_ACCESS_KEY_ID' not in os.environ,
-    reason="Snowflake admin account is not accessible."
-)
-def test_get_data_user_stage(tmpdir, conn_cnx, db_parameters):
+def test_get_data_user_stage(is_public_test, tmpdir, conn_cnx, db_parameters):
     """
     SNOW-20927: get failed with 404 error
     """
-    assert 'AWS_ACCESS_KEY_ID' in os.environ, 'AWS_ACCESS_KEY_ID is missing'
-    assert 'AWS_SECRET_ACCESS_KEY' in os.environ, \
-        'AWS_SECRET_ACCESS_KEY is missing'
+    if is_public_test or 'AWS_ACCESS_KEY_ID' not in os.environ:
+        pytest.skip('This test requires to change the internal parameter')
 
     default_s3bucket = os.getenv('SF_AWS_USER_BUCKET',
                                  "sfc-dev1-regression/{0}/reg".format(