#!/usr/bin/env python

from __future__ import absolute_import, division, print_function, unicode_literals

import datetime
import os
import random
import re
import string
import sys
import pytest

try:
    from io import BytesIO
except ImportError:
    from StringIO import StringIO as BytesIO

from dropbox import (
    create_session,
    Dropbox,
    DropboxOAuth2Flow,
    DropboxTeam,
    session,
    stone_serializers,
)
from dropbox.dropbox_client import PATH_ROOT_HEADER, SELECT_USER_HEADER
from dropbox.exceptions import (
    ApiError,
    AuthError,
    # BadInputError,
    PathRootError,
)
from dropbox.files import (
    DeleteResult,
    ListFolderError,
    PathOrLink,
    SharedLinkFileInfo,
)
from dropbox.common import (
    PathRoot,
    PathRoot_validator,
)
from dropbox.session import SSLError

# Key Types
REFRESH_TOKEN_KEY = "REFRESH_TOKEN"
ACCESS_TOKEN_KEY = "DROPBOX_TOKEN"
CLIENT_ID_KEY = "CLIENT_ID"
CLIENT_SECRET_KEY = "CLIENT_SECRET"
# App Types
SCOPED_KEY = "SCOPED"
LEGACY_KEY = "LEGACY"
# User Types
USER_KEY = "USER"
TEAM_KEY = "TEAM"
# Misc types
SHARED_LINK_KEY = "DROPBOX_SHARED_LINK"

def format_env_name(app_type=SCOPED_KEY, user_type=USER_KEY, key_type=ACCESS_TOKEN_KEY):
    return '{}_{}_{}'.format(app_type, user_type, key_type)

def _value_from_env_or_die(env_name):
    value = os.environ.get(env_name)
    if value is None:
        print('Set {} environment variable to a valid value.'.format(env_name),
              file=sys.stderr)
        sys.exit(1)
    return value

_TRUSTED_CERTS_FILE = os.path.join(os.path.dirname(__file__), "trusted-certs.crt")
_EXPIRED_CERTS_FILE = os.path.join(os.path.dirname(__file__), "expired-certs.crt")

# enables testing both with and without a manually-provided CA bundle
@pytest.fixture(params=[None, _TRUSTED_CERTS_FILE], ids=["no-pinning", "pinning"])
def dbx_session(request):
    return create_session(ca_certs=request.param)


@pytest.fixture()
def dbx_from_env(dbx_session):
    oauth2_token = _value_from_env_or_die(format_env_name())
    return Dropbox(oauth2_token, session=dbx_session)


@pytest.fixture()
def refresh_dbx_from_env(dbx_session):
    refresh_token = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, REFRESH_TOKEN_KEY))
    app_key = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_ID_KEY))
    app_secret = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_SECRET_KEY))
    return Dropbox(oauth2_refresh_token=refresh_token,
                   app_key=app_key, app_secret=app_secret,
                   session=dbx_session)


@pytest.fixture()
def dbx_team_from_env(dbx_session):
    team_oauth2_token = _value_from_env_or_die(
        format_env_name(SCOPED_KEY, TEAM_KEY, ACCESS_TOKEN_KEY))
    return DropboxTeam(team_oauth2_token, session=dbx_session)


@pytest.fixture()
def dbx_app_auth_from_env(dbx_session):
    app_key = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_ID_KEY))
    app_secret = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_SECRET_KEY))
    return Dropbox(app_key=app_key, app_secret=app_secret, session=dbx_session)


@pytest.fixture()
def dbx_share_url_from_env():
    return _value_from_env_or_die(SHARED_LINK_KEY)


MALFORMED_TOKEN = 'asdf'
INVALID_TOKEN = 'z' * 62

# Need bytes type for Python3
DUMMY_PAYLOAD = string.ascii_letters.encode('ascii')

RANDOM_FOLDER = random.sample(string.ascii_letters, 15)
TIMESTAMP = str(datetime.datetime.utcnow())
STATIC_FILE = "/test.txt"

@pytest.fixture(scope='module')
def pytest_setup():
    print("Setup")
    dbx = Dropbox(_value_from_env_or_die(format_env_name()))

    try:
        dbx.files_delete(STATIC_FILE)
    except Exception:
        print("File not found")

    try:
        dbx.files_delete('/Test/%s' % TIMESTAMP)
    except Exception:
        print("File not found")

