Upgrades moto to newer version (~=2.0) (apache#15051)

According to getmoto/moto#3535 (comment)
1.3.17 version of moto with a fix to be compatible with mock> 4.0.3 is
not going to be released because of breaking changes. Therefore we need
to migrate to newer version of moto.

At the same time we can get rid of the old botocore limitation, which
was added apparently to handle some test errors. We are relying fully
on what boto3 depends on.

Upgrading dependencies also discovered that mysql tests need to
be fixed because upgraded version of dependencies cause some test
failure (those turned out to be badly written tests).

(cherry picked from commit e8aa3de)

Author: potiuk

setup.py

@@ -194,8 +194,7 @@ def get_sphinx_theme_version() -> str:
 # If you change this mark you should also change ./scripts/ci/check_order_setup.py
 # Start dependencies group
 amazon = [
-    'boto3>=1.15.0,<1.16.0',
-    'botocore>=1.18.0,<1.19.0',
+    'boto3>=1.15.0,<1.18.0',
     'watchtower~=0.7.3',
 ]
 apache_beam = [
@@ -217,7 +216,7 @@ def get_sphinx_theme_version() -> str:
     'azure-keyvault>=4.1.0',
     'azure-kusto-data>=0.0.43,<0.1',
     'azure-mgmt-containerinstance>=1.5.0,<2.0',
-    'azure-mgmt-datafactory>=0.13.0',
+    'azure-mgmt-datafactory>=1.0.0,<2.0',
     'azure-mgmt-datalake-store>=0.5.0',
     'azure-mgmt-resource>=2.2.0',
     'azure-storage-blob>=12.7.0',
@@ -315,6 +314,7 @@ def get_sphinx_theme_version() -> str:
     'grpcio-gcp>=0.2.2',
     'json-merge-patch~=0.2',
     'pandas-gbq',
+    'plyvel',
 ]
 grpc = [
     'google-auth>=1.0.0, <2.0.0dev',
@@ -364,7 +364,7 @@ def get_sphinx_theme_version() -> str:
 ]
 mysql = [
     'mysql-connector-python>=8.0.11, <=8.0.22',
-    'mysqlclient>=1.3.6,<1.4',
+    'mysqlclient>=1.3.6,<3',
 ]
 neo4j = ['neo4j>=4.2.1']
 odbc = [
@@ -390,7 +390,7 @@ def get_sphinx_theme_version() -> str:
     'pinotdb>0.1.2,<1.0.0',
 ]
 plexus = [
-    'arrow>=0.16.0',
+    'arrow>=0.16.0,<1.0.0',
 ]
 postgres = [
     'psycopg2-binary>=2.7.4',
@@ -468,6 +468,7 @@ def get_sphinx_theme_version() -> str:
 # End dependencies group
 
 devel = [
+    'aws_xray_sdk',
     'beautifulsoup4~=4.7.1',
     'black',
     'blinker',
@@ -485,24 +486,23 @@ def get_sphinx_theme_version() -> str:
     'ipdb',
     'jira',
     'jsonpath-ng',
-    # HACK: Moto is not compatible with newer versions
-    # See: https://github.com/spulec/moto/issues/3535
-    'mock<4.0.3',
+    'jsondiff',
     'mongomock',
-    'moto<2',
+    'moto~=2.0',
     'mypy==0.770',
     'parameterized',
     'paramiko',
     'pipdeptree',
     'pre-commit',
-    'pylint',
+    'pylint>=2.7.0',
     'pysftp',
     'pytest~=6.0',
     'pytest-cov',
     'pytest-instafail',
     'pytest-rerunfailures~=9.1',
     'pytest-timeouts',
     'pytest-xdist',
+    'python-jose',
     'pywinrm',
     'qds-sdk>=1.9.6',
     'requests_mock',

tests/operators/test_generic_transfer.py

@@ -17,6 +17,7 @@
 # under the License.
 
 import unittest
+from contextlib import closing
 
 import pytest
 from parameterized import parameterized
@@ -43,9 +44,11 @@ def setUp(self):
 
     def tearDown(self):
         drop_tables = {'test_mysql_to_mysql', 'test_airflow'}
-        with MySqlHook().get_conn() as conn:
+        with closing(MySqlHook().get_conn()) as conn:
             for table in drop_tables:
-                conn.execute(f"DROP TABLE IF EXISTS {table}")
+                # Previous version tried to run execute directly on dbapi call, which was accidentally working
+                with closing(conn.cursor()) as cur:
+                    cur.execute(f"DROP TABLE IF EXISTS {table}")
 
     @parameterized.expand(
         [

tests/providers/amazon/aws/.gitignore

@@ -0,0 +1 @@
+local

tests/providers/amazon/aws/hooks/test_s3.py

@@ -36,6 +36,18 @@
     mock_s3 = None
 
 
+# This class needs to be separated out because if there are earlier mocks in the same class
+# the tests will fail on teardown.
+class TestAwsS3HookNoMock:
+    def test_check_for_bucket_raises_error_with_invalid_conn_id(self, monkeypatch):
+        monkeypatch.delenv('AWS_PROFILE', raising=False)
+        monkeypatch.delenv('AWS_ACCESS_KEY_ID', raising=False)
+        monkeypatch.delenv('AWS_SECRET_ACCESS_KEY', raising=False)
+        hook = S3Hook(aws_conn_id="does_not_exist")
+        with pytest.raises(NoCredentialsError):
+            hook.check_for_bucket("test-non-existing-bucket")
+
+
 @pytest.mark.skipif(mock_s3 is None, reason='moto package not present')
 class TestAwsS3Hook:
     @mock_s3
@@ -52,14 +64,6 @@ def test_check_for_bucket(self, s3_bucket):
         assert hook.check_for_bucket(s3_bucket) is True
         assert hook.check_for_bucket('not-a-bucket') is False
 
-    def test_check_for_bucket_raises_error_with_invalid_conn_id(self, s3_bucket, monkeypatch):
-        monkeypatch.delenv('AWS_PROFILE', raising=False)
-        monkeypatch.delenv('AWS_ACCESS_KEY_ID', raising=False)
-        monkeypatch.delenv('AWS_SECRET_ACCESS_KEY', raising=False)
-        hook = S3Hook(aws_conn_id="does_not_exist")
-        with pytest.raises(NoCredentialsError):
-            hook.check_for_bucket(s3_bucket)
-
     @mock_s3
     def test_get_bucket(self):
         hook = S3Hook()
@@ -156,14 +160,6 @@ def test_check_for_key(self, s3_bucket):
         assert hook.check_for_key('b', s3_bucket) is False
         assert hook.check_for_key(f's3://{s3_bucket}//b') is False
 
-    def test_check_for_key_raises_error_with_invalid_conn_id(self, monkeypatch, s3_bucket):
-        monkeypatch.delenv('AWS_PROFILE', raising=False)
-        monkeypatch.delenv('AWS_ACCESS_KEY_ID', raising=False)
-        monkeypatch.delenv('AWS_SECRET_ACCESS_KEY', raising=False)
-        hook = S3Hook(aws_conn_id="does_not_exist")
-        with pytest.raises(NoCredentialsError):
-            hook.check_for_key('a', s3_bucket)
-
     def test_get_key(self, s3_bucket):
         hook = S3Hook()
         bucket = hook.get_bucket(s3_bucket)

tests/providers/amazon/aws/hooks/test_secrets_manager.py

@@ -43,12 +43,17 @@ def test_get_secret_string(self):
         secret_value = '{"user": "test"}'
         hook = SecretsManagerHook(aws_conn_id='aws_default')
 
-        param = {
+        create_param = {
+            'Name': secret_name,
+        }
+
+        put_param = {
             'SecretId': secret_name,
             'SecretString': secret_value,
         }
 
-        hook.get_conn().put_secret_value(**param)
+        hook.get_conn().create_secret(**create_param)
+        hook.get_conn().put_secret_value(**put_param)
 
         secret = hook.get_secret(secret_name)
         assert secret == secret_value
@@ -60,12 +65,17 @@ def test_get_secret_dict(self):
         secret_value = '{"user": "test"}'
         hook = SecretsManagerHook(aws_conn_id='aws_default')
 
-        param = {
+        create_param = {
+            'Name': secret_name,
+        }
+
+        put_param = {
             'SecretId': secret_name,
             'SecretString': secret_value,
         }
 
-        hook.get_conn().put_secret_value(**param)
+        hook.get_conn().create_secret(**create_param)
+        hook.get_conn().put_secret_value(**put_param)
 
         secret = hook.get_secret_as_dict(secret_name)
         assert secret == json.loads(secret_value)
@@ -76,13 +86,17 @@ def test_get_secret_binary(self):
         secret_name = "arn:aws:secretsmanager:us-east-2:999999999999:secret:db_cluster-YYYYYYY"
         secret_value_binary = base64.b64encode(b'{"username": "test"}')
         hook = SecretsManagerHook(aws_conn_id='aws_default')
+        create_param = {
+            'Name': secret_name,
+        }
 
-        param = {
+        put_param = {
             'SecretId': secret_name,
             'SecretBinary': secret_value_binary,
         }
 
-        hook.get_conn().put_secret_value(**param)
+        hook.get_conn().create_secret(**create_param)
+        hook.get_conn().put_secret_value(**put_param)
 
         secret = hook.get_secret(secret_name)
         assert secret == base64.b64decode(secret_value_binary)

tests/providers/amazon/aws/secrets/test_secrets_manager.py

@@ -32,12 +32,19 @@ def test_aws_secrets_manager_get_connections(self, mock_get_uri):
 
     @mock_secretsmanager
     def test_get_conn_uri(self):
+
+        secret_id = 'airflow/connections/test_postgres'
+        create_param = {
+            'Name': secret_id,
+        }
+
         param = {
-            'SecretId': 'airflow/connections/test_postgres',
+            'SecretId': secret_id,
             'SecretString': 'postgresql://airflow:airflow@host:5432/airflow',
         }
 
         secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
         secrets_manager_backend.client.put_secret_value(**param)
 
         returned_uri = secrets_manager_backend.get_conn_uri(conn_id="test_postgres")
@@ -50,22 +57,36 @@ def test_get_conn_uri_non_existent_key(self):
         SecretsManagerBackend.get_connections should return None
         """
         conn_id = "test_mysql"
+
+        secret_id = 'airflow/connections/test_postgres'
+        create_param = {
+            'Name': secret_id,
+        }
+
         param = {
-            'SecretId': 'airflow/connections/test_postgres',
+            'SecretId': secret_id,
             'SecretString': 'postgresql://airflow:airflow@host:5432/airflow',
         }
 
         secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
         secrets_manager_backend.client.put_secret_value(**param)
 
         assert secrets_manager_backend.get_conn_uri(conn_id=conn_id) is None
         assert [] == secrets_manager_backend.get_connections(conn_id=conn_id)
 
     @mock_secretsmanager
     def test_get_variable(self):
-        param = {'SecretId': 'airflow/variables/hello', 'SecretString': 'world'}
+
+        secret_id = 'airflow/variables/hello'
+        create_param = {
+            'Name': secret_id,
+        }
+
+        param = {'SecretId': secret_id, 'SecretString': 'world'}
 
         secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
         secrets_manager_backend.client.put_secret_value(**param)
 
         returned_uri = secrets_manager_backend.get_variable('hello')
@@ -77,9 +98,14 @@ def test_get_variable_non_existent_key(self):
         Test that if Variable key is not present,
         SystemsManagerParameterStoreBackend.get_variables should return None
         """
-        param = {'SecretId': 'airflow/variables/hello', 'SecretString': 'world'}
+        secret_id = 'airflow/variables/hello'
+        create_param = {
+            'Name': secret_id,
+        }
+        param = {'SecretId': secret_id, 'SecretString': 'world'}
 
         secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
         secrets_manager_backend.client.put_secret_value(**param)
 
         assert secrets_manager_backend.get_variable("test_mysql") is None