# -*- coding: utf-8 -*-
# Copyright (c) 2012, Google, Inc.
# All rights reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish, dis-
# tribute, sublicense, and/or sell copies of the Software, and to permit
# persons to whom the Software is furnished to do so, subject to the fol-
# lowing conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.

"""Integration tests for GS versioning support."""

from xml import sax

from boto import handler
from boto.gs import acl
from tests.integration.gs.testcase import GSTestCase


class GSVersioningTest(GSTestCase):

    def testVersioningToggle(self):
        b = self._MakeBucket()
        self.assertFalse(b.get_versioning_status())
        b.configure_versioning(True)
        self.assertTrue(b.get_versioning_status())
        b.configure_versioning(False)
        self.assertFalse(b.get_versioning_status())

    def testDeleteVersionedKey(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation

        s2 = "test2"
        k.set_contents_from_string(s2)
        k = b.get_key("foo")
        g2 = k.generation

        versions = list(b.list_versions())
        self.assertEqual(len(versions), 2)
        self.assertEqual(versions[0].name, "foo")
        self.assertEqual(versions[1].name, "foo")
        generations = [k.generation for k in versions]
        self.assertIn(g1, generations)
        self.assertIn(g2, generations)

        # Delete "current" version and make sure that version is no longer
        # visible from a basic GET call.
        b.delete_key("foo", generation=None)
        self.assertIsNone(b.get_key("foo"))

        # Both old versions should still be there when listed using the versions
        # query parameter.
        versions = list(b.list_versions())
        self.assertEqual(len(versions), 2)
        self.assertEqual(versions[0].name, "foo")
        self.assertEqual(versions[1].name, "foo")
        generations = [k.generation for k in versions]
        self.assertIn(g1, generations)
        self.assertIn(g2, generations)

        # Delete generation 2 and make sure it's gone.
        b.delete_key("foo", generation=g2)
        versions = list(b.list_versions())
        self.assertEqual(len(versions), 1)
        self.assertEqual(versions[0].name, "foo")
        self.assertEqual(versions[0].generation, g1)

        # Delete generation 1 and make sure it's gone.
        b.delete_key("foo", generation=g1)
        versions = list(b.list_versions())
        self.assertEqual(len(versions), 0)

    def testGetVersionedKey(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation
        o1 = k.get_contents_as_string()
        self.assertEqual(o1, s1)

        s2 = "test2"
        k.set_contents_from_string(s2)
        k = b.get_key("foo")
        g2 = k.generation
        self.assertNotEqual(g2, g1)
        o2 = k.get_contents_as_string()
        self.assertEqual(o2, s2)

        k = b.get_key("foo", generation=g1)
        self.assertEqual(k.get_contents_as_string(), s1)
        k = b.get_key("foo", generation=g2)
        self.assertEqual(k.get_contents_as_string(), s2)

    def testVersionedBucketCannedAcl(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation

        s2 = "test2"
        k.set_contents_from_string(s2)
        k = b.get_key("foo")
        g2 = k.generation

        acl1g1 = b.get_acl("foo", generation=g1)
        acl1g2 = b.get_acl("foo", generation=g2)
        owner1g1 = acl1g1.owner.id
        owner1g2 = acl1g2.owner.id
        self.assertEqual(owner1g1, owner1g2)
        entries1g1 = acl1g1.entries.entry_list
        entries1g2 = acl1g2.entries.entry_list
        self.assertEqual(len(entries1g1), len(entries1g2))

        b.set_acl("public-read", key_name="foo", generation=g1)

        acl2g1 = b.get_acl("foo", generation=g1)
        acl2g2 = b.get_acl("foo", generation=g2)
        entries2g1 = acl2g1.entries.entry_list
        entries2g2 = acl2g2.entries.entry_list
        self.assertEqual(len(entries2g2), len(entries1g2))
        public_read_entries1 = [e for e in entries2g1 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        public_read_entries2 = [e for e in entries2g2 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        self.assertEqual(len(public_read_entries1), 1)
        self.assertEqual(len(public_read_entries2), 0)

    def testVersionedBucketXmlAcl(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation

        s2 = "test2"
        k.set_contents_from_string(s2)
        k = b.get_key("foo")
        g2 = k.generation

        acl1g1 = b.get_acl("foo", generation=g1)
        acl1g2 = b.get_acl("foo", generation=g2)
        owner1g1 = acl1g1.owner.id
        owner1g2 = acl1g2.owner.id
        self.assertEqual(owner1g1, owner1g2)
        entries1g1 = acl1g1.entries.entry_list
        entries1g2 = acl1g2.entries.entry_list
        self.assertEqual(len(entries1g1), len(entries1g2))

        acl_xml = (
            '<ACCESSControlList><EntrIes><Entry>'    +
            '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' +
            '</Entry></EntrIes></ACCESSControlList>')
        aclo = acl.ACL()
        h = handler.XmlHandler(aclo, b)
        sax.parseString(acl_xml, h)

        b.set_acl(aclo, key_name="foo", generation=g1)

        acl2g1 = b.get_acl("foo", generation=g1)
        acl2g2 = b.get_acl("foo", generation=g2)
        entries2g1 = acl2g1.entries.entry_list
        entries2g2 = acl2g2.entries.entry_list
        self.assertEqual(len(entries2g2), len(entries1g2))
        public_read_entries1 = [e for e in entries2g1 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        public_read_entries2 = [e for e in entries2g2 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        self.assertEqual(len(public_read_entries1), 1)
        self.assertEqual(len(public_read_entries2), 0)

    def testVersionedObjectCannedAcl(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation

        s2 = "test2"
        k.set_contents_from_string(s2)
        k = b.get_key("foo")
        g2 = k.generation

        acl1g1 = b.get_acl("foo", generation=g1)
        acl1g2 = b.get_acl("foo", generation=g2)
        owner1g1 = acl1g1.owner.id
        owner1g2 = acl1g2.owner.id
        self.assertEqual(owner1g1, owner1g2)
        entries1g1 = acl1g1.entries.entry_list
        entries1g2 = acl1g2.entries.entry_list
        self.assertEqual(len(entries1g1), len(entries1g2))

        b.set_acl("public-read", key_name="foo", generation=g1)

        acl2g1 = b.get_acl("foo", generation=g1)
        acl2g2 = b.get_acl("foo", generation=g2)
        entries2g1 = acl2g1.entries.entry_list
        entries2g2 = acl2g2.entries.entry_list
        self.assertEqual(len(entries2g2), len(entries1g2))
        public_read_entries1 = [e for e in entries2g1 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        public_read_entries2 = [e for e in entries2g2 if e.permission == "READ"
                                and e.scope.type == acl.ALL_USERS]
        self.assertEqual(len(public_read_entries1), 1)
        self.assertEqual(len(public_read_entries2), 0)

    def testCopyVersionedKey(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        s1 = "test1"
        k.set_contents_from_string(s1)

        k = b.get_key("foo")
        g1 = k.generation

        s2 = "test2"
        k.set_contents_from_string(s2)

        b2 = self._MakeVersionedBucket()
        b2.copy_key("foo2", b.name, "foo", src_generation=g1)

        k2 = b2.get_key("foo2")
        s3 = k2.get_contents_as_string()
        self.assertEqual(s3, s1)

    def testKeyGenerationUpdatesOnSet(self):
        b = self._MakeVersionedBucket()
        k = b.new_key("foo")
        self.assertIsNone(k.generation)
        k.set_contents_from_string("test1")
        g1 = k.generation
        self.assertRegexpMatches(g1, r'[0-9]+')
        self.assertEqual(k.metageneration, '1')
        k.set_contents_from_string("test2")
        g2 = k.generation
        self.assertNotEqual(g1, g2)
        self.assertRegexpMatches(g2, r'[0-9]+')
        self.assertGreater(int(g2), int(g1))
        self.assertEqual(k.metageneration, '1')