# Copyright 2016 Google Inc.
#
# Licensed 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.

"""Testable usage examples for Google Cloud Storage API wrapper

Each example function takes a ``client`` argument (which must be an instance
of :class:`google.cloud.storage.client.Client`) and uses it to perform a task
with the API.

To facilitate running the examples as system tests, each example is also passed
a ``to_delete`` list;  the function adds to the list any objects created which
need to be deleted during teardown.
"""

from google.cloud import storage


def snippet(func):
    """Mark ``func`` as a snippet example function."""
    func._snippet = True
    return func


@snippet
def storage_get_started(client, to_delete):
    # [START storage_get_started]
    client = storage.Client()
    bucket = client.get_bucket('bucket-id-here')
    # Then do other things...
    blob = bucket.get_blob('/remote/path/to/file.txt')
    assert blob.download_as_string() == 'My old contents!'
    blob.upload_from_string('New contents!')
    blob2 = bucket.blob('/remote/path/storage.txt')
    blob2.upload_from_filename(filename='/local/path.txt')
    # [END storage_get_started]

    to_delete.append(bucket)


@snippet
def client_bucket_acl(client, to_delete):
    bucket_name = 'system-test-bucket'
    bucket = client.bucket(bucket_name)
    bucket.create()

    # [START client_bucket_acl]
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    acl = bucket.acl
    # [END client_bucket_acl]
    to_delete.append(bucket)

    # [START acl_user_settings]
    acl.user('me@example.org').grant_read()
    acl.all_authenticated().grant_write()
    # [END acl_user_settings]

    # [START acl_save]
    acl.save()
    # [END acl_save]

    # [START acl_revoke_write]
    acl.all().grant_read()
    acl.all().revoke_write()
    # [END acl_revoke_write]

    # [START acl_save_bucket]
    bucket.acl.save(acl=acl)
    # [END acl_save_bucket]

    # [START acl_print]
    print(list(acl))
    # [{'role': 'OWNER', 'entity': 'allUsers'}, ...]
    # [END acl_print]


@snippet
def download_to_file(client, to_delete):
    # [START download_to_file]
    from google.cloud.storage import Blob

    client = storage.Client(project='my-project')
    bucket = client.get_bucket('my-bucket')
    encryption_key = 'c7f32af42e45e85b9848a6a14dd2a8f6'
    blob = Blob('secure-data', bucket, encryption_key=encryption_key)
    blob.upload_from_string('my secret message.')
    with open('/tmp/my-secure-file', 'wb') as file_obj:
        blob.download_to_file(file_obj)
    # [END download_to_file]

    to_delete.append(blob)


@snippet
def upload_from_file(client, to_delete):
    # [START upload_from_file]
    from google.cloud.storage import Blob

    client = storage.Client(project='my-project')
    bucket = client.get_bucket('my-bucket')
    encryption_key = 'aa426195405adee2c8081bb9e7e74b19'
    blob = Blob('secure-data', bucket, encryption_key=encryption_key)
    with open('my-file', 'rb') as my_file:
        blob.upload_from_file(my_file)
    # [END upload_from_file]

    to_delete.append(blob)


@snippet
def get_blob(client, to_delete):
    from google.cloud.storage.blob import Blob
    # [START get_blob]
    client = storage.Client()
    bucket = client.get_bucket('my-bucket')
    assert isinstance(bucket.get_blob('/path/to/blob.txt'), Blob)
    # <Blob: my-bucket, /path/to/blob.txt>
    assert not bucket.get_blob('/does-not-exist.txt')
    # None
    # [END get_blob]

    to_delete.append(bucket)