@pytest.mark.usefixtures(
    "pytest_setup",
    "dbx_from_env",
    "refresh_dbx_from_env",
    "dbx_app_auth_from_env",
    "dbx_share_url_from_env",
)
class TestDropbox:
    def test_multi_auth(self, dbx_from_env, dbx_app_auth_from_env, dbx_share_url_from_env):
        # Test for user (with oauth token)
        preview_result, resp = dbx_from_env.files_get_thumbnail_v2(
            PathOrLink.link(SharedLinkFileInfo(url=dbx_share_url_from_env))
        )
        assert resp.status_code == 200

        # Test for app (with app key and secret)
        preview_result, resp = dbx_from_env.files_get_thumbnail_v2(
            PathOrLink.link(SharedLinkFileInfo(url=dbx_share_url_from_env))
        )
        assert resp.status_code == 200

    def test_refresh(self, refresh_dbx_from_env):
        refresh_dbx_from_env.users_get_current_account()

    def test_app_auth(self, dbx_app_auth_from_env):
        dbx_app_auth_from_env.check_app(query="hello world")

    def test_downscope(self, refresh_dbx_from_env):
        refresh_dbx_from_env.users_get_current_account()
        refresh_dbx_from_env.refresh_access_token(scope=['files.metadata.read'])
        with pytest.raises(AuthError):
            # Should fail because downscoped to not include needed scope
            refresh_dbx_from_env.users_get_current_account()

    def test_rpc(self, dbx_from_env):
        dbx_from_env.files_list_folder('')

        # Test API error
        random_folder_path = '/' + \
                             ''.join(RANDOM_FOLDER)
        with pytest.raises(ApiError) as cm:
            dbx_from_env.files_list_folder(random_folder_path)
        assert isinstance(cm.value.error, ListFolderError)

    def test_upload_download(self, dbx_from_env):
        # Upload file
        random_filename = ''.join(RANDOM_FOLDER)
        random_path = '/Test/%s/%s' % (TIMESTAMP, random_filename)
        test_contents = DUMMY_PAYLOAD
        dbx_from_env.files_upload(test_contents, random_path)

        # Download file
        _, resp = dbx_from_env.files_download(random_path)
        assert DUMMY_PAYLOAD == resp.content

        # Cleanup folder
        dbx_from_env.files_delete('/Test/%s' % TIMESTAMP)

    def test_bad_upload_types(self, dbx_from_env):
        with pytest.raises(TypeError):
            dbx_from_env.files_upload(BytesIO(b'test'), '/Test')

    def test_clone_when_user_linked(self, dbx_from_env):
        new_dbx = dbx_from_env.clone()
        assert dbx_from_env is not new_dbx
        assert isinstance(new_dbx, dbx_from_env.__class__)

    def test_with_path_root_constructor(self, dbx_from_env):
        # Verify valid mode types
        for path_root in (
            PathRoot.home,
            PathRoot.root("123"),
            PathRoot.namespace_id("123"),
        ):
            dbx_new = dbx_from_env.with_path_root(path_root)
            assert dbx_new is not dbx_from_env

            expected = stone_serializers.json_encode(PathRoot_validator, path_root)
            assert dbx_new._headers.get(PATH_ROOT_HEADER) == expected

        # verify invalid mode raises ValueError
        with pytest.raises(ValueError):
            dbx_from_env.with_path_root(None)

    def test_path_root(self, dbx_from_env):
        root_info = dbx_from_env.users_get_current_account().root_info
        root_ns = root_info.root_namespace_id
        home_ns = root_info.home_namespace_id

        # verify home mode
        dbxpr = dbx_from_env.with_path_root(PathRoot.home)
        dbxpr.files_list_folder('')

        # verify root mode
        dbxpr = dbx_from_env.with_path_root(PathRoot.root(root_ns))
        dbxpr.files_list_folder('')

        # verify namespace_id mode
        dbxpr = dbx_from_env.with_path_root(PathRoot.namespace_id(home_ns))
        dbxpr.files_list_folder('')

    def test_path_root_err(self, dbx_from_env):
        # verify invalid namespace return is_no_permission error
        dbxpr = dbx_from_env.with_path_root(PathRoot.namespace_id("1234567890"))
        with pytest.raises(PathRootError) as cm:
            dbxpr.files_list_folder('')
        assert cm.value.error.is_no_permission()

        dbxpr = dbx_from_env.with_path_root(PathRoot.root("1234567890"))
        with pytest.raises(PathRootError) as cm:
            dbxpr.files_list_folder('')
        assert cm.value.error.is_invalid_root()

    def test_versioned_route(self, dbx_from_env):
        # Upload a test file
        dbx_from_env.files_upload(DUMMY_PAYLOAD, STATIC_FILE)

        # Delete the file with v2 route
        resp = dbx_from_env.files_delete_v2(STATIC_FILE)
        # Verify response type is of v2 route
        assert isinstance(resp, DeleteResult)

