How to use provides method in avocado

Best Python code snippet using avocado_python

test_declarations.py

Source:test_declarations.py

...847class Test_alsoProvides(unittest.TestCase):848    def _callFUT(self, *args, **kw):849        from zope.interface.declarations import alsoProvides850        return alsoProvides(*args, **kw)851    def test_wo_existing_provides(self):852        from zope.interface.declarations import ProvidesClass853        from zope.interface.interface import InterfaceClass854        IFoo = InterfaceClass("IFoo")855        class Foo(object):856            pass857        obj = Foo()858        self._callFUT(obj, IFoo)859        self.assertTrue(isinstance(obj.__provides__, ProvidesClass))860        self.assertEqual(list(obj.__provides__), [IFoo])861    def test_w_existing_provides(self):862        from zope.interface.declarations import directlyProvides863        from zope.interface.declarations import ProvidesClass864        from zope.interface.interface import InterfaceClass865        IFoo = InterfaceClass("IFoo")866        IBar = InterfaceClass("IBar")867        class Foo(object):868            pass869        obj = Foo()870        directlyProvides(obj, IFoo)871        self._callFUT(obj, IBar)872        self.assertTrue(isinstance(obj.__provides__, ProvidesClass))873        self.assertEqual(list(obj.__provides__), [IFoo, IBar])874class Test_noLongerProvides(unittest.TestCase):875    def _callFUT(self, *args, **kw):876        from zope.interface.declarations import noLongerProvides877        return noLongerProvides(*args, **kw)878    def test_wo_existing_provides(self):879        from zope.interface.interface import InterfaceClass880        IFoo = InterfaceClass("IFoo")881        class Foo(object):882            pass883        obj = Foo()884        self._callFUT(obj, IFoo)885        self.assertEqual(list(obj.__provides__), [])886    def test_w_existing_provides_hit(self):887        from zope.interface.declarations import directlyProvides888        from zope.interface.interface import InterfaceClass889        IFoo = InterfaceClass("IFoo")890        class Foo(object):891            pass892        obj = Foo()893        directlyProvides(obj, IFoo)894        self._callFUT(obj, IFoo)895        self.assertEqual(list(obj.__provides__), [])896    def test_w_existing_provides_miss(self):897        from zope.interface.declarations import directlyProvides898        from zope.interface.interface import InterfaceClass899        IFoo = InterfaceClass("IFoo")900        IBar = InterfaceClass("IBar")901        class Foo(object):902            pass903        obj = Foo()904        directlyProvides(obj, IFoo)905        self._callFUT(obj, IBar)906        self.assertEqual(list(obj.__provides__), [IFoo])907    def test_w_iface_implemented_by_class(self):908        from zope.interface.declarations import implementer909        from zope.interface.interface import InterfaceClass910        IFoo = InterfaceClass("IFoo")911        @implementer(IFoo)912        class Foo(object):913            pass914        obj = Foo()915        self.assertRaises(ValueError, self._callFUT, obj, IFoo)916class ClassProvidesBaseFallbackTests(unittest.TestCase):917    def _getTargetClass(self):918        from zope.interface.declarations import ClassProvidesBaseFallback919        return ClassProvidesBaseFallback920    def _makeOne(self, klass, implements):921        # Don't instantiate directly:  the C version can't have attributes922        # assigned.923        class Derived(self._getTargetClass()):924            def __init__(self, k, i):925                self._cls = k926                self._implements = i927        return Derived(klass, implements)928    def test_w_same_class_via_class(self):929        from zope.interface.interface import InterfaceClass930        IFoo = InterfaceClass("IFoo")931        class Foo(object):932            pass933        cpbp = Foo.__provides__ = self._makeOne(Foo, IFoo)934        self.assertTrue(Foo.__provides__ is cpbp)935    def test_w_same_class_via_instance(self):936        from zope.interface.interface import InterfaceClass937        IFoo = InterfaceClass("IFoo")938        class Foo(object):939            pass940        foo = Foo()941        cpbp = Foo.__provides__ = self._makeOne(Foo, IFoo)942        self.assertTrue(foo.__provides__ is IFoo)943    def test_w_different_class(self):944        from zope.interface.interface import InterfaceClass945        IFoo = InterfaceClass("IFoo")946        class Foo(object):947            pass948        class Bar(Foo):949            pass950        bar = Bar()951        cpbp = Foo.__provides__ = self._makeOne(Foo, IFoo)952        self.assertRaises(AttributeError, getattr, Bar, '__provides__')953        self.assertRaises(AttributeError, getattr, bar, '__provides__')954class ClassProvidesBaseTests(ClassProvidesBaseFallbackTests):955    # Repeat tests for C optimizations956    def _getTargetClass(self):957        from zope.interface.declarations import ClassProvidesBase958        return ClassProvidesBase959class ClassProvidesTests(unittest.TestCase):960    def _getTargetClass(self):961        from zope.interface.declarations import ClassProvides962        return ClassProvides963    def _makeOne(self, *args, **kw):964        return self._getTargetClass()(*args, **kw)965    def test_w_simple_metaclass(self):966        from zope.interface.declarations import implementer967        from zope.interface.interface import InterfaceClass968        IFoo = InterfaceClass("IFoo")969        IBar = InterfaceClass("IBar")970        @implementer(IFoo)971        class Foo(object):972            pass973        cp = Foo.__provides__ = self._makeOne(Foo, type(Foo), IBar)974        self.assertTrue(Foo.__provides__ is cp)975        self.assertEqual(list(Foo().__provides__), [IFoo])976    def test___reduce__(self):977        from zope.interface.declarations import implementer978        from zope.interface.interface import InterfaceClass979        IFoo = InterfaceClass("IFoo")980        IBar = InterfaceClass("IBar")981        @implementer(IFoo)982        class Foo(object):983            pass984        cp = Foo.__provides__ = self._makeOne(Foo, type(Foo), IBar)985        self.assertEqual(cp.__reduce__(),986                         (self._getTargetClass(), (Foo, type(Foo), IBar)))987class Test_directlyProvidedBy(unittest.TestCase):988    def _callFUT(self, *args, **kw):989        from zope.interface.declarations import directlyProvidedBy990        return directlyProvidedBy(*args, **kw)991    def test_wo_declarations_in_class_or_instance(self):992        class Foo(object):993            pass994        foo = Foo()995        self.assertEqual(list(self._callFUT(foo)), [])996    def test_w_declarations_in_class_but_not_instance(self):997        from zope.interface.declarations import implementer998        from zope.interface.interface import InterfaceClass999        IFoo = InterfaceClass("IFoo")1000        @implementer(IFoo)1001        class Foo(object):1002            pass1003        foo = Foo()1004        self.assertEqual(list(self._callFUT(foo)), [])1005    def test_w_declarations_in_instance_but_not_class(self):1006        from zope.interface.declarations import directlyProvides1007        from zope.interface.interface import InterfaceClass1008        IFoo = InterfaceClass("IFoo")1009        class Foo(object):1010            pass1011        foo = Foo()1012        directlyProvides(foo, IFoo)1013        self.assertEqual(list(self._callFUT(foo)), [IFoo])1014    def test_w_declarations_in_instance_and_class(self):1015        from zope.interface.declarations import directlyProvides1016        from zope.interface.declarations import implementer1017        from zope.interface.interface import InterfaceClass1018        IFoo = InterfaceClass("IFoo")1019        IBar = InterfaceClass("IBar")1020        @implementer(IFoo)1021        class Foo(object):1022            pass1023        foo = Foo()1024        directlyProvides(foo, IBar)1025        self.assertEqual(list(self._callFUT(foo)), [IBar])1026class Test_classProvides(unittest.TestCase, _Py3ClassAdvice):1027    def _getFUT(self):1028        from zope.interface.declarations import classProvides1029        return classProvides1030    def test_called_from_function(self):1031        import warnings1032        from zope.interface.declarations import classProvides1033        from zope.interface.interface import InterfaceClass1034        from zope.interface._compat import PYTHON31035        IFoo = InterfaceClass("IFoo")1036        globs = {'classProvides': classProvides, 'IFoo': IFoo}1037        locs = {}1038        CODE = "\n".join([1039            'def foo():',1040            '    classProvides(IFoo)'1041            ])1042        exec(CODE, globs, locs)1043        foo = locs['foo']1044        with warnings.catch_warnings(record=True) as log:1045            warnings.resetwarnings()1046            self.assertRaises(TypeError, foo)1047            if not PYTHON3:1048                self.assertEqual(len(log), 0) # no longer warn1049    def test_called_twice_from_class(self):1050        import warnings1051        from zope.interface.declarations import classProvides1052        from zope.interface.interface import InterfaceClass1053        from zope.interface._compat import PYTHON31054        IFoo = InterfaceClass("IFoo")1055        IBar = InterfaceClass("IBar")1056        globs = {'classProvides': classProvides, 'IFoo': IFoo, 'IBar': IBar}1057        locs = {}1058        CODE = "\n".join([1059            'class Foo(object):',1060            '    classProvides(IFoo)',1061            '    classProvides(IBar)',1062            ])1063        with warnings.catch_warnings(record=True) as log:1064            warnings.resetwarnings()1065            try:1066                exec(CODE, globs, locs)1067            except TypeError:1068                if not PYTHON3:1069                    self.assertEqual(len(log), 0) # no longer warn1070            else:1071                self.fail("Didn't raise TypeError")1072    def test_called_once_from_class(self):1073        from zope.interface.declarations import classProvides1074        from zope.interface.interface import InterfaceClass1075        IFoo = InterfaceClass("IFoo")1076        globs = {'classProvides': classProvides, 'IFoo': IFoo}1077        locs = {}1078        CODE = "\n".join([1079            'class Foo(object):',1080            '    classProvides(IFoo)',1081            ])1082        if self._run_generated_code(CODE, globs, locs):1083            Foo = locs['Foo']1084            spec = Foo.__providedBy__1085            self.assertEqual(list(spec), [IFoo])1086# Test _classProvides_advice through classProvides, its only caller.1087class Test_provider(unittest.TestCase):1088    def _getTargetClass(self):1089        from zope.interface.declarations import provider1090        return provider1091    def _makeOne(self, *args, **kw):1092        return self._getTargetClass()(*args, **kw)1093    def test_w_class(self):1094        from zope.interface.declarations import ClassProvides1095        from zope.interface.interface import InterfaceClass1096        IFoo = InterfaceClass("IFoo")1097        @self._makeOne(IFoo)1098        class Foo(object):1099            pass1100        self.assertTrue(isinstance(Foo.__provides__, ClassProvides))1101        self.assertEqual(list(Foo.__provides__), [IFoo])1102class Test_moduleProvides(unittest.TestCase):1103    def _getFUT(self):1104        from zope.interface.declarations import moduleProvides1105        return moduleProvides1106    def test_called_from_function(self):1107        from zope.interface.declarations import moduleProvides1108        from zope.interface.interface import InterfaceClass1109        IFoo = InterfaceClass("IFoo")1110        globs = {'__name__': 'zope.interface.tests.foo',1111                 'moduleProvides': moduleProvides, 'IFoo': IFoo}1112        locs = {}1113        CODE = "\n".join([1114            'def foo():',1115            '    moduleProvides(IFoo)'1116            ])1117        exec(CODE, globs, locs)1118        foo = locs['foo']1119        self.assertRaises(TypeError, foo)1120    def test_called_from_class(self):1121        from zope.interface.declarations import moduleProvides1122        from zope.interface.interface import InterfaceClass1123        IFoo = InterfaceClass("IFoo")1124        globs = {'__name__': 'zope.interface.tests.foo',1125                 'moduleProvides': moduleProvides, 'IFoo': IFoo}1126        locs = {}1127        CODE = "\n".join([1128            'class Foo(object):',1129            '    moduleProvides(IFoo)',1130            ])1131        try:1132            exec(CODE, globs, locs)1133        except TypeError:1134            pass1135        else:1136            assert False, 'TypeError not raised'1137    def test_called_once_from_module_scope(self):1138        from zope.interface.declarations import moduleProvides1139        from zope.interface.interface import InterfaceClass1140        IFoo = InterfaceClass("IFoo")1141        globs = {'__name__': 'zope.interface.tests.foo',1142                 'moduleProvides': moduleProvides, 'IFoo': IFoo}1143        CODE = "\n".join([1144            'moduleProvides(IFoo)',1145            ])1146        exec(CODE, globs)1147        spec = globs['__provides__']1148        self.assertEqual(list(spec), [IFoo])1149    def test_called_twice_from_module_scope(self):1150        from zope.interface.declarations import moduleProvides1151        from zope.interface.interface import InterfaceClass1152        IFoo = InterfaceClass("IFoo")1153        globs = {'__name__': 'zope.interface.tests.foo',1154                 'moduleProvides': moduleProvides, 'IFoo': IFoo}1155        locs = {}1156        CODE = "\n".join([1157            'moduleProvides(IFoo)',1158            'moduleProvides(IFoo)',1159            ])1160        try:1161            exec(CODE, globs)1162        except TypeError:1163            pass1164        else:1165            assert False, 'TypeError not raised'1166class Test_getObjectSpecificationFallback(unittest.TestCase):1167    def _callFUT(self, *args, **kw):1168        from zope.interface.declarations import getObjectSpecificationFallback1169        return getObjectSpecificationFallback(*args, **kw)1170    def test_wo_existing_provides_classless(self):1171        the_dict = {}1172        class Foo(object):1173            def __getattribute__(self, name):1174                # Emulate object w/o any class1175                if name == '__class__':1176                    raise AttributeError(name)1177                try:1178                    return the_dict[name]1179                except KeyError:1180                    raise AttributeError(name)1181            def __setattr__(self, name, value):1182                the_dict[name] = value1183        foo = Foo()1184        spec = self._callFUT(foo)1185        self.assertEqual(list(spec), [])1186    def test_existing_provides_is_spec(self):1187        from zope.interface.declarations import directlyProvides1188        from zope.interface.interface import InterfaceClass1189        IFoo = InterfaceClass("IFoo")1190        def foo():1191            pass1192        directlyProvides(foo, IFoo)1193        spec = self._callFUT(foo)1194        self.assertTrue(spec is foo.__provides__)1195    def test_existing_provides_is_not_spec(self):1196        def foo():1197            pass1198        foo.__provides__ = object() # not a valid spec1199        spec = self._callFUT(foo)1200        self.assertEqual(list(spec), [])1201    def test_existing_provides(self):1202        from zope.interface.declarations import directlyProvides1203        from zope.interface.interface import InterfaceClass1204        IFoo = InterfaceClass("IFoo")1205        class Foo(object):1206            pass1207        foo = Foo()1208        directlyProvides(foo, IFoo)1209        spec = self._callFUT(foo)1210        self.assertEqual(list(spec), [IFoo])1211    def test_wo_provides_on_class_w_implements(self):1212        from zope.interface.declarations import implementer1213        from zope.interface.interface import InterfaceClass1214        IFoo = InterfaceClass("IFoo")1215        @implementer(IFoo)1216        class Foo(object):1217            pass1218        foo = Foo()1219        spec = self._callFUT(foo)1220        self.assertEqual(list(spec), [IFoo])1221    def test_wo_provides_on_class_wo_implements(self):1222        class Foo(object):1223            pass1224        foo = Foo()1225        spec = self._callFUT(foo)1226        self.assertEqual(list(spec), [])1227class Test_getObjectSpecification(Test_getObjectSpecificationFallback):1228    # Repeat tests for C optimizations1229    def _callFUT(self, *args, **kw):1230        from zope.interface.declarations import getObjectSpecification1231        return getObjectSpecification(*args, **kw)1232class Test_providedByFallback(unittest.TestCase):1233    def _callFUT(self, *args, **kw):1234        from zope.interface.declarations import providedByFallback1235        return providedByFallback(*args, **kw)1236    def test_wo_providedBy_on_class_wo_implements(self):1237        class Foo(object):1238            pass1239        foo = Foo()1240        spec = self._callFUT(foo)1241        self.assertEqual(list(spec), [])1242    def test_w_providedBy_valid_spec(self):1243        from zope.interface.declarations import Provides1244        from zope.interface.interface import InterfaceClass1245        IFoo = InterfaceClass("IFoo")1246        class Foo(object):1247            pass1248        foo = Foo()1249        foo.__providedBy__ = Provides(Foo, IFoo)1250        spec = self._callFUT(foo)1251        self.assertEqual(list(spec), [IFoo])1252    def test_w_providedBy_invalid_spec(self):1253        class Foo(object):1254            pass1255        foo = Foo()1256        foo.__providedBy__ = object()1257        spec = self._callFUT(foo)1258        self.assertEqual(list(spec), [])1259    def test_w_providedBy_invalid_spec_class_w_implements(self):1260        from zope.interface.declarations import implementer1261        from zope.interface.interface import InterfaceClass1262        IFoo = InterfaceClass("IFoo")1263        @implementer(IFoo)1264        class Foo(object):1265            pass1266        foo = Foo()1267        foo.__providedBy__ = object()1268        spec = self._callFUT(foo)1269        self.assertEqual(list(spec), [IFoo])1270    def test_w_providedBy_invalid_spec_w_provides_no_provides_on_class(self):1271        class Foo(object):1272            pass1273        foo = Foo()1274        foo.__providedBy__ = object()1275        expected = foo.__provides__ = object()1276        spec = self._callFUT(foo)1277        self.assertTrue(spec is expected)1278    def test_w_providedBy_invalid_spec_w_provides_diff_provides_on_class(self):1279        class Foo(object):1280            pass1281        foo = Foo()1282        foo.__providedBy__ = object()1283        expected = foo.__provides__ = object()1284        Foo.__provides__ = object()1285        spec = self._callFUT(foo)1286        self.assertTrue(spec is expected)1287    def test_w_providedBy_invalid_spec_w_provides_same_provides_on_class(self):1288        from zope.interface.declarations import implementer1289        from zope.interface.interface import InterfaceClass1290        IFoo = InterfaceClass("IFoo")1291        @implementer(IFoo)1292        class Foo(object):1293            pass1294        foo = Foo()1295        foo.__providedBy__ = object()1296        foo.__provides__ = Foo.__provides__ = object()1297        spec = self._callFUT(foo)1298        self.assertEqual(list(spec), [IFoo])1299class Test_providedBy(Test_providedByFallback):1300    # Repeat tests for C optimizations1301    def _callFUT(self, *args, **kw):1302        from zope.interface.declarations import providedBy1303        return providedBy(*args, **kw)1304class ObjectSpecificationDescriptorFallbackTests(unittest.TestCase):1305    def _getTargetClass(self):1306        from zope.interface.declarations \1307            import ObjectSpecificationDescriptorFallback1308        return ObjectSpecificationDescriptorFallback1309    def _makeOne(self, *args, **kw):1310        return self._getTargetClass()(*args, **kw)1311    def test_accessed_via_class(self):1312        from zope.interface.declarations import Provides1313        from zope.interface.interface import InterfaceClass1314        IFoo = InterfaceClass("IFoo")1315        class Foo(object):1316            pass1317        Foo.__provides__ = Provides(Foo, IFoo)1318        Foo.__providedBy__ = self._makeOne()1319        self.assertEqual(list(Foo.__providedBy__), [IFoo])1320    def test_accessed_via_inst_wo_provides(self):1321        from zope.interface.declarations import implementer1322        from zope.interface.declarations import Provides1323        from zope.interface.interface import InterfaceClass1324        IFoo = InterfaceClass("IFoo")1325        IBar = InterfaceClass("IBar")1326        @implementer(IFoo)1327        class Foo(object):1328            pass1329        Foo.__provides__ = Provides(Foo, IBar)1330        Foo.__providedBy__ = self._makeOne()1331        foo = Foo()1332        self.assertEqual(list(foo.__providedBy__), [IFoo])1333    def test_accessed_via_inst_w_provides(self):1334        from zope.interface.declarations import directlyProvides1335        from zope.interface.declarations import implementer1336        from zope.interface.declarations import Provides1337        from zope.interface.interface import InterfaceClass1338        IFoo = InterfaceClass("IFoo")1339        IBar = InterfaceClass("IBar")1340        IBaz = InterfaceClass("IBaz")1341        @implementer(IFoo)1342        class Foo(object):1343            pass1344        Foo.__provides__ = Provides(Foo, IBar)1345        Foo.__providedBy__ = self._makeOne()1346        foo = Foo()1347        directlyProvides(foo, IBaz)...

