import os
import random
import subprocess

from unittest.mock import MagicMock
from unittest.mock import patch

import uuid

import boto3

from garcon_contrib.aws import garcon_s3
from garcon_contrib.aws.utils import garcon_s3 as s3_utils

from moto import mock_aws
import pytest
import smart_open


def create_bucket(b_name=None):
    """Create an S3 bucket.

    Note:
        This cannot be a fixture – otherwise the mock doesn't retain the state
        of the bucket (including files that may have been saved).

    Return:
        boto3.s3.Bucket: newly created S3 bucket
    """

    bucket_name = b_name or str(uuid.uuid4()).lower()[:5]
    s3 = boto3.resource('s3')
    s3.create_bucket(Bucket=bucket_name)
    return s3.Bucket(bucket_name)


def create_random_file(bucket, file_name=None):
    """Create a random file in an S3 bucket.

    Return:
        boto3.s3.Key: newly created S3 bucket key containing a random string
    """

    s3 = boto3.resource('s3')
    file_name = file_name or str(uuid.uuid4()).lower()[:5]
    current_file = s3.Object(bucket, file_name)
    current_file.put(Body='random value')
    return current_file


def get_bucket_url(bucket):
    """Get the bucket url for a given bucket.

    Return:
        str: a well-formed S3 url to the given bucket
    """

    return 's3://{bucket_name}/'.format(bucket_name=bucket.name)


@pytest.mark.skipif(
    os.environ.get('PYTHON_SKIP_MOTO'),
    reason='Moto is not supported in this environment.')
@mock_aws
def test_remove_files_from_path(monkeypatch):
    """Test the test_remove_files_from_path for best case scenario.
    File(s) exist and are deleted properly.
    """

    bucket = create_bucket()
    bucket_url = get_bucket_url(bucket)
    file_count = int(random.random() * 10) + 1

    for i in range(file_count):
        create_random_file(bucket.name)

    task_response_return_files = garcon_s3.remove_files_from_path(
        MagicMock(), bucket_url, True)
    deleted_files = task_response_return_files.get('s3.files_removed')
    assert len(deleted_files) == file_count

    task_response_not_return_files = garcon_s3.remove_files_from_path(
        MagicMock(), bucket_url, False)
    assert task_response_not_return_files == {}


@patch.object(garcon_s3.boto3, "resource")
def test_remove_file_with_errors(resource_mock, monkeypatch):
    """Test the test_remove_files_from_path in case of an error from S3.
    """
    mock_key = MagicMock()
    mock_key.key = 'test_key'

    mock_bucket = MagicMock
    mock_bucket.objects = MagicMock()
    mock_bucket.objects.filter = lambda Prefix: [mock_key]
    mock_bucket.delete_objects = MagicMock()
    mock_bucket.delete_objects.return_value = {'Errors': 'Some error'}

    monkeypatch.setattr(
        resource_mock, 'Bucket',
        MagicMock(return_value=mock_bucket))

    monkeypatch.setattr(
        s3_utils, 'extract_bucket_path', MagicMock(
            return_value=('cucumbers', 'destination_s3_path')))

    with pytest.raises(Exception):
        garcon_s3.remove_files_from_path(
            MagicMock(), 's3://cucumbers/test/test', True)


@pytest.mark.skipif(
    os.environ.get('PYTHON_SKIP_MOTO'),
    reason='Moto is not supported in this environment.')
@mock_aws
def test_create_object(monkeypatch):
    """Test file being created on s3
    """

    monkeypatch.setattr(
        s3_utils, 'extract_bucket_path',
        MagicMock(return_value=('somekey', 'somevalue')))

    with pytest.raises(AssertionError):
        garcon_s3.create_object(
            MagicMock(),
            's3://somebucket/somefolder/',
            'some content')


@pytest.mark.skipif(
    os.environ.get('PYTHON_SKIP_MOTO'),
    reason='Moto is not supported in this environment.')
@mock_aws
def test_join_s3_objects(monkeypatch):
    """Test command in the right syntax
    """

    bucket = create_bucket('test_bucket')

    file_count = 3

    for i in range(file_count):
        create_random_file(bucket.name, 'd/file{}'.format(i))

    expected_command = 'cat  <(aws s3 cp s3://test_bucket/d/file0 -)'
    expected_command += '  <(aws s3 cp s3://test_bucket/d/file1 -)'
    expected_command += '  <(aws s3 cp s3://test_bucket/d/file2 -)'
    expected_command += '  | aws s3 cp - s3://test_bucket/dest_object'
    expected_command = expected_command.format(bucket=bucket.name)

    bucket.objects.filter = lambda Prefix: [
            {'key': 'd/file0'},
            {'key': 'd/file1'},
            {'key': 'd/file2'}]

    with patch('subprocess.call', return_value=None):
        garcon_s3.join_s3_objects(
            MagicMock(),
            's3://test_bucket/d',
            's3://test_bucket/dest_object')
        assert subprocess.call.called
        subprocess.call.assert_called_with(
            ['bash', '-c', expected_command])


@patch.object(garcon_s3.boto3, "resource")
def test_stream_join_s3_objects(resource_mock, monkeypatch):
    """Test stream join s3 object
    """

    mock_key = MagicMock()
    mock_key.key = 'obj1'

    mock_bucket = MagicMock
    mock_bucket.objects = MagicMock()
    mock_bucket.objects.filter = lambda Prefix: [mock_key]

    smart_open_obj = MagicMock()
    smart_open_obj.write = lambda line: None

    monkeypatch.setattr(
        s3_utils, 'extract_bucket_path',
        MagicMock(return_value=('sample_bucket', 'sample_key')))

    monkeypatch.setattr(
        smart_open, 'smart_open',
        MagicMock(return_value=smart_open_obj))

    monkeypatch.setattr(
        resource_mock, 'Bucket',
        MagicMock(return_value=mock_bucket))

    garcon_s3.stream_join_s3_objects(
        MagicMock(), 'source_s3_path', 'destination_s3_path')

    assert smart_open_obj.__enter__.called
    assert smart_open_obj.__exit__.called