@pytest.mark.usefixtures(
    "pytest_setup",
    "dbx_team_from_env",
)
class TestDropboxTeam:
    def test_team(self, dbx_team_from_env):
        dbx_team_from_env.team_groups_list()
        r = dbx_team_from_env.team_members_list()
        if r.members:
            # Only test assuming a member if there is a member
            team_member_id = r.members[0].profile.team_member_id
            dbx_team_from_env.as_user(team_member_id).files_list_folder('')

    def test_as_user(self, dbx_team_from_env):
        dbx_as_user = dbx_team_from_env.as_user('1')
        path_root = PathRoot.root("123")

        dbx_new = dbx_as_user.with_path_root(path_root)

        assert isinstance(dbx_new, Dropbox)
        assert dbx_new._headers.get(SELECT_USER_HEADER) == '1'

        expected = stone_serializers.json_encode(PathRoot_validator, path_root)
        assert dbx_new._headers.get(PATH_ROOT_HEADER) == expected

    def test_as_admin(self, dbx_team_from_env):
        dbx_as_admin = dbx_team_from_env.as_admin('1')
        assert isinstance(dbx_as_admin, Dropbox)

    def test_clone_when_team_linked(self, dbx_team_from_env):
        new_dbxt = dbx_team_from_env.clone()
        assert dbx_team_from_env is not new_dbxt
        assert isinstance(new_dbxt, dbx_team_from_env.__class__)

def test_default_oauth2_urls():
    flow_obj = DropboxOAuth2Flow('dummy_app_key', 'dummy_app_secret',
        'http://localhost/dummy', 'dummy_session', 'dbx-auth-csrf-token')

    assert re.match(
        r'^https://{}/oauth2/authorize\?'.format(re.escape(session.WEB_HOST)),
        flow_obj._get_authorize_url('http://localhost/redirect', 'state', 'legacy'),
    )

    assert flow_obj.build_url(
        '/oauth2/authorize'
    ) == 'https://{}/oauth2/authorize'.format(session.API_HOST)

    assert flow_obj.build_url(
        '/oauth2/authorize', host=session.WEB_HOST
    ) == 'https://{}/oauth2/authorize'.format(session.WEB_HOST)

def test_bad_auth(dbx_session):
    # Test malformed token
    malformed_token_dbx = Dropbox(MALFORMED_TOKEN, session=dbx_session)
    # TODO: backend is no longer returning `BadInputError`
    # with pytest.raises(BadInputError,) as cm:
    #     malformed_token_dbx.files_list_folder('')
    # assert 'token is malformed' in cm.value.message
    with pytest.raises(AuthError):
        malformed_token_dbx.files_list_folder('')

    # Test reasonable-looking invalid token
    invalid_token_dbx = Dropbox(INVALID_TOKEN, session=dbx_session)
    with pytest.raises(AuthError) as cm:
        invalid_token_dbx.files_list_folder('')
    assert cm.value.error.is_invalid_access_token()

def test_bad_pins():
    # sanity-check: if we're pinning using expired pins, we should fail w/ an SSL error
    _dbx = Dropbox("dummy_token", ca_certs=_EXPIRED_CERTS_FILE)
    with pytest.raises(SSLError,):
        _dbx.files_list_folder('')

def test_bad_pins_session():
    _session = create_session(ca_certs=_EXPIRED_CERTS_FILE)
    _dbx = Dropbox("dummy_token2", session=_session)
    with pytest.raises(SSLError,):
        _dbx.files_list_folder('')