Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run avocado automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
