a b @sxddlZddlZddlZddlmZeeZdddZGdddZ GdddZ d d Z d d Z d dZ ddZdS)N)RecordModifiedErrorcsfdd}|S)aA decorator which can be used to mark functions as deprecated. replacement is a callable that will be called with the same args as the decorated function. >>> import pytest >>> @deprecated() ... def foo1(x): ... return x ... >>> pytest.warns(DeprecationWarning, foo1, 1) 1 >>> def newfun(x): ... return 0 ... >>> @deprecated(newfun, '1.1') ... def foo2(x): ... return x ... >>> pytest.warns(DeprecationWarning, foo2, 1) 0 >>> csfdd}|S)Ncsjdj}dur|d7}dur2|d7}tj|tddtrX|i|S|i|SdS)Nz%s is deprecatedzwill be removed in version %s;z; use %s instead) stacklevel)__name__warningswarnDeprecationWarningcallable)argskwargsmsg)oldfun replacementversionN/home/httpd/docs/test/DocsMgr/lib/python3.9/site-packages/concurrency/utils.pyinner"s   z(deprecated..outer..innerr)r rrr)r router!s zdeprecated..outerr)rrrrrr deprecated src@s8eZdZdZdZiZddZddZddZd d Z dS) ConcurrencyTestMixina Mixin class to test Models that use `VersionField` this class offer a simple test scenario. Its purpose is to discover some conflict in the `save()` inheritance:: from concurrency.utils import ConcurrencyTestMixin from myproject.models import MyModel class MyModelTest(ConcurrencyTestMixin, TestCase): concurrency_model = TestModel0 concurrency_kwargs = {'username': 'test'} NcKs,t|j}|||jjjfi|dS)Nr)dictconcurrency_kwargsupdateconcurrency_modelobjectsZ get_or_create)selfr r rrr_get_concurrency_targetFs  z,ConcurrencyTestMixin._get_concurrency_targetcCspddlm}|}|}||}||}||ksHJd||f||jdus^J|t|jdS)Nrz+got same row with different version (%s/%s))concurrency.apiapirget_revision_of_objectZsavepk assertRaisesr)rrtargetZ target_copyZv1Zv2rrrtest_concurrency_conflictMs   z.ConcurrencyTestMixin.test_concurrency_conflictcCs6ddlm}|}||}|t|d|dS)Nrzversion is not null %s)rrrr assertFalsebool)rrr#rrrrtest_concurrency_safetyYs  z,ConcurrencyTestMixin.test_concurrency_safetycCsD|j}|t|dd|j|jj}|||jjvd|jdS)N_concurrencymetaz&%s is not under concurrency managementz$%s: version field not in meta.fields)r assertTruehasattrr(fieldZ_metafields)rr#Zrevision_fieldrrrtest_concurrency_management`s z0ConcurrencyTestMixin.test_concurrency_management) r __module__ __qualname____doc__rrrr$r'r-rrrrr3s rc@s eZdZdS)ConcurrencyAdminTestMixinN)rr.r/rrrrr1ksr1cCs|jjj|jdS)z3 Reload model instance from the database # )r!) __class__rgetr!)Zmodel_instancerrrrefetchosr4cCsJt|r|}nt|r|}n|j}z|jWStyD|jYS0dS)zN Returns the classname of an object r a class :param o: :return: N)inspectisclassr r2r/AttributeErrorr)or#rrr get_classnamevs  r9cCsTg}t|dr*||j|t|nt|r:|jS|sJtd|d|S)a'Returns the fully qualified class name of an object or a class :param o: object or class :return: class name >>> import concurrency.fields >>> fqn('str') Traceback (most recent call last): ... ValueError: Invalid argument `str` >>> class A: ... def method(self): ... pass >>> str(fqn(A)) 'concurrency.utils.A' >>> str(fqn(A())) 'concurrency.utils.A' >>> str(fqn(concurrency.fields)) 'concurrency.fields' >>> str(fqn(A.method)) 'concurrency.utils.A.method' r.zInvalid argument `%s`.) r*appendr.r9r5ismoduler ValueErrorjoin)r8partsrrrfqns    r@cCsFt}|D]2}t|dr2t|ts2|t|q ||q t|S)a  flatten(sequence) -> list Returns a single, flat list which contains all elements retrieved from the sequence and all recursively contained sub-sequences (iterables). :param sequence: any object that implements iterable protocol (see: :ref:`typeiter`) :return: list Examples: >>> from adminactions.utils import flatten >>> [1, 2, [3,4], (5,6)] [1, 2, [3, 4], (5, 6)] >>> flatten([[[1,2,3], (42,None)], [4,5], [6], 7, (8,9,10)]) [1, 2, 3, 42, None, 4, 5, 6, 7, 8, 9, 10]__iter__)listr* isinstancestrextendflattenr;)iterableresultelrrrrFs  rF)NN)r5loggingrZconcurrency.exceptionsr getLoggerrloggerrrr1r4r9r@rFrrrrs   )81