#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import unittest
from tempfile import NamedTemporaryFile
from unittest import mock

from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.transfers.gcs_to_s3 import GCSToS3Operator

try:
    from moto import mock_s3
except ImportError:
    mock_s3 = None

TASK_ID = 'test-gcs-list-operator'
GCS_BUCKET = 'test-bucket'
DELIMITER = '.csv'
PREFIX = 'TEST'
S3_BUCKET = 's3://bucket/'
MOCK_FILES = ["TEST1.csv", "TEST2.csv", "TEST3.csv"]
S3_ACL_POLICY = "private-read"


class TestGCSToS3Operator(unittest.TestCase):

    # Test1: incremental behaviour (just some files missing)
    @mock_s3
    @mock.patch('airflow.providers.amazon.aws.transfers.gcs_to_s3.GCSHook')
    def test_execute_incremental(self, mock_hook):
        mock_hook.return_value.list.return_value = MOCK_FILES
        with NamedTemporaryFile() as f:
            gcs_provide_file = mock_hook.return_value.provide_file
            gcs_provide_file.return_value.__enter__.return_value.name = f.name

            operator = GCSToS3Operator(
                task_id=TASK_ID,
                bucket=GCS_BUCKET,
                prefix=PREFIX,
                delimiter=DELIMITER,
                dest_aws_conn_id="aws_default",
                dest_s3_key=S3_BUCKET,
                replace=False,