declarations.py

Source:declarations.py

1##############################################################################2# Copyright (c) 2003 Zope Corporation and Contributors.3# All Rights Reserved.4#5# This software is subject to the provisions of the Zope Public License,6# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.7# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED8# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED9# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS10# FOR A PARTICULAR PURPOSE.11##############################################################################12"""Implementation of interface declarations13There are three flavors of declarations:14  - Declarations are used to simply name declared interfaces.15  - ImplementsDeclarations are used to express the interfaces that a16    class implements (that instances of the class provides).17    Implements specifications support inheriting interfaces.18  - ProvidesDeclarations are used to express interfaces directly19    provided by objects.20$Id: declarations.py 72236 2007-01-26 16:50:29Z alga $21"""22__docformat__ = 'restructuredtext'23import sys24import weakref25from zope.interface.interface import InterfaceClass, Specification26from ro import mergeOrderings, ro27import exceptions28from types import ClassType, ModuleType29from zope.interface.advice import addClassAdvisor30# Registry of class-implementation specifications31BuiltinImplementationSpecifications = {}32class Declaration(Specification):33    """Interface declarations"""34    def __init__(self, *interfaces):35        Specification.__init__(self, _normalizeargs(interfaces))36    def changed(self, originally_changed):37        Specification.changed(self, originally_changed)38        try:39            del self._v_attrs40        except AttributeError:41            pass42    def __contains__(self, interface):43        """Test whether an interface is in the specification44        for example:45          >>> from zope.interface import Interface46          >>> class I1(Interface): pass47          ...48          >>> class I2(I1): pass49          ...50          >>> class I3(Interface): pass51          ...52          >>> class I4(I3): pass53          ...54          >>> spec = Declaration(I2, I3)55          >>> spec = Declaration(I4, spec)56          >>> int(I1 in spec)57          058          >>> int(I2 in spec)59          160          >>> int(I3 in spec)61          162          >>> int(I4 in spec)63          164        """65        return self.extends(interface) and interface in self.interfaces()66    def __iter__(self):67        """Return an iterator for the interfaces in the specification68        for example:69          >>> from zope.interface import Interface70          >>> class I1(Interface): pass71          ...72          >>> class I2(I1): pass73          ...74          >>> class I3(Interface): pass75          ...76          >>> class I4(I3): pass77          ...78          >>> spec = Declaration(I2, I3)79          >>> spec = Declaration(I4, spec)80          >>> i = iter(spec)81          >>> i.next().getName()82          'I4'83          >>> i.next().getName()84          'I2'85          >>> i.next().getName()86          'I3'87          >>> list(i)88          []89        """90        return self.interfaces()91    def flattened(self):92        """Return an iterator of all included and extended interfaces93        for example:94          >>> from zope.interface import Interface95          >>> class I1(Interface): pass96          ...97          >>> class I2(I1): pass98          ...99          >>> class I3(Interface): pass100          ...101          >>> class I4(I3): pass102          ...103          >>> spec = Declaration(I2, I3)104          >>> spec = Declaration(I4, spec)105          >>> i = spec.flattened()106          >>> i.next().getName()107          'I4'108          >>> i.next().getName()109          'I2'110          >>> i.next().getName()111          'I1'112          >>> i.next().getName()113          'I3'114          >>> i.next().getName()115          'Interface'116          >>> list(i)117          []118        """119        return iter(self.__iro__)120    def __sub__(self, other):121        """Remove interfaces from a specification122        Examples:123          >>> from zope.interface import Interface124          >>> class I1(Interface): pass125          ...126          >>> class I2(I1): pass127          ...128          >>> class I3(Interface): pass129          ...130          >>> class I4(I3): pass131          ...132          >>> spec = Declaration()133          >>> [iface.getName() for iface in spec]134          []135          >>> spec -= I1136          >>> [iface.getName() for iface in spec]137          []138          >>> spec -= Declaration(I1, I2)139          >>> [iface.getName() for iface in spec]140          []141          >>> spec = Declaration(I2, I4)142          >>> [iface.getName() for iface in spec]143          ['I2', 'I4']144          >>> [iface.getName() for iface in spec - I4]145          ['I2']146          >>> [iface.getName() for iface in spec - I1]147          ['I4']148          >>> [iface.getName() for iface149          ...  in spec - Declaration(I3, I4)]150          ['I2']151        """152        return Declaration(153            *[i for i in self.interfaces()154                if not [j for j in other.interfaces()155                        if i.extends(j, 0)]156                ]157                )158    def __add__(self, other):159        """Add two specifications or a specification and an interface160        Examples:161          >>> from zope.interface import Interface162          >>> class I1(Interface): pass163          ...164          >>> class I2(I1): pass165          ...166          >>> class I3(Interface): pass167          ...168          >>> class I4(I3): pass169          ...170          >>> spec = Declaration()171          >>> [iface.getName() for iface in spec]172          []173          >>> [iface.getName() for iface in spec+I1]174          ['I1']175          >>> [iface.getName() for iface in I1+spec]176          ['I1']177          >>> spec2 = spec178          >>> spec += I1179          >>> [iface.getName() for iface in spec]180          ['I1']181          >>> [iface.getName() for iface in spec2]182          []183          >>> spec2 += Declaration(I3, I4)184          >>> [iface.getName() for iface in spec2]185          ['I3', 'I4']186          >>> [iface.getName() for iface in spec+spec2]187          ['I1', 'I3', 'I4']188          >>> [iface.getName() for iface in spec2+spec]189          ['I3', 'I4', 'I1']190        """191        seen = {}192        result = []193        for i in self.interfaces():194            if i not in seen:195                seen[i] = 1196                result.append(i)197        for i in other.interfaces():198            if i not in seen:199                seen[i] = 1200                result.append(i)201        return Declaration(*result)202    __radd__ = __add__203##############################################################################204#205# Implementation specifications206#207# These specify interfaces implemented by instances of classes208class Implements(Declaration):209    # class whose specification should be used as additional base210    inherit = None211    # interfaces actually declared for a class212    declared = ()213    __name__ = '?'214    def __repr__(self):215        return '<implementedBy %s>' % (self.__name__)216    def __reduce__(self):217        return implementedBy, (self.inherit, )218def implementedByFallback(cls):219    """Return the interfaces implemented for a class' instances220      The value returned is an IDeclaration.221      for example:222        >>> from zope.interface import Interface223        >>> class I1(Interface): pass224        ...225        >>> class I2(I1): pass226        ...227        >>> class I3(Interface): pass228        ...229        >>> class I4(I3): pass230        ...231        >>> class C1(object):232        ...   implements(I2)233        >>> class C2(C1):234        ...   implements(I3)235        >>> [i.getName() for i in implementedBy(C2)]236        ['I3', 'I2']237      Really, any object should be able to receive a successful answer, even238      an instance:239        >>> class Callable(object):240        ...     def __call__(self):241        ...         return self242        >>> implementedBy(Callable())243        <implementedBy zope.interface.declarations.?>244      Note that the name of the spec ends with a '?', because the `Callable`245      instance does not have a `__name__` attribute.246      """247    # This also manages storage of implementation specifications248    try:249        spec = cls.__dict__.get('__implemented__')250    except AttributeError:251        # we can't get the class dict. This is probably due to a252        # security proxy.  If this is the case, then probably no253        # descriptor was installed for the class.254        # We don't want to depend directly on zope.security in255        # zope.interface, but we'll try to make reasonable256        # accommodations in an indirect way.257        # We'll check to see if there's an implements:258        spec = getattr(cls, '__implemented__', None)259        if spec is None:260            # There's no spec stred in the class. Maybe its a builtin:261            spec = BuiltinImplementationSpecifications.get(cls)262            if spec is not None:263                return spec264            return _empty265        if spec.__class__ == Implements:266            # we defaulted to _empty or there was a spec. Good enough.267            # Return it.268            return spec269        # TODO: need old style __implements__ compatibility?270        # Hm, there's an __implemented__, but it's not a spec. Must be271        # an old-style declaration. Just compute a spec for it272        return Declaration(*_normalizeargs((spec, )))273    if isinstance(spec, Implements):274        return spec275    if spec is None:276        spec = BuiltinImplementationSpecifications.get(cls)277        if spec is not None:278            return spec279    # TODO: need old style __implements__ compatibility?280    if spec is not None:281        # old-style __implemented__ = foo declaration282        spec = (spec, ) # tuplefy, as it might be just an int283        spec = Implements(*_normalizeargs(spec))284        spec.inherit = None    # old-style implies no inherit285        del cls.__implemented__ # get rid of the old-style declaration286    else:287        try:288            bases = cls.__bases__289        except AttributeError:290            if not callable(cls):291                raise TypeError("ImplementedBy called for non-factory", cls)292            bases = ()293        spec = Implements(*[implementedBy(c) for c in bases])294        spec.inherit = cls295    spec.__name__ = (getattr(cls, '__module__', '?') or '?') + \296                    '.' + (getattr(cls, '__name__', '?') or '?')297    try:298        cls.__implemented__ = spec299        if not hasattr(cls, '__providedBy__'):300            cls.__providedBy__ = objectSpecificationDescriptor301        if (isinstance(cls, DescriptorAwareMetaClasses)302            and303            '__provides__' not in cls.__dict__):304            # Make sure we get a __provides__ descriptor305            cls.__provides__ = ClassProvides(306                cls,307                getattr(cls, '__class__', type(cls)),308                )309    except TypeError:310        if not isinstance(cls, type):311            raise TypeError("ImplementedBy called for non-type", cls)312        BuiltinImplementationSpecifications[cls] = spec313    return spec314implementedBy = implementedByFallback315def classImplementsOnly(cls, *interfaces):316    """Declare the only interfaces implemented by instances of a class317      The arguments after the class are one or more interfaces or interface318      specifications (``IDeclaration`` objects).319      The interfaces given (including the interfaces in the specifications)320      replace any previous declarations.321      Consider the following example:322        >>> from zope.interface import Interface323        >>> class I1(Interface): pass324        ...325        >>> class I2(Interface): pass326        ...327        >>> class I3(Interface): pass328        ...329        >>> class I4(Interface): pass330        ...331        >>> class A(object):332        ...   implements(I3)333        >>> class B(object):334        ...   implements(I4)335        >>> class C(A, B):336        ...   pass337        >>> classImplementsOnly(C, I1, I2)338        >>> [i.getName() for i in implementedBy(C)]339        ['I1', 'I2']340      Instances of ``C`` provide only ``I1``, ``I2``, and regardless of341      whatever interfaces instances of ``A`` and ``B`` implement.342      """343    spec = implementedBy(cls)344    spec.declared = ()345    spec.inherit = None346    classImplements(cls, *interfaces)347def classImplements(cls, *interfaces):348    """Declare additional interfaces implemented for instances of a class349      The arguments after the class are one or more interfaces or350      interface specifications (``IDeclaration`` objects).351      The interfaces given (including the interfaces in the specifications)352      are added to any interfaces previously declared.353      Consider the following example:354        >>> from zope.interface import Interface355        >>> class I1(Interface): pass356        ...357        >>> class I2(Interface): pass358        ...359        >>> class I3(Interface): pass360        ...361        >>> class I4(Interface): pass362        ...363        >>> class I5(Interface): pass364        ...365        >>> class A(object):366        ...   implements(I3)367        >>> class B(object):368        ...   implements(I4)369        >>> class C(A, B):370        ...   pass371        >>> classImplements(C, I1, I2)372        >>> [i.getName() for i in implementedBy(C)]373        ['I1', 'I2', 'I3', 'I4']374        >>> classImplements(C, I5)375        >>> [i.getName() for i in implementedBy(C)]376        ['I1', 'I2', 'I5', 'I3', 'I4']377      Instances of ``C`` provide ``I1``, ``I2``, ``I5``, and whatever378      interfaces instances of ``A`` and ``B`` provide.379      """380    spec = implementedBy(cls)381    spec.declared += tuple(_normalizeargs(interfaces))382    # compute the bases383    bases = []384    seen = {}385    for b in spec.declared:386        if b not in seen:387            seen[b] = 1388            bases.append(b)389    if spec.inherit is not None:390        for c in spec.inherit.__bases__:391            b = implementedBy(c)392            if b not in seen:393                seen[b] = 1394                bases.append(b)395    spec.__bases__ = tuple(bases)396def _implements_advice(cls):397    interfaces, classImplements = cls.__dict__['__implements_advice_data__']398    del cls.__implements_advice_data__399    classImplements(cls, *interfaces)400    return cls401class implementer:402    def __init__(self, *interfaces):403        self.interfaces = interfaces404    def __call__(self, ob):405        if isinstance(ob, DescriptorAwareMetaClasses):406            raise TypeError("Can't use implementer with classes.  Use one of "407                            "the class-declaration functions instead."408                            )409        spec = Implements(*self.interfaces)410        try:411            ob.__implemented__ = spec412        except AttributeError:413            raise TypeError("Can't declare implements", ob)414        return ob415def _implements(name, interfaces, classImplements):416    frame = sys._getframe(2)417    locals = frame.f_locals418    # Try to make sure we were called from a class def. In 2.2.0 we can't419    # check for __module__ since it doesn't seem to be added to the locals420    # until later on.421    if (locals is frame.f_globals) or (422        ('__module__' not in locals) and sys.version_info[:3] > (2, 2, 0)):423        raise TypeError(name+" can be used only from a class definition.")424    if '__implements_advice_data__' in locals:425        raise TypeError(name+" can be used only once in a class definition.")426    locals['__implements_advice_data__'] = interfaces, classImplements427    addClassAdvisor(_implements_advice, depth=3)428def implements(*interfaces):429    """Declare interfaces implemented by instances of a class430      This function is called in a class definition.431      The arguments are one or more interfaces or interface432      specifications (IDeclaration objects).433      The interfaces given (including the interfaces in the434      specifications) are added to any interfaces previously435      declared.436      Previous declarations include declarations for base classes437      unless implementsOnly was used.438      This function is provided for convenience. It provides a more439      convenient way to call classImplements. For example::440        implements(I1)441      is equivalent to calling::442        classImplements(C, I1)443      after the class has been created.444      Consider the following example::445        >>> from zope.interface import Interface446        >>> class IA1(Interface): pass447        ...448        >>> class IA2(Interface): pass449        ...450        >>> class IB(Interface): pass451        ...452        >>> class IC(Interface): pass453        ...454        >>> class A(object): implements(IA1, IA2)455        ...456        >>> class B(object): implements(IB)457        ...458        >>> class C(A, B):459        ...    implements(IC)460        >>> ob = C()461        >>> int(IA1 in providedBy(ob))462        1463        >>> int(IA2 in providedBy(ob))464        1465        >>> int(IB in providedBy(ob))466        1467        >>> int(IC in providedBy(ob))468        1469      Instances of ``C`` implement ``I1``, ``I2``, and whatever interfaces470      instances of ``A`` and ``B`` implement.471      """472    _implements("implements", interfaces, classImplements)473def implementsOnly(*interfaces):474    """Declare the only interfaces implemented by instances of a class475      This function is called in a class definition.476      The arguments are one or more interfaces or interface477      specifications (IDeclaration objects).478      Previous declarations including declarations for base classes479      are overridden.480      This function is provided for convenience. It provides a more481      convenient way to call classImplementsOnly. For example::482        implementsOnly(I1)483      is equivalent to calling::484        classImplementsOnly(I1)485      after the class has been created.486      Consider the following example::487        >>> from zope.interface import Interface488        >>> class IA1(Interface): pass489        ...490        >>> class IA2(Interface): pass491        ...492        >>> class IB(Interface): pass493        ...494        >>> class IC(Interface): pass495        ...496        >>> class A(object): implements(IA1, IA2)497        ...498        >>> class B(object): implements(IB)499        ...500        >>> class C(A, B):501        ...    implementsOnly(IC)502        >>> ob = C()503        >>> int(IA1 in providedBy(ob))504        0505        >>> int(IA2 in providedBy(ob))506        0507        >>> int(IB in providedBy(ob))508        0509        >>> int(IC in providedBy(ob))510        1511      Instances of ``C`` implement ``IC``, regardless of what512      instances of ``A`` and ``B`` implement.513      """514    _implements("implementsOnly", interfaces, classImplementsOnly)515##############################################################################516#517# Instance declarations518class Provides(Declaration):  # Really named ProvidesClass519    """Implement __provides__, the instance-specific specification520    When an object is pickled, we pickle the interfaces that it implements.521    """522    def __init__(self, cls, *interfaces):523        self.__args = (cls, ) + interfaces524        self._cls = cls525        Declaration.__init__(self, *(interfaces + (implementedBy(cls), )))526    def __reduce__(self):527        return Provides, self.__args528    __module__ = 'zope.interface'529    def __get__(self, inst, cls):530        """Make sure that a class __provides__ doesn't leak to an instance531        For example:532          >>> from zope.interface import Interface533          >>> class IFooFactory(Interface): pass534          ...535          >>> class C(object):536          ...   pass537          >>> C.__provides__ = ProvidesClass(C, IFooFactory)538          >>> [i.getName() for i in C.__provides__]539          ['IFooFactory']540          >>> getattr(C(), '__provides__', 0)541          0542        """543        if inst is None and cls is self._cls:544            # We were accessed through a class, so we are the class'545            # provides spec. Just return this object, but only if we are546            # being called on the same class that we were defined for:547            return self548        raise AttributeError('__provides__')549ProvidesClass = Provides550# Registry of instance declarations551# This is a memory optimization to allow objects to share specifications.552InstanceDeclarations = weakref.WeakValueDictionary()553def Provides(*interfaces):554    """Cache instance declarations555      Instance declarations are shared among instances that have the same556      declaration. The declarations are cached in a weak value dictionary.557      (Note that, in the examples below, we are going to make assertions about558       the size of the weakvalue dictionary.  For the assertions to be559       meaningful, we need to force garbage collection to make sure garbage560       objects are, indeed, removed from the system. Depending on how Python561       is run, we may need to make multiple calls to be sure.  We provide a562       collect function to help with this:563       >>> import gc564       >>> def collect():565       ...     for i in range(4):566       ...         gc.collect()567      )568      >>> collect()569      >>> before = len(InstanceDeclarations)570      >>> class C(object):571      ...    pass572      >>> from zope.interface import Interface573      >>> class I(Interface):574      ...    pass575      >>> c1 = C()576      >>> c2 = C()577      >>> len(InstanceDeclarations) == before578      1579      >>> directlyProvides(c1, I)580      >>> len(InstanceDeclarations) == before + 1581      1582      >>> directlyProvides(c2, I)583      >>> len(InstanceDeclarations) == before + 1584      1585      >>> del c1586      >>> collect()587      >>> len(InstanceDeclarations) == before + 1588      1589      >>> del c2590      >>> collect()591      >>> len(InstanceDeclarations) == before592      1593      """594    spec = InstanceDeclarations.get(interfaces)595    if spec is None:596        spec = ProvidesClass(*interfaces)597        InstanceDeclarations[interfaces] = spec598    return spec599Provides.__safe_for_unpickling__ = True600DescriptorAwareMetaClasses = ClassType, type601def directlyProvides(object, *interfaces):602    """Declare interfaces declared directly for an object603      The arguments after the object are one or more interfaces or interface604      specifications (``IDeclaration`` objects).605      The interfaces given (including the interfaces in the specifications)606      replace interfaces previously declared for the object.607      Consider the following example:608        >>> from zope.interface import Interface609        >>> class I1(Interface): pass610        ...611        >>> class I2(Interface): pass612        ...613        >>> class IA1(Interface): pass614        ...615        >>> class IA2(Interface): pass616        ...617        >>> class IB(Interface): pass618        ...619        >>> class IC(Interface): pass620        ...621        >>> class A(object): implements(IA1, IA2)622        ...623        >>> class B(object): implements(IB)624        ...625        >>> class C(A, B):626        ...    implements(IC)627        >>> ob = C()628        >>> directlyProvides(ob, I1, I2)629        >>> int(I1 in providedBy(ob))630        1631        >>> int(I2 in providedBy(ob))632        1633        >>> int(IA1 in providedBy(ob))634        1635        >>> int(IA2 in providedBy(ob))636        1637        >>> int(IB in providedBy(ob))638        1639        >>> int(IC in providedBy(ob))640        1641      The object, ``ob`` provides ``I1``, ``I2``, and whatever interfaces642      instances have been declared for instances of ``C``.643      To remove directly provided interfaces, use ``directlyProvidedBy`` and644      subtract the unwanted interfaces. For example:645        >>> directlyProvides(ob, directlyProvidedBy(ob)-I2)646        >>> int(I1 in providedBy(ob))647        1648        >>> int(I2 in providedBy(ob))649        0650      removes I2 from the interfaces directly provided by ``ob``. The object,651      ``ob`` no longer directly provides ``I2``, although it might still652      provide ``I2`` if it's class implements ``I2``.653      To add directly provided interfaces, use ``directlyProvidedBy`` and654      include additional interfaces.  For example:655        >>> int(I2 in providedBy(ob))656        0657        >>> directlyProvides(ob, directlyProvidedBy(ob), I2)658      adds ``I2`` to the interfaces directly provided by ob::659        >>> int(I2 in providedBy(ob))660        1661      """662    # We need to avoid setting this attribute on meta classes that663    # don't support descriptors.664    # We can do away with this check when we get rid of the old EC665    cls = getattr(object, '__class__', None)666    if cls is not None and getattr(cls,  '__class__', None) is cls:667        # It's a meta class (well, at least it it could be an extension class)668        if not isinstance(object, DescriptorAwareMetaClasses):669            raise TypeError("Attempt to make an interface declaration on a "670                            "non-descriptor-aware class")671    interfaces = _normalizeargs(interfaces)672    if cls is None:673        cls = type(object)674    issub = False675    for damc in DescriptorAwareMetaClasses:676        if issubclass(cls, damc):677            issub = True678            break679    if issub:680        # we have a class or type.  We'll use a special descriptor681        # that provides some extra caching682        object.__provides__ = ClassProvides(object, cls, *interfaces)683    else:684        object.__provides__ = Provides(cls, *interfaces)685def alsoProvides(object, *interfaces):686    """Declare interfaces declared directly for an object687    The arguments after the object are one or more interfaces or interface688    specifications (``IDeclaration`` objects).689    The interfaces given (including the interfaces in the specifications) are690    added to the interfaces previously declared for the object.691    Consider the following example:692      >>> from zope.interface import Interface693      >>> class I1(Interface): pass694      ...695      >>> class I2(Interface): pass696      ...697      >>> class IA1(Interface): pass698      ...699      >>> class IA2(Interface): pass700      ...701      >>> class IB(Interface): pass702      ...703      >>> class IC(Interface): pass704      ...705      >>> class A(object): implements(IA1, IA2)706      ...707      >>> class B(object): implements(IB)708      ...709      >>> class C(A, B):710      ...    implements(IC)711      >>> ob = C()712      >>> directlyProvides(ob, I1)713      >>> int(I1 in providedBy(ob))714      1715      >>> int(I2 in providedBy(ob))716      0717      >>> int(IA1 in providedBy(ob))718      1719      >>> int(IA2 in providedBy(ob))720      1721      >>> int(IB in providedBy(ob))722      1723      >>> int(IC in providedBy(ob))724      1725      >>> alsoProvides(ob, I2)726      >>> int(I1 in providedBy(ob))727      1728      >>> int(I2 in providedBy(ob))729      1730      >>> int(IA1 in providedBy(ob))731      1732      >>> int(IA2 in providedBy(ob))733      1734      >>> int(IB in providedBy(ob))735      1736      >>> int(IC in providedBy(ob))737      1738    The object, ``ob`` provides ``I1``, ``I2``, and whatever interfaces739    instances have been declared for instances of ``C``. Notice that the740    alsoProvides just extends the provided interfaces.741    """742    directlyProvides(object, directlyProvidedBy(object), *interfaces)743def noLongerProvides(object, interface):744    """745    This removes a directly provided interface from an object.746    Consider the following two interfaces:747      >>> from zope.interface import Interface748      >>> class I1(Interface): pass749      ...750      >>> class I2(Interface): pass751      ...752    ``I1`` is provided through the class, ``I2`` is directly provided753    by the object:754    755      >>> class C(object):756      ...    implements(I1)757      >>> c = C()758      >>> alsoProvides(c, I2)759      >>> I2.providedBy(c)760      True761    Remove I2 from c again:762      763      >>> noLongerProvides(c, I2)764      >>> I2.providedBy(c)765      False766    Removing an interface that is provided through the class is not possible:767      >>> noLongerProvides(c, I1)768      Traceback (most recent call last):769      ...770      ValueError: Can only remove directly provided interfaces.771    """772    directlyProvides(object, directlyProvidedBy(object)-interface)773    if interface.providedBy(object):774        raise ValueError("Can only remove directly provided interfaces.")775class ClassProvidesBasePy(object):776    def __get__(self, inst, cls):777        if cls is self._cls:778            # We only work if called on the class we were defined for779            if inst is None:780                # We were accessed through a class, so we are the class'781                # provides spec. Just return this object as is:782                return self783            return self._implements784        raise AttributeError('__provides__')785ClassProvidesBase = ClassProvidesBasePy786# Try to get C base:787try:788    import _zope_interface_coptimizations789except ImportError:790    pass791else:792    from _zope_interface_coptimizations import ClassProvidesBase793class ClassProvides(Declaration, ClassProvidesBase):794    """Special descriptor for class __provides__795    The descriptor caches the implementedBy info, so that796    we can get declarations for objects without instance-specific797    interfaces a bit quicker.798    For example:799      >>> from zope.interface import Interface800      >>> class IFooFactory(Interface):801      ...     pass802      >>> class IFoo(Interface):803      ...     pass804      >>> class C(object):805      ...     implements(IFoo)806      ...     classProvides(IFooFactory)807      >>> [i.getName() for i in C.__provides__]808      ['IFooFactory']809      >>> [i.getName() for i in C().__provides__]810      ['IFoo']811    """812    def __init__(self, cls, metacls, *interfaces):813        self._cls = cls814        self._implements = implementedBy(cls)815        self.__args = (cls, metacls, ) + interfaces816        Declaration.__init__(self, *(interfaces + (implementedBy(metacls), )))817    def __reduce__(self):818        return self.__class__, self.__args819    # Copy base-class method for speed820    __get__ = ClassProvidesBase.__get__821def directlyProvidedBy(object):822    """Return the interfaces directly provided by the given object823    The value returned is an ``IDeclaration``.824    """825    provides = getattr(object, "__provides__", None)826    if (provides is None # no spec827        or828        # We might have gotten the implements spec, as an829        # optimization. If so, it's like having only one base, that we830        # lop off to exclude class-supplied declarations:831        isinstance(provides, Implements)832        ):833        return _empty834    # Strip off the class part of the spec:835    return Declaration(provides.__bases__[:-1])836def classProvides(*interfaces):837    """Declare interfaces provided directly by a class838      This function is called in a class definition.839      The arguments are one or more interfaces or interface specifications840      (``IDeclaration`` objects).841      The given interfaces (including the interfaces in the specifications)842      are used to create the class's direct-object interface specification.843      An error will be raised if the module class has an direct interface844      specification. In other words, it is an error to call this function more845      than once in a class definition.846      Note that the given interfaces have nothing to do with the interfaces847      implemented by instances of the class.848      This function is provided for convenience. It provides a more convenient849      way to call directlyProvides for a class. For example::850        classProvides(I1)851      is equivalent to calling::852        directlyProvides(theclass, I1)853      after the class has been created.854      For example:855        >>> from zope.interface import Interface856        >>> class IFoo(Interface): pass857        ...858        >>> class IFooFactory(Interface): pass859        ...860        >>> class C(object):861        ...   implements(IFoo)862        ...   classProvides(IFooFactory)863        >>> [i.getName() for i in C.__providedBy__]864        ['IFooFactory']865        >>> [i.getName() for i in C().__providedBy__]866        ['IFoo']867      if equivalent to:868        >>> from zope.interface import Interface869        >>> class IFoo(Interface): pass870        ...871        >>> class IFooFactory(Interface): pass872        ...873        >>> class C(object):874        ...   implements(IFoo)875        >>> directlyProvides(C, IFooFactory)876        >>> [i.getName() for i in C.__providedBy__]877        ['IFooFactory']878        >>> [i.getName() for i in C().__providedBy__]879        ['IFoo']880      If classProvides is called outside of a class definition, it fails.881        >>> classProvides(IFooFactory)882        Traceback (most recent call last):883        ...884        TypeError: classProvides can be used only from a class definition.885      """886    frame = sys._getframe(1)887    locals = frame.f_locals888    # Try to make sure we were called from a class def889    if (locals is frame.f_globals) or ('__module__' not in locals):890        raise TypeError("classProvides can be used only from a class definition.")891    if '__provides__' in locals:892        raise TypeError(893            "classProvides can only be used once in a class definition.")894    locals["__provides__"] = _normalizeargs(interfaces)895    addClassAdvisor(_classProvides_advice, depth=2)896def _classProvides_advice(cls):897    interfaces = cls.__dict__['__provides__']898    del cls.__provides__899    directlyProvides(cls, *interfaces)900    return cls901def moduleProvides(*interfaces):902    """Declare interfaces provided by a module903    This function is used in a module definition.904    The arguments are one or more interfaces or interface specifications905    (``IDeclaration`` objects).906    The given interfaces (including the interfaces in the specifications) are907    used to create the module's direct-object interface specification.  An908    error will be raised if the module already has an interface specification.909    In other words, it is an error to call this function more than once in a910    module definition.911    This function is provided for convenience. It provides a more convenient912    way to call directlyProvides. For example::913      moduleImplements(I1)914    is equivalent to::915      directlyProvides(sys.modules[__name__], I1)916    """917    frame = sys._getframe(1)918    locals = frame.f_locals919    # Try to make sure we were called from a class def920    if (locals is not frame.f_globals) or ('__name__' not in locals):921        raise TypeError(922            "moduleProvides can only be used from a module definition.")923    if '__provides__' in locals:924        raise TypeError(925            "moduleProvides can only be used once in a module definition.")926    locals["__provides__"] = Provides(ModuleType,927                                      *_normalizeargs(interfaces))928##############################################################################929#930# Declaration querying support931def ObjectSpecification(direct, cls):932    """Provide object specifications933    These combine information for the object and for it's classes.934    For example:935      >>> from zope.interface import Interface936      >>> class I1(Interface): pass937      ...938      >>> class I2(Interface): pass939      ...940      >>> class I3(Interface): pass941      ...942      >>> class I31(I3): pass943      ...944      >>> class I4(Interface): pass945      ...946      >>> class I5(Interface): pass947      ...948      >>> class A(object): implements(I1)949      ...950      >>> class B(object): __implemented__ = I2951      ...952      >>> class C(A, B): implements(I31)953      ...954      >>> c = C()955      >>> directlyProvides(c, I4)956      >>> [i.getName() for i in providedBy(c)]957      ['I4', 'I31', 'I1', 'I2']958      >>> [i.getName() for i in providedBy(c).flattened()]959      ['I4', 'I31', 'I3', 'I1', 'I2', 'Interface']960      >>> int(I1 in providedBy(c))961      1962      >>> int(I3 in providedBy(c))963      0964      >>> int(providedBy(c).extends(I3))965      1966      >>> int(providedBy(c).extends(I31))967      1968      >>> int(providedBy(c).extends(I5))969      0970      >>> class COnly(A, B): implementsOnly(I31)971      ...972      >>> class D(COnly): implements(I5)973      ...974      >>> c = D()975      >>> directlyProvides(c, I4)976      >>> [i.getName() for i in providedBy(c)]977      ['I4', 'I5', 'I31']978      >>> [i.getName() for i in providedBy(c).flattened()]979      ['I4', 'I5', 'I31', 'I3', 'Interface']980      >>> int(I1 in providedBy(c))981      0982      >>> int(I3 in providedBy(c))983      0984      >>> int(providedBy(c).extends(I3))985      1986      >>> int(providedBy(c).extends(I1))987      0988      >>> int(providedBy(c).extends(I31))989      1990      >>> int(providedBy(c).extends(I5))991      1992    """993    return Provides(cls, direct)994def getObjectSpecification(ob):995    provides = getattr(ob, '__provides__', None)996    if provides is not None:997        return provides998    try:999        cls = ob.__class__1000    except AttributeError:1001        # We can't get the class, so just consider provides1002        return _empty1003    return implementedBy(cls)1004def providedBy(ob):1005    # Here we have either a special object, an old-style declaration1006    # or a descriptor1007    # Try to get __providedBy__1008    try:1009        r = ob.__providedBy__1010    except AttributeError:1011        # Not set yet. Fall back to lower-level thing that computes it1012        return getObjectSpecification(ob)1013    try:1014        # We might have gotten a descriptor from an instance of a1015        # class (like an ExtensionClass) that doesn't support1016        # descriptors.  We'll make sure we got one by trying to get1017        # the only attribute, which all specs have.1018        r.extends1019    except AttributeError:1020        # The object's class doesn't understand descriptors.1021        # Sigh. We need to get an object descriptor, but we have to be1022        # careful.  We want to use the instance's __provides__, if1023        # there is one, but only if it didn't come from the class.1024        try:1025            r = ob.__provides__1026        except AttributeError:1027            # No __provides__, so just fall back to implementedBy1028            return implementedBy(ob.__class__)1029        # We need to make sure we got the __provides__ from the1030        # instance. We'll do this by making sure we don't get the same1031        # thing from the class:1032        try:1033            cp = ob.__class__.__provides__1034        except AttributeError:1035            # The ob doesn't have a class or the class has no1036            # provides, assume we're done:1037            return r1038        if r is cp:1039            # Oops, we got the provides from the class. This means1040            # the object doesn't have it's own. We should use implementedBy1041            return implementedBy(ob.__class__)1042    return r1043class ObjectSpecificationDescriptorPy(object):1044    """Implement the `__providedBy__` attribute1045    The `__providedBy__` attribute computes the interfaces peovided by1046    an object.1047    """1048    def __get__(self, inst, cls):1049        """Get an object specification for an object1050        For example:1051          >>> from zope.interface import Interface1052          >>> class IFoo(Interface): pass1053          ...1054          >>> class IFooFactory(Interface): pass1055          ...1056          >>> class C(object):1057          ...   implements(IFoo)1058          ...   classProvides(IFooFactory)1059          >>> [i.getName() for i in C.__providedBy__]1060          ['IFooFactory']1061          >>> [i.getName() for i in C().__providedBy__]1062          ['IFoo']1063        """1064        # Get an ObjectSpecification bound to either an instance or a class,1065        # depending on how we were accessed.1066        if inst is None:1067            return getObjectSpecification(cls)1068        provides = getattr(inst, '__provides__', None)1069        if provides is not None:1070            return provides1071        return implementedBy(cls)1072ObjectSpecificationDescriptor = ObjectSpecificationDescriptorPy1073##############################################################################1074def _normalizeargs(sequence, output = None):1075    """Normalize declaration arguments1076    Normalization arguments might contain Declarions, tuples, or single1077    interfaces.1078    Anything but individial interfaces or implements specs will be expanded.1079    """1080    if output is None:1081        output = []1082    cls = sequence.__class__1083    if InterfaceClass in cls.__mro__ or Implements in cls.__mro__:1084        output.append(sequence)1085    else:1086        for v in sequence:1087            _normalizeargs(v, output)1088    return output1089_empty = Declaration()1090try:1091    import _zope_interface_coptimizations1092except ImportError:1093    pass1094else:1095    from _zope_interface_coptimizations import implementedBy, providedBy1096    from _zope_interface_coptimizations import getObjectSpecification1097    from _zope_interface_coptimizations import ObjectSpecificationDescriptor...

