# -*- coding: utf-8 -*-
# Copyright (c) 2012 OpenStack Foundation
#
# 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.

from collections import Counter
import numbers
from six.moves import urllib
import unittest
import os
import tarfile
import zlib
import mock
import six
from six import BytesIO
from shutil import rmtree
from tempfile import mkdtemp
from eventlet import sleep
from mock import patch, call
from test.unit.common.middleware.helpers import FakeSwift
from
 swift.common import utils, constraints
from swift.common.header_key_dict import HeaderKeyDict
from swift.common.middleware import bulk
from swift.common.swob
 import Request, Response, HTTPException, \
    HTTPNoContent, HTTPCreated
from swift.common.http import HTTP_NOT_FOUND, HTTP_UNAUTHORIZED


class FakeApp(object):
    def __init__(self):
        self.calls = 0
        self.delete_paths = []
        self.max_pathlen = 100
        self.del_cont_total_calls = 2
        self.del_cont_cur_call = 0

    def __call__(self, env, start_response):
        self.calls += 1
        if env['PATH_INFO'].startswith('/unauth/'):
            if env['PATH_INFO'].endswith('/c/f_ok'):
                return Response(status='204 No Content')(env, start_response)
            return Response(status=401)(env, start_response)
        if env['PATH_INFO'].startswith('/create_cont/'):
            if env['REQUEST_METHOD'] == 'HEAD':
                return Response(status='404 Not Found')(env, start_response)
            return Response(status='201 Created')(env, start_response)
        if env['PATH_INFO'].startswith('/create_cont_fail/'):
            if env['REQUEST_METHOD'] == 'HEAD':
                return Response(status='403 Forbidden')(env, start_response)
            return Response(status='404 Not Found')(env, start_response)
        if env['PATH_INFO'].startswith('/create_obj_unauth/'):
            if env['PATH_INFO'].endswith('/cont'):
                return Response(status='201 Created')(env, start_response)
            return Response(status=401)(env, start_response)
        if env['PATH_INFO'].startswith('/tar_works/'):
            if len(env['PATH_INFO']) > self.max_pathlen:
                return Response(status='400 Bad Request')(env, start_response)
            return Response(status='201 Created')(env, start_response)
        if env['PATH_INFO'].startswith('/tar_works_cont_head_fail/'):
            if env['REQUEST_METHOD'] == 'HEAD':
                return Response(status='404 Not Found')(env, start_response)
            if len(env['PATH_INFO']) > 100:
                return Response(status='400 Bad Request')(env, start_response)
            return Response(status='201 Created')(env, start_response)
        if (env['PATH_INFO'].startswith('/delete_works/')
                and env['REQUEST_METHOD'] == 'DELETE'):
            self.delete_paths.append(env['PATH_INFO'])
            if len(env['PATH_INFO']) > self.max_pathlen:
                return Response(status='400 Bad Request')(env, start_response)
            if env['PATH_INFO'].endswith('404'):
                return Response(status='404 Not Found')(env, start_response)
            if env['PATH_INFO'].endswith('badutf8'):
                return Response(
                    status='412 Precondition Failed')(env, start_response)
            return Response(status='204 No Content')(env, start_response)
        if env['PATH_INFO'].startswith('/delete_cont_fail/'):
            return Response(status='409 Conflict')(env, start_response)
        if env['PATH_INFO'].startswith('/broke/'):
            return Response(status='500 Internal Error')(env, start_response)
        if env['PATH_INFO'].startswith('/delete_cont_success_after_attempts/'):
            if self.del_cont_cur_call < self.del_cont_total_calls:
                self.del_cont_cur_call += 1
                return Response(status='409 Conflict')(env, start_response)
            else:
                return Response(status='204 No Content')(env, start_response)


def build_dir_tree(start_path, tree_obj):
    if isinstance(tree_obj, list):
        for obj in tree_obj:
            build_dir_tree(start_path, obj)
    if isinstance(tree_obj, dict):
        for dir_name, obj in tree_obj.items():
            dir_path = os.path.join(start_path, dir_name)
            os.mkdir(dir_path)
            build_dir_tree(dir_path, obj)
    if isinstance(tree_obj, six.text_type):
        tree_obj = tree_obj.encode('utf8')
    if isinstance(tree_obj, str):
        obj_path = os.path.join(start_path, tree_obj)
        with open(obj_path, 'w+') as tree_file:
            tree_file.write('testing')


def build_tar_tree(tar, start_path, tree_obj, base_path=''):
    if isinstance(tree_obj, list):
        for obj in tree_obj:
            build_tar_tree(tar, start_path, obj, base_path=base_path)
    if isinstance(tree_obj, dict):
        for dir_name, obj in tree_obj.items():
            dir_path = os.path.join(start_path, dir_name)
            tar_info = tarfile.TarInfo(dir_path[len(base_path):])
            tar_info.type = tarfile.DIRTYPE
            tar.addfile(tar_info)
            build_tar_tree(tar, dir_path, obj, base_path=base_path)
    if isinstance(tree_obj, six.text_type):
        tree_obj = tree_obj.encode('utf8')
    if isinstance(tree_obj, str):
        obj_path = os.path.join(start_path, tree_obj)
        tar_info = tarfile.TarInfo('./' + obj_path[len(base_path):])
        tar.addfile(tar_info)


class TestUntarMetadata(unittest.TestCase):
    def setUp(self):
        self.app = FakeSwift()
        self.bulk = bulk.filter_factory({})(self.app)
        self.testdir = mkdtemp(suffix='tmp_test_bulk')

    def tearDown(self):
        rmtree(self.testdir, ignore_errors=1)

    def test_extract_metadata(self):
        self.app.register('HEAD', '/v1/a/c?extract-archive=tar',
                          HTTPNoContent, {}, None)
        self.app.register('PUT', '/v1/a/c/obj1?extract-archive=tar',
                          HTTPCreated, {}, None)
        self.app.register('PUT', '/v1/a/c/obj2?extract-archive=tar',
                          HTTPCreated, {}, None)

        # It's a real pain to instantiate TarInfo objects directly; they
        # really want to come from a file on disk or a tarball. So, we write
        # out some files and add pax headers to them as they get placed into
        # the tarball.
        with open(os.path.join(self.testdir, "obj1"), "w") as fh1:
            fh1.write("obj1 contents\n")
        with open(os.path.join(self.testdir, "obj2"), "w") as fh2:
            fh2.write("obj2 contents\n")

        tar_ball = BytesIO()
        tar_file = tarfile.TarFile.open(fileobj=tar_ball, mode="w",
                                        format=tarfile.PAX_FORMAT)

        # With GNU tar 1.27.1 or later (possibly 1.27 as well), a file with
        # extended attribute user.thingy = dingy gets put into the tarfile
        # with pax_headers containing key/value pair
        # (SCHILY.xattr.user.thingy, dingy), both unicode strings (py2: type
        # unicode, not type str).
        #
        # With BSD tar (libarchive), you get key/value pair
        # (LIBARCHIVE.xattr.user.thingy, dingy), which strikes me as
        # gratuitous incompatibility.
        #
        # Still, we'll support uploads with both. Just heap more code on the
        # problem until you can forget it's under there.
        with open(os.path.join(self.testdir, "obj1")) as fh1:
            tar_info1 = tar_file.gettarinfo(fileobj=fh1,
                                            arcname="obj1")
            tar_info1.pax_headers[u'SCHILY.xattr.user.mime_type'] = \
                u'application/food-diary'
            tar