@snippet
def delete_blob(client, to_delete):
    # [START delete_blob]
    from google.cloud.exceptions import NotFound
    client = storage.Client()
    bucket = client.get_bucket('my-bucket')
    blobs = list(bucket.list_blobs())
    assert len(blobs) > 0
    # [<Blob: my-bucket, my-file.txt>]
    bucket.delete_blob('my-file.txt')
    try:
        bucket.delete_blob('doesnt-exist')
    except NotFound:
        pass
    # [END delete_blob]

    blob = None
    # [START delete_blobs]
    bucket.delete_blobs([blob], on_error=lambda blob: None)
    # [END delete_blobs]

    to_delete.append(bucket)


@snippet
def configure_website(client, to_delete):
    bucket_name = 'test-bucket'
    # [START configure_website]
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    bucket.configure_website('index.html', '404.html')
    # [END configure_website]

    # [START make_public]
    bucket.make_public(recursive=True, future=True)
    # [END make_public]

    to_delete.append(bucket)


@snippet
def get_bucket(client, to_delete):
    import google
    # [START get_bucket]
    try:
        bucket = client.get_bucket('my-bucket')
    except google.cloud.exceptions.NotFound:
        print('Sorry, that bucket does not exist!')
    # [END get_bucket]
    to_delete.append(bucket)


@snippet
def lookup_bucket(client, to_delete):
    from google.cloud.storage.bucket import Bucket
    # [START lookup_bucket]
    bucket = client.lookup_bucket('doesnt-exist')
    assert not bucket
    # None
    bucket = client.lookup_bucket('my-bucket')
    assert isinstance(bucket, Bucket)
    # <Bucket: my-bucket>
    # [END lookup_bucket]

    to_delete.append(bucket)


@snippet
def create_bucket(client, to_delete):
    from google.cloud.storage import Bucket
    # [START create_bucket]
    bucket = client.create_bucket('my-bucket')
    assert isinstance(bucket, Bucket)
    # <Bucket: my-bucket>
    # [END create_bucket]

    to_delete.append(bucket)


@snippet
def list_buckets(client, to_delete):
    # [START list_buckets]
    for bucket in client.list_buckets():
        print(bucket)
    # [END list_buckets]

    for bucket in client.list_buckets():
        to_delete.append(bucket)


@snippet
def policy_document(client, to_delete):
    # pylint: disable=unused-argument
    # [START policy_document]
    bucket = client.bucket('my-bucket')
    conditions = [
        ['starts-with', '$key', ''],
        {'acl': 'public-read'}]

    policy = bucket.generate_upload_policy(conditions)

    # Generate an upload form using the form fields.
    policy_fields = ''.join(
        '<input type="hidden" name="{key}" value="{value}">'.format(
            key=key, value=value)
        for key, value in policy.items()
    )

    upload_form = (
        '<form action="http://{bucket_name}.storage.googleapis.com"'
        '   method="post" enctype="multipart/form-data">'
        '<input type="text" name="key" value="my-test-key">'
        '<input type="hidden" name="bucket" value="{bucket_name}">'
        '<input type="hidden" name="acl" value="public-read">'
        '<input name="file" type="file">'
        '<input type="submit" value="Upload">'
        '{policy_fields}'
        '</form>').format(bucket_name=bucket.name, policy_fields=policy_fields)

    print(upload_form)
    # [END policy_document]


def _line_no(func):
    code = getattr(func, '__code__', None) or getattr(func, 'func_code')
    return code.co_firstlineno


def _find_examples():
    funcs = [obj for obj in globals().values()
             if getattr(obj, '_snippet', False)]
    for func in sorted(funcs, key=_line_no):
        yield func


def _name_and_doc(func):
    return func.__name__, func.__doc__


def main():
    client = storage.Client()
    for example in _find_examples():
        to_delete = []
        print('%-25s: %s' % _name_and_doc(example))
        try:
            example(client, to_delete)
        except AssertionError as failure:
            print('   FAIL: %s' % (failure,))
        except Exception as error:  # pylint: disable=broad-except
            print('  ERROR: %r' % (error,))
        for item in to_delete:
            item.delete()


if __name__ == '__main__':
    main()