test_flow_dependencies.py

Source:test_flow_dependencies.py

...211        flow = lf.Flow('lf', utils.TaskOneArgOneReturn('rt', requires='x',212                                                       provides='x'))213        self.assertEqual(set('x'), flow.requires)214        self.assertEqual(set('x'), flow.provides)215    def test_retry_in_linear_flow_no_requirements_no_provides(self):216        flow = lf.Flow('lf', retry.AlwaysRevert('rt'))217        self.assertEqual(set(), flow.requires)218        self.assertEqual(set(), flow.provides)219    def test_retry_in_linear_flow_with_requirements(self):220        flow = lf.Flow('lf', retry.AlwaysRevert('rt', requires=['x', 'y']))221        self.assertEqual(set(['x', 'y']), flow.requires)222        self.assertEqual(set(), flow.provides)223    def test_retry_in_linear_flow_with_provides(self):224        flow = lf.Flow('lf', retry.AlwaysRevert('rt', provides=['x', 'y']))225        self.assertEqual(set(), flow.requires)226        self.assertEqual(set(['x', 'y']), flow.provides)227    def test_retry_in_linear_flow_requires_and_provides(self):228        flow = lf.Flow('lf', retry.AlwaysRevert('rt',229                                                requires=['x', 'y'],230                                                provides=['a', 'b']))231        self.assertEqual(set(['x', 'y']), flow.requires)232        self.assertEqual(set(['a', 'b']), flow.provides)233    def test_retry_requires_and_provides_same_value(self):234        flow = lf.Flow('lf', retry.AlwaysRevert('rt',235                                                requires=['x', 'y'],236                                                provides=['x', 'y']))237        self.assertEqual(set(['x', 'y']), flow.requires)238        self.assertEqual(set(['x', 'y']), flow.provides)239    def test_retry_in_unordered_flow_no_requirements_no_provides(self):240        flow = uf.Flow('uf', retry.AlwaysRevert('rt'))241        self.assertEqual(set(), flow.requires)242        self.assertEqual(set(), flow.provides)243    def test_retry_in_unordered_flow_with_requirements(self):244        flow = uf.Flow('uf', retry.AlwaysRevert('rt', requires=['x', 'y']))245        self.assertEqual(set(['x', 'y']), flow.requires)246        self.assertEqual(set(), flow.provides)247    def test_retry_in_unordered_flow_with_provides(self):248        flow = uf.Flow('uf', retry.AlwaysRevert('rt', provides=['x', 'y']))249        self.assertEqual(set(), flow.requires)250        self.assertEqual(set(['x', 'y']), flow.provides)251    def test_retry_in_unordered_flow_requires_and_provides(self):252        flow = uf.Flow('uf', retry.AlwaysRevert('rt',253                                                requires=['x', 'y'],254                                                provides=['a', 'b']))255        self.assertEqual(set(['x', 'y']), flow.requires)256        self.assertEqual(set(['a', 'b']), flow.provides)257    def test_retry_in_graph_flow_no_requirements_no_provides(self):258        flow = gf.Flow('gf', retry.AlwaysRevert('rt'))259        self.assertEqual(set(), flow.requires)260        self.assertEqual(set(), flow.provides)261    def test_retry_in_graph_flow_with_requirements(self):262        flow = gf.Flow('gf', retry.AlwaysRevert('rt', requires=['x', 'y']))263        self.assertEqual(set(['x', 'y']), flow.requires)264        self.assertEqual(set(), flow.provides)265    def test_retry_in_graph_flow_with_provides(self):266        flow = gf.Flow('gf', retry.AlwaysRevert('rt', provides=['x', 'y']))267        self.assertEqual(set(), flow.requires)268        self.assertEqual(set(['x', 'y']), flow.provides)269    def test_retry_in_graph_flow_requires_and_provides(self):270        flow = gf.Flow('gf', retry.AlwaysRevert('rt',271                                                requires=['x', 'y'],272                                                provides=['a', 'b']))273        self.assertEqual(set(['x', 'y']), flow.requires)274        self.assertEqual(set(['a', 'b']), flow.provides)275    def test_linear_flow_retry_and_task(self):276        flow = lf.Flow('lf', retry.AlwaysRevert('rt',277                                                requires=['x', 'y'],278                                                provides=['a', 'b']))279        flow.add(utils.TaskMultiArgOneReturn(rebind=['a', 'x', 'c'],280                                             provides=['z']))281        self.assertEqual(set(['x', 'y', 'c']), flow.requires)282        self.assertEqual(set(['a', 'b', 'z']), flow.provides)283    def test_unordered_flow_retry_and_task(self):284        flow = uf.Flow('uf', retry.AlwaysRevert('rt',285                                                requires=['x', 'y'],286                                                provides=['a', 'b']))287        flow.add(utils.TaskMultiArgOneReturn(rebind=['a', 'x', 'c'],288                                             provides=['z']))289        self.assertEqual(set(['x', 'y', 'c']), flow.requires)290        self.assertEqual(set(['a', 'b', 'z']), flow.provides)291    def test_unordered_flow_retry_and_task_same_requires_provides(self):292        flow = uf.Flow('uf', retry.AlwaysRevert('rt', requires=['x']))293        flow.add(utils.TaskOneReturn(provides=['x']))294        self.assertEqual(set(['x']), flow.requires)295        self.assertEqual(set(['x']), flow.provides)296    def test_unordered_flow_retry_and_task_provide_same_value(self):297        flow = uf.Flow('uf', retry.AlwaysRevert('rt', provides=['x']))298        flow.add(utils.TaskOneReturn('t1', provides=['x']))299        self.assertEqual(set(['x']), flow.provides)300    def test_unordered_flow_retry_two_tasks_provide_same_value(self):301        flow = uf.Flow('uf', retry.AlwaysRevert('rt', provides=['y']))302        flow.add(utils.TaskOneReturn('t1', provides=['x']),303                 utils.TaskOneReturn('t2', provides=['x']))304        self.assertEqual(set(['x', 'y']), flow.provides)305    def test_graph_flow_retry_and_task(self):...

