##############################################################################
#
# Copyright (c) 2011 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################

import re
import unittest

from persistent._compat import copy_reg
from persistent._compat import PYPY
from persistent._compat import PYTHON3 as PY3
from persistent.tests.utils import skipIfNoCExtension

_is_pypy3 = PYPY and PY3

# pylint:disable=R0904,W0212,E1101
# pylint:disable=attribute-defined-outside-init,too-many-lines
# pylint:disable=blacklisted-name,useless-object-inheritance
# Hundreds of unused jar and OID vars make this useless
# pylint:disable=unused-variable


class _Persistent_Base(object):

    # py2/3 compat
    assertRaisesRegex = getattr(unittest.TestCase,
                                'assertRaisesRegex',
                                unittest.TestCase.assertRaisesRegexp)

    def _getTargetClass(self):
        # concrete testcase classes must override
        raise NotImplementedError()

    def _makeCache(self, jar):
        # concrete testcase classes must override
        raise NotImplementedError()

    def _makeRealCache(self, jar):
        return self._makeCache(jar)

    def _makeOne(self, *args, **kw):
        return self._getTargetClass()(*args, **kw)

    def _makeJar(self):
        from zope.interface import implementer
        from persistent.interfaces import IPersistentDataManager

        @implementer(IPersistentDataManager)
        class _Jar(object):
            _cache = None
            # Set this to a value to have our `setstate`
            # pass it through to the object's __setstate__
            setstate_calls_object = None

            # Set this to a value to have our `setstate`
            # set the _p_serial of the object
            setstate_sets_serial = None
            def __init__(self):
                self._loaded = []
                self._registered = []
            def setstate(self, obj):
                self._loaded.append(obj._p_oid)
                if self.setstate_calls_object is not None:
                    obj.__setstate__(self.setstate_calls_object)
                if self.setstate_sets_serial is not None:
                    obj._p_serial = self.setstate_sets_serial
            def register(self, obj):
                self._registered.append(obj._p_oid)

        jar = _Jar()
        jar._cache = self._makeCache(jar)
        return jar

    def _makeBrokenJar(self):
        from zope.interface import implementer
        from persistent.interfaces import IPersistentDataManager

        @implementer(IPersistentDataManager)
        class _BrokenJar(object):
            def __init__(self):
                self.called = 0
            def register(self, ob):
                self.called += 1
                raise NotImplementedError()
            def setstate(self, ob):
                raise NotImplementedError()

        jar = _BrokenJar()
        jar._cache = self._makeCache(jar)
        return jar

    def _makeOneWithJar(self, klass=None, broken_jar=False):
        OID = b'\x01' * 8
        if klass is not None:
            inst = klass()
        else:
            inst = self._makeOne()
        jar = self._makeJar() if not broken_jar else self._makeBrokenJar()
        jar._cache.new_ghost(OID, inst) # assigns _p_jar, _p_oid
        return inst, jar, OID

    def test_class_conforms_to_IPersistent(self):
        from zope.interface.verify import verifyClass
        from persistent.interfaces import IPersistent
        verifyClass(IPersistent, self._getTargetClass())

    def test_instance_conforms_to_IPersistent(self):
        from zope.interface.verify import verifyObject
        from persistent.interfaces import IPersistent
        verifyObject(IPersistent, self._makeOne())

    def test_instance_cannot_be_weakly_referenced(self):
        if PYPY: # pragma: no cover
            self.skipTest('On PyPy, everything can be weakly referenced')
        import weakref
        inst = self._makeOne()
        with self.assertRaises(TypeError):
            weakref.ref(inst)

    def test_ctor(self):
        from persistent.persistence import _INITIAL_SERIAL
        inst = self._makeOne()
        self.assertEqual(inst._p_jar, None)
        self.assertEqual(inst._p_oid, None)
        self.assertEqual(inst._p_serial, _INITIAL_SERIAL)
        self.assertEqual(inst._p_changed, False)
        self.assertEqual(inst._p_sticky, False)
        self.assertEqual(inst._p_status, 'unsaved')

    def test_del_jar_no_jar(self):
        inst = self._makeOne()
        del inst._p_jar  # does not raise
        self.assertEqual(inst._p_jar, None)

    def test_del_jar_while_in_cache(self):
        inst, _, OID = self._makeOneWithJar()
        def _test():
            del inst._p_jar
        self.assertRaises(ValueError, _test)

    def test_del_jar_like_ZODB_abort(self):
        # When a ZODB connection aborts, it removes registered objects from
        # the cache, deletes their jar, deletes their OID, and finally sets
        # p_changed to false
        inst, jar, OID = self._makeOneWithJar()
        del jar._cache[OID]
        del inst._p_jar
        self.assertEqual(inst._p_jar, None)

    def test_del_jar_of_inactive_object_that_has_no_state(self):
        # If an object is ghosted, and we try to delete its
        # jar, we shouldn't activate the object.

        # Simulate a POSKeyError on _p_activate; this can happen aborting
        # a transaction using ZEO
        broken_jar = self._makeBrokenJar()
        inst = self._makeOne()
        inst._p_oid = 42
        inst._p_jar = broken_jar

        # make it inactive
        inst._p_deactivate()
        self.assertEqual(inst._p_status, "ghost")

        # delete the jar; if we activated the object, the broken
        # jar would raise NotImplementedError
        del inst._p_jar

    def test_assign_p_jar_w_new_jar(self):
        inst, jar, OID = self._makeOneWithJar()
        new_jar = self._makeJar()

        with self.assertRaisesRegex(ValueError,
                                    "can not change _p_jar of cached object"):
            inst._p_jar = new_jar

    def test_assign_p_jar_w_valid_jar(self):
        jar = self._makeJar()
        inst = self._makeOne()
        inst._p_jar = jar
        self.assertEqual(inst._p_status, 'saved')
        self.assertTrue(inst._p_jar is jar)
        inst._p_jar = jar # reassign only to same DM

    def test_assign_p_jar_not_in_cache_allowed(self):
        jar = self._makeJar()
        inst = self._makeOne()
        inst._p_jar = jar
        # Both of these are allowed
        inst._p_jar = self._makeJar()
        inst._p_jar = None
        self.assertEqual(inst._p_jar, None)

    def test_assign_p_oid_w_invalid_oid(self):
        inst, jar, OID = self._makeOneWithJar()

        with self.assertRaisesRegex(ValueError,
                                    'can not change _p_oid of cached object'):
            inst._p_oid = object()

    def test_assign_p_oid_w_valid_oid(self):
        OID = b'\x01' * 8
        inst = self._makeOne()
        inst._p_oid = OID
        self.assertEqual(inst._p_oid, OID)
        inst._p_oid = OID  # reassign only same OID

    def test_assign_p_oid_w_new_oid_wo_jar(self):
        OID1 = b'\x01' * 8
        OID2 = b'\x02' * 8
        inst = self._makeOne()
        inst._p_oid = OID1
        inst._p_oid = OID2
        self.assertEqual(inst._p_oid, OID2)

    def test_assign_p_oid_w_None_wo_jar(self):
        OID1 = b'\x01' * 8
        inst = self._makeOne()
        inst._p_oid = OID1
        inst._p_oid = None
        self.assertEqual(inst._p_oid, None)

    def test_assign_p_oid_w_new_oid_w_jar(self):
        inst, jar, OID = self._makeOneWithJar()
        new_OID = b'\x02' * 8
        def _test():
            inst._p_oid = new_OID
        self.assertRaises(ValueError, _test)

    def test_assign_p_oid_not_in_cache_allowed(self):
        jar = self._makeJar()
        inst = self._makeOne()
        inst._p_jar = jar
        inst._p_oid = 1 # anything goes
        inst._p_oid = 42
        self.assertEqual(inst._p_oid, 42)

    def test_delete_p_oid_wo_jar(self):
        OID = b'\x01' * 8
        inst = self._makeOne()
        inst._p_oid = OID
        del inst._p_oid
        self.assertEqual(inst._p_oid, None)

    def test_delete_p_oid_w_jar(self):
        inst, jar, OID = self._makeOneWithJar()
        with self.assertRaises(ValueError):
            del inst._p_oid

    def test_delete_p_oid_of_subclass_calling_p_delattr(self):
        class P(self._getTargetClass()):
            def __delattr__(self, name):
                super(P, self)._p_delattr(name)
                raise AssertionError("Should not get here")

        inst, _jar, _oid = self._makeOneWithJar(klass=P)
        with self.assertRaises(ValueError):
            del inst._p_oid

    def test_del_oid_like_ZODB_abort(self):
        # When a ZODB connection aborts, it removes registered objects from
        # the cache, deletes their jar, deletes their OID, and finally sets
        # p_changed to false
        inst, jar, OID = self._makeOneWithJar()
        del jar._cache[OID]
        del inst._p_oid
        self.assertEqual(inst._p_oid, None)

    def test_assign_p_serial_w_invalid_type(self):
        inst = self._makeOne()
        def _test():
            inst._p_serial = object()
        self.assertRaises(ValueError, _test)

    def test_assign_p_serial_w_None(self):
        inst = self._makeOne()
        def _test():
            inst._p_serial = None
        self.assertRaises(ValueError, _test)

    def test_assign_p_serial_too_short(self):
        inst = self._makeOne()
        def _test():
            inst._p_serial = b'\x01\x02\x03'
        self.assertRaises(ValueError, _test)

    def test_assign_p_serial_too_long(self):
        inst = self._makeOne()
        def _test():
            inst._p_serial = b'\x01\x02\x03' * 3
        self.assertRaises(ValueError, _test)

    def test_assign_p_serial_w_valid_serial(self):
        SERIAL = b'\x01' * 8
        inst = self._makeOne()
        inst._p_serial = SERIAL
        self.assertEqual(inst._p_serial, SERIAL)

    def test_delete_p_serial(self):
        from persistent.persistence import _INITIAL_SERIAL
        SERIAL = b'\x01' * 8
        inst = self._makeOne()
        inst._p_serial = SERIAL
        self.assertEqual(inst._p_serial, SERIAL)
        del inst._p_serial
        self.assertEqual(inst._p_serial, _INITIAL_SERIAL)

    def test_query_p_changed_unsaved(self):
        inst = self._makeOne()
        self.assertEqual(inst._p_changed, False)

    def test_query_p_changed_ghost(self):
        inst, jar, OID = self._makeOneWithJar()
        inst._p_deactivate()
        self.assertEqual(inst._p_changed, None)

    def test_query_p_changed_saved(self):
        inst, jar, OID = self._makeOneWithJar()
        inst._p_activate()
        self.assertEqual(inst._p_changed, False)

    def test_query_p_changed_changed(self):
        inst, jar, OID = self._makeOneWithJar()
        inst._p_activate()
        inst._p_changed = True
        self.assertEqual(inst._p_changed, True)

    def test_assign_p_changed_none_from_unsaved(self):
        inst = self._makeOne()
        inst._p_changed = None
        self.assertEqual(inst._p_status, 'unsaved')

    def test_assign_p_changed_true_from_unsaved(self):
        inst = self._makeOne()
        inst._p_changed = True
        self.assertEqual(inst._p_status, 'unsaved')