test_amphora_flows.py

Source:test_amphora_flows.py

1# Copyright 2015 Hewlett-Packard Development Company, L.P.2#3# Licensed under the Apache License, Version 2.0 (the "License"); you may4# not use this file except in compliance with the License. You may obtain5# a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12# License for the specific language governing permissions and limitations13# under the License.14#15import mock16from oslo_config import cfg17from oslo_config import fixture as oslo_fixture18from taskflow.patterns import linear_flow as flow19from octavia.common import constants20from octavia.controller.worker.flows import amphora_flows21import octavia.tests.unit.base as base22AUTH_VERSION = '2'23# NOTE: We patch the get_network_driver for all the calls so we don't24# inadvertently make real calls.25@mock.patch('octavia.common.utils.get_network_driver')26class TestAmphoraFlows(base.TestCase):27    def setUp(self):28        super(TestAmphoraFlows, self).setUp()29        self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))30        self.conf.config(31            group="controller_worker",32            amphora_driver='amphora_haproxy_rest_driver')33        self.conf.config(group="nova", enable_anti_affinity=False)34        self.AmpFlow = amphora_flows.AmphoraFlows()35    def test_get_create_amphora_flow(self, mock_get_net_driver):36        amp_flow = self.AmpFlow.get_create_amphora_flow()37        self.assertIsInstance(amp_flow, flow.Flow)38        self.assertIn(constants.AMPHORA, amp_flow.provides)39        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)40        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)41        self.assertIn(constants.SERVER_PEM, amp_flow.provides)42        self.assertEqual(5, len(amp_flow.provides))43        self.assertEqual(1, len(amp_flow.requires))44    def test_get_create_amphora_flow_cert(self, mock_get_net_driver):45        self.AmpFlow = amphora_flows.AmphoraFlows()46        amp_flow = self.AmpFlow.get_create_amphora_flow()47        self.assertIsInstance(amp_flow, flow.Flow)48        self.assertIn(constants.AMPHORA, amp_flow.provides)49        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)50        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)51        self.assertEqual(5, len(amp_flow.provides))52        self.assertEqual(1, len(amp_flow.requires))53    def test_get_create_amphora_for_lb_flow(self, mock_get_net_driver):54        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(55            'SOMEPREFIX', constants.ROLE_STANDALONE)56        self.assertIsInstance(amp_flow, flow.Flow)57        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)58        self.assertIn(constants.AMPHORA, amp_flow.provides)59        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)60        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)61        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)62        self.assertIn(constants.SERVER_PEM, amp_flow.provides)63        self.assertEqual(5, len(amp_flow.provides))64        self.assertEqual(2, len(amp_flow.requires))65    def test_get_cert_create_amphora_for_lb_flow(self, mock_get_net_driver):66        self.AmpFlow = amphora_flows.AmphoraFlows()67        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(68            'SOMEPREFIX', constants.ROLE_STANDALONE)69        self.assertIsInstance(amp_flow, flow.Flow)70        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)71        self.assertIn(constants.AMPHORA, amp_flow.provides)72        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)73        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)74        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)75        self.assertIn(constants.SERVER_PEM, amp_flow.provides)76        self.assertEqual(5, len(amp_flow.provides))77        self.assertEqual(2, len(amp_flow.requires))78    def test_get_cert_master_create_amphora_for_lb_flow(79            self, mock_get_net_driver):80        self.AmpFlow = amphora_flows.AmphoraFlows()81        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(82            'SOMEPREFIX', constants.ROLE_MASTER)83        self.assertIsInstance(amp_flow, flow.Flow)84        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)85        self.assertIn(constants.AMPHORA, amp_flow.provides)86        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)87        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)88        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)89        self.assertIn(constants.SERVER_PEM, amp_flow.provides)90        self.assertEqual(5, len(amp_flow.provides))91        self.assertEqual(2, len(amp_flow.requires))92    def test_get_cert_master_rest_anti_affinity_create_amphora_for_lb_flow(93            self, mock_get_net_driver):94        self.conf.config(group="nova", enable_anti_affinity=True)95        self.AmpFlow = amphora_flows.AmphoraFlows()96        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(97            'SOMEPREFIX', constants.ROLE_MASTER)98        self.assertIsInstance(amp_flow, flow.Flow)99        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)100        self.assertIn(constants.SERVER_GROUP_ID, amp_flow.requires)101        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)102        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)103        self.assertIn(constants.SERVER_PEM, amp_flow.provides)104        self.assertEqual(5, len(amp_flow.provides))105        self.assertEqual(3, len(amp_flow.requires))106        self.conf.config(group="nova", enable_anti_affinity=False)107    def test_get_cert_backup_create_amphora_for_lb_flow(108            self, mock_get_net_driver):109        self.AmpFlow = amphora_flows.AmphoraFlows()110        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(111            'SOMEPREFIX', constants.ROLE_BACKUP)112        self.assertIsInstance(amp_flow, flow.Flow)113        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)114        self.assertIn(constants.AMPHORA, amp_flow.provides)115        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)116        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)117        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)118        self.assertIn(constants.SERVER_PEM, amp_flow.provides)119        self.assertEqual(5, len(amp_flow.provides))120        self.assertEqual(2, len(amp_flow.requires))121    def test_get_cert_bogus_create_amphora_for_lb_flow(122            self, mock_get_net_driver):123        self.AmpFlow = amphora_flows.AmphoraFlows()124        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(125            'SOMEPREFIX', 'BOGUS_ROLE')126        self.assertIsInstance(amp_flow, flow.Flow)127        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)128        self.assertIn(constants.AMPHORA, amp_flow.provides)129        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)130        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)131        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)132        self.assertIn(constants.SERVER_PEM, amp_flow.provides)133        self.assertEqual(5, len(amp_flow.provides))134        self.assertEqual(2, len(amp_flow.requires))135    def test_get_cert_backup_rest_anti_affinity_create_amphora_for_lb_flow(136            self, mock_get_net_driver):137        self.conf.config(group="nova", enable_anti_affinity=True)138        self.AmpFlow = amphora_flows.AmphoraFlows()139        amp_flow = self.AmpFlow._get_create_amp_for_lb_subflow(140            'SOMEPREFIX', constants.ROLE_BACKUP)141        self.assertIsInstance(amp_flow, flow.Flow)142        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)143        self.assertIn(constants.SERVER_GROUP_ID, amp_flow.requires)144        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)145        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)146        self.assertIn(constants.SERVER_PEM, amp_flow.provides)147        self.assertEqual(5, len(amp_flow.provides))148        self.assertEqual(3, len(amp_flow.requires))149        self.conf.config(group="nova", enable_anti_affinity=False)150    def test_get_delete_amphora_flow(self, mock_get_net_driver):151        amp_flow = self.AmpFlow.get_delete_amphora_flow()152        self.assertIsInstance(amp_flow, flow.Flow)153        self.assertIn(constants.AMPHORA, amp_flow.requires)154        self.assertEqual(0, len(amp_flow.provides))155        self.assertEqual(1, len(amp_flow.requires))156    def test_allocate_amp_to_lb_decider(self, mock_get_net_driver):157        history = mock.MagicMock()158        values = mock.MagicMock(side_effect=[['TEST'], [None]])159        history.values = values160        result = self.AmpFlow._allocate_amp_to_lb_decider(history)161        self.assertTrue(result)162        result = self.AmpFlow._allocate_amp_to_lb_decider(history)163        self.assertFalse(result)164    def test_create_new_amp_for_lb_decider(self, mock_get_net_driver):165        history = mock.MagicMock()166        values = mock.MagicMock(side_effect=[[None], ['TEST']])167        history.values = values168        result = self.AmpFlow._create_new_amp_for_lb_decider(history)169        self.assertTrue(result)170        result = self.AmpFlow._create_new_amp_for_lb_decider(history)171        self.assertFalse(result)172    def test_get_failover_flow_allocated(self, mock_get_net_driver):173        amp_flow = self.AmpFlow.get_failover_flow(174            status=constants.AMPHORA_ALLOCATED)175        self.assertIsInstance(amp_flow, flow.Flow)176        self.assertIn(constants.FAILED_AMPHORA, amp_flow.requires)177        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)178        self.assertIn(constants.AMP_DATA, amp_flow.provides)179        self.assertIn(constants.AMPHORA, amp_flow.provides)180        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)181        self.assertIn(constants.AMPHORAE_NETWORK_CONFIG, amp_flow.provides)182        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)183        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)184        self.assertIn(constants.LISTENERS, amp_flow.provides)185        self.assertIn(constants.LOADBALANCER, amp_flow.provides)186        self.assertEqual(3, len(amp_flow.requires))187        self.assertEqual(11, len(amp_flow.provides))188        amp_flow = self.AmpFlow.get_failover_flow(189            role=constants.ROLE_MASTER, status=constants.AMPHORA_ALLOCATED)190        self.assertIsInstance(amp_flow, flow.Flow)191        self.assertIn(constants.FAILED_AMPHORA, amp_flow.requires)192        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)193        self.assertIn(constants.AMP_DATA, amp_flow.provides)194        self.assertIn(constants.AMPHORA, amp_flow.provides)195        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)196        self.assertIn(constants.AMPHORAE_NETWORK_CONFIG, amp_flow.provides)197        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)198        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)199        self.assertIn(constants.LISTENERS, amp_flow.provides)200        self.assertIn(constants.LOADBALANCER, amp_flow.provides)201        self.assertEqual(3, len(amp_flow.requires))202        self.assertEqual(11, len(amp_flow.provides))203        amp_flow = self.AmpFlow.get_failover_flow(204            role=constants.ROLE_BACKUP, status=constants.AMPHORA_ALLOCATED)205        self.assertIsInstance(amp_flow, flow.Flow)206        self.assertIn(constants.FAILED_AMPHORA, amp_flow.requires)207        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)208        self.assertIn(constants.AMP_DATA, amp_flow.provides)209        self.assertIn(constants.AMPHORA, amp_flow.provides)210        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)211        self.assertIn(constants.AMPHORAE_NETWORK_CONFIG, amp_flow.provides)212        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)213        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)214        self.assertIn(constants.LISTENERS, amp_flow.provides)215        self.assertIn(constants.LOADBALANCER, amp_flow.provides)216        self.assertEqual(3, len(amp_flow.requires))217        self.assertEqual(11, len(amp_flow.provides))218        amp_flow = self.AmpFlow.get_failover_flow(219            role='BOGUSROLE', status=constants.AMPHORA_ALLOCATED)220        self.assertIsInstance(amp_flow, flow.Flow)221        self.assertIn(constants.FAILED_AMPHORA, amp_flow.requires)222        self.assertIn(constants.LOADBALANCER_ID, amp_flow.requires)223        self.assertIn(constants.AMP_DATA, amp_flow.provides)224        self.assertIn(constants.AMPHORA, amp_flow.provides)225        self.assertIn(constants.AMPHORA_ID, amp_flow.provides)226        self.assertIn(constants.AMPHORAE_NETWORK_CONFIG, amp_flow.provides)227        self.assertIn(constants.COMPUTE_ID, amp_flow.provides)228        self.assertIn(constants.COMPUTE_OBJ, amp_flow.provides)229        self.assertIn(constants.LISTENERS, amp_flow.provides)230        self.assertIn(constants.LOADBALANCER, amp_flow.provides)231        self.assertEqual(3, len(amp_flow.requires))232        self.assertEqual(11, len(amp_flow.provides))233    def test_get_failover_flow_spare(self, mock_get_net_driver):234        amp_flow = self.AmpFlow.get_failover_flow(235            status=constants.AMPHORA_READY)236        self.assertIsInstance(amp_flow, flow.Flow)237        self.assertIn(constants.FAILED_AMPHORA, amp_flow.requires)238        self.assertEqual(1, len(amp_flow.requires))239        self.assertEqual(0, len(amp_flow.provides))240    def test_cert_rotate_amphora_flow(self, mock_get_net_driver):241        self.AmpFlow = amphora_flows.AmphoraFlows()242        amp_rotate_flow = self.AmpFlow.cert_rotate_amphora_flow()243        self.assertIsInstance(amp_rotate_flow, flow.Flow)244        self.assertIn(constants.SERVER_PEM, amp_rotate_flow.provides)245        self.assertIn(constants.AMPHORA, amp_rotate_flow.requires)246        self.assertEqual(1, len(amp_rotate_flow.provides))247        self.assertEqual(2, len(amp_rotate_flow.requires))248    def test_get_vrrp_subflow(self, mock_get_net_driver):249        vrrp_subflow = self.AmpFlow.get_vrrp_subflow('123')250        self.assertIsInstance(vrrp_subflow, flow.Flow)251        self.assertIn(constants.LOADBALANCER, vrrp_subflow.provides)252        self.assertIn(constants.LOADBALANCER, vrrp_subflow.requires)253        self.assertEqual(1, len(vrrp_subflow.provides))254        self.assertEqual(1, len(vrrp_subflow.requires))255    def test_get_post_map_lb_subflow(self, mock_get_net_driver):256        self.AmpFlow = amphora_flows.AmphoraFlows()257        amp_flow = self.AmpFlow._get_post_map_lb_subflow(258            'SOMEPREFIX', constants.ROLE_MASTER)259        self.assertIsInstance(amp_flow, flow.Flow)260        self.assertIn(constants.AMPHORA_ID, amp_flow.requires)261        self.assertIn(constants.AMPHORA, amp_flow.provides)262        self.assertEqual(1, len(amp_flow.provides))263        self.assertEqual(1, len(amp_flow.requires))264        amp_flow = self.AmpFlow._get_post_map_lb_subflow(265            'SOMEPREFIX', constants.ROLE_BACKUP)266        self.assertIsInstance(amp_flow, flow.Flow)267        self.assertIn(constants.AMPHORA_ID, amp_flow.requires)268        self.assertIn(constants.AMPHORA, amp_flow.provides)269        self.assertEqual(1, len(amp_flow.provides))270        self.assertEqual(1, len(amp_flow.requires))271        amp_flow = self.AmpFlow._get_post_map_lb_subflow(272            'SOMEPREFIX', constants.ROLE_STANDALONE)273        self.assertIsInstance(amp_flow, flow.Flow)274        self.assertIn(constants.AMPHORA_ID, amp_flow.requires)275        self.assertIn(constants.AMPHORA, amp_flow.provides)276        self.assertEqual(1, len(amp_flow.provides))277        self.assertEqual(1, len(amp_flow.requires))278        amp_flow = self.AmpFlow._get_post_map_lb_subflow(279            'SOMEPREFIX', 'BOGUS_ROLE')280        self.assertIsInstance(amp_flow, flow.Flow)281        self.assertIn(constants.AMPHORA_ID, amp_flow.requires)282        self.assertIn(constants.AMPHORA, amp_flow.provides)283        self.assertEqual(1, len(amp_flow.provides))...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.