How to use register_if method in Slash

Best Python code snippet using slash

option_types.py

Source:option_types.py

...118    def __get__(self, obj: object, objtype: Any) -> _OptT | _DefaultT:119        ...120    def __get__(self, obj, objtype):121        if obj is None:122            if self._register_if(objtype):123                return OptionsInfo(self._flag_names, self.get_flag_options(objtype))124            return None125        long_name = self._flag_names[-1]126        option_value = getattr(obj.options, long_name[2:].replace("-", "_"))127        if option_value is None:128            return None129        return self._convert_(option_value)130# The type of the list members.131# NB: We don't provide constraints, as our `XListOption` types act like a set of contraints132_ListMemberT = TypeVar("_ListMemberT")133class _ListOptionBase(134    _OptionBase["tuple[_ListMemberT, ...]", "tuple[_ListMemberT, ...]"],135    Generic[_ListMemberT],136):...

test_notification.py

Source:test_notification.py

...20from masakari.objects import base21from masakari.objects import fields22from masakari import test23class TestNotificationBase(test.NoDBTestCase):24    @base.MasakariObjectRegistry.register_if(False)25    class TestObject(base.MasakariObject):26        VERSION = '1.0'27        fields = {28            'field_1': fields.StringField(),29            'field_2': fields.IntegerField(),30            'not_important_field': fields.IntegerField(),31        }32    @base.MasakariObjectRegistry.register_if(False)33    class TestNotificationPayload(notification.NotificationPayloadBase):34        VERSION = '1.0'35        SCHEMA = {36            'field_1': ('source_field', 'field_1'),37            'field_2': ('source_field', 'field_2'),38        }39        fields = {40            'extra_field': fields.StringField(),  # filled by ctor41            'field_1': fields.StringField(),  # filled by the schema42            'field_2': fields.IntegerField(),   # filled by the schema43        }44        def populate_schema(self, source_field):45            super(TestNotificationBase.TestNotificationPayload,46                  self).populate_schema(source_field=source_field)47    @base.MasakariObjectRegistry.register_if(False)48    class TestNotificationPayloadEmptySchema(49        notification.NotificationPayloadBase):50        VERSION = '1.0'51        fields = {52            'extra_field': fields.StringField(),  # filled by ctor53        }54    @notification.notification_sample('test-update-1.json')55    @notification.notification_sample('test-update-2.json')56    @base.MasakariObjectRegistry.register_if(False)57    class TestNotification(notification.NotificationBase):58        VERSION = '1.0'59        fields = {60            'payload': fields.ObjectField('TestNotificationPayload')61        }62    @base.MasakariObjectRegistry.register_if(False)63    class TestNotificationEmptySchema(notification.NotificationBase):64        VERSION = '1.0'65        fields = {66            'payload': fields.ObjectField('TestNotificationPayloadEmptySchema')67        }68    fake_service = {69        'created_at': timeutils.utcnow().replace(microsecond=0),70        'updated_at': None,71        'deleted_at': None,72        'deleted': False,73        'id': 123,74        'host': 'fake-host',75        'binary': 'masakari-fake',76        'topic': 'fake-service-topic',77        'report_count': 1,78        'forced_down': False,79        'disabled': False,80        'disabled_reason': None,81        'last_seen_up': None,82        'version': 1}83    expected_payload = {84        'masakari_object.name': 'TestNotificationPayload',85        'masakari_object.data': {86            'extra_field': 'test string',87            'field_1': 'test1',88            'field_2': 42},89        'masakari_object.version': '1.0',90        'masakari_object.namespace': 'masakari'}91    def setUp(self):92        super(TestNotificationBase, self).setUp()93        mock_context = mock.Mock()94        mock_context.to_dict.return_value = {}95        self.publisher = notification.NotificationPublisher(96            context=mock_context, host='fake-host',97            binary='masakari-fake')98        self.my_obj = self.TestObject(field_1='test1',99                                      field_2=42,100                                      not_important_field=13)101        self.payload = self.TestNotificationPayload(102            extra_field='test string')103        self.payload.populate_schema(source_field=self.my_obj)104        self.notification = self.TestNotification(105            event_type=notification.EventType(106                object='test_object',107                action=fields.EventNotificationAction.SEGMENT_CREATE,108                phase=fields.EventNotificationPhase.START),109            publisher=self.publisher,110            priority=fields.EventNotificationPriority.INFO,111            payload=self.payload)112    def _verify_notification(self, mock_notifier, mock_context,113                             expected_event_type,114                             expected_payload):115        mock_notifier.prepare.assert_called_once_with(116            publisher_id='masakari-fake:fake-host')117        mock_notify = mock_notifier.prepare.return_value.info118        self.assertTrue(mock_notify.called)119        self.assertEqual(mock_notify.call_args[0][0], mock_context)120        self.assertEqual(mock_notify.call_args[1]['event_type'],121                         expected_event_type)122        actual_payload = mock_notify.call_args[1]['payload']123        self.assertJsonEqual(expected_payload, actual_payload)124    @mock.patch('masakari.rpc.NOTIFIER')125    def test_emit_notification(self, mock_notifier):126        mock_context = mock.Mock()127        mock_context.to_dict.return_value = {}128        self.notification.emit(mock_context)129        self._verify_notification(130            mock_notifier,131            mock_context,132            expected_event_type='segment.create.start',133            expected_payload=self.expected_payload)134    @mock.patch('masakari.rpc.NOTIFIER')135    def test_emit_with_host_and_binary_as_publisher(self, mock_notifier):136        noti = self.TestNotification(137            event_type=notification.EventType(138                object='test_object',139                action=fields.EventNotificationAction.SEGMENT_CREATE),140            publisher=notification.NotificationPublisher(141                host='fake-host', binary='masakari-fake'),142            priority=fields.EventNotificationPriority.INFO,143            payload=self.payload)144        mock_context = mock.Mock()145        mock_context.to_dict.return_value = {}146        noti.emit(mock_context)147        self._verify_notification(148            mock_notifier,149            mock_context,150            expected_event_type='segment.create',151            expected_payload=self.expected_payload)152    @mock.patch('masakari.rpc.NOTIFIER')153    def test_emit_event_type_without_phase(self, mock_notifier):154        noti = self.TestNotification(155            event_type=notification.EventType(156                object='test_object',157                action=fields.EventNotificationAction.SEGMENT_CREATE),158            publisher=self.publisher,159            priority=fields.EventNotificationPriority.INFO,160            payload=self.payload)161        mock_context = mock.Mock()162        mock_context.to_dict.return_value = {}163        noti.emit(mock_context)164        self._verify_notification(165            mock_notifier,166            mock_context,167            expected_event_type='segment.create',168            expected_payload=self.expected_payload)169    @mock.patch('masakari.rpc.NOTIFIER')170    def test_not_possible_to_emit_if_not_populated(self, mock_notifier):171        non_populated_payload = self.TestNotificationPayload(172            extra_field='test string')173        noti = self.TestNotification(174            event_type=notification.EventType(175                object='test_object',176                action=fields.EventNotificationAction.SEGMENT_CREATE),177            publisher=self.publisher,178            priority=fields.EventNotificationPriority.INFO,179            payload=non_populated_payload)180        mock_context = mock.Mock()181        self.assertRaises(AssertionError, noti.emit, mock_context)182        self.assertFalse(mock_notifier.called)183    @mock.patch('masakari.rpc.NOTIFIER')184    def test_empty_schema(self, mock_notifier):185        non_populated_payload = self.TestNotificationPayloadEmptySchema(186            extra_field='test string')187        noti = self.TestNotificationEmptySchema(188            event_type=notification.EventType(189                object='test_object',190                action=fields.EventNotificationAction.SEGMENT_CREATE),191            publisher=self.publisher,192            priority=fields.EventNotificationPriority.INFO,193            payload=non_populated_payload)194        mock_context = mock.Mock()195        mock_context.to_dict.return_value = {}196        noti.emit(mock_context)197        self._verify_notification(198            mock_notifier,199            mock_context,200            expected_event_type='segment.create',201            expected_payload={202                'masakari_object.name': 'TestNotificationPayloadEmptySchema',203                'masakari_object.data': {'extra_field': 'test string'},204                'masakari_object.version': '1.0',205                'masakari_object.namespace': 'masakari'})206    def test_sample_decorator(self):207        self.assertEqual(2, len(self.TestNotification.samples))208        self.assertIn('test-update-1.json', self.TestNotification.samples)209        self.assertIn('test-update-2.json', self.TestNotification.samples)210class TestNotificationObjectVersions(test.NoDBTestCase):211    def test_notification_payload_version_depends_on_the_schema(self):212        @base.MasakariObjectRegistry.register_if(False)213        class TestNotificationPayload(notification.NotificationPayloadBase):214            VERSION = '1.0'215            SCHEMA = {216                'field_1': ('source_field', 'field_1'),217                'field_2': ('source_field', 'field_2'),218            }219            fields = {220                'extra_field': fields.StringField(),  # filled by ctor221                'field_1': fields.StringField(),  # filled by the schema222                'field_2': fields.IntegerField(),   # filled by the schema223            }224        checker = fixture.ObjectVersionChecker(225            {'TestNotificationPayload': (TestNotificationPayload,)})226        old_hash = checker.get_hashes(extra_data_func=get_extra_data)...

test_objects.py

Source:test_objects.py

...71            primitive['bar'] = 'old%s' % primitive['bar']72class RandomMixInWithNoFields(object):73    """Used to test object inheritance using a mixin that has no fields."""74    pass75@base.CyborgObjectRegistry.register_if(False)76class TestSubclassedObject(RandomMixInWithNoFields, MyObj):77    fields = {'new_field': fields.StringField()}78class TestObjToPrimitive(test.base.TestCase):79    def test_obj_to_primitive_list(self):80        @base.CyborgObjectRegistry.register_if(False)81        class MyObjElement(base.CyborgObject):82            fields = {'foo': fields.IntegerField()}83            def __init__(self, foo):84                super(MyObjElement, self).__init__()85                self.foo = foo86        @base.CyborgObjectRegistry.register_if(False)87        class MyList(base.ObjectListBase, base.CyborgObject):88            fields = {'objects': fields.ListOfObjectsField('MyObjElement')}89        mylist = MyList()90        mylist.objects = [MyObjElement(1), MyObjElement(2), MyObjElement(3)]91        self.assertEqual([1, 2, 3],92                         [x['foo'] for x in base.obj_to_primitive(mylist)])93    def test_obj_to_primitive_dict(self):94        base.CyborgObjectRegistry.register(MyObj)95        myobj = MyObj(foo=1, bar='foo')96        self.assertEqual({'foo': 1, 'bar': 'foo'},97                         base.obj_to_primitive(myobj))98    def test_obj_to_primitive_recursive(self):99        base.CyborgObjectRegistry.register(MyObj)100        class MyList(base.ObjectListBase, base.CyborgObject):101            fields = {'objects': fields.ListOfObjectsField('MyObj')}102        mylist = MyList(objects=[MyObj(), MyObj()])103        for i, value in enumerate(mylist):104            value.foo = i105        self.assertEqual([{'foo': 0}, {'foo': 1}],106                         base.obj_to_primitive(mylist))107    def test_obj_to_primitive_with_ip_addr(self):108        @base.CyborgObjectRegistry.register_if(False)109        class TestObject(base.CyborgObject):110            fields = {'addr': fields.IPAddressField(),111                      'cidr': fields.IPNetworkField()}112        obj = TestObject(addr='1.2.3.4', cidr='1.1.1.1/16')113        self.assertEqual({'addr': '1.2.3.4', 'cidr': '1.1.1.1/16'},114                         base.obj_to_primitive(obj))115def compare_obj(test, obj, db_obj, subs=None, allow_missing=None,116                comparators=None):117    """Compare a CyborgObject and a dict-like database object.118    This automatically converts TZ-aware datetimes and iterates over119    the fields of the object.120    :param:test: The TestCase doing the comparison121    :param:obj: The CyborgObject to examine122    :param:db_obj: The dict-like database object to use as reference...

fake_objects.py

Source:fake_objects.py

...14#    License for the specific language governing permissions and limitations15#    under the License.16from oslo_utils import versionutils17from cinder import objects18@objects.base.CinderObjectRegistry.register_if(False)19class ChildObject(objects.base.CinderObject):20    VERSION = '1.2'21    fields = {22        'scheduled_at': objects.base.fields.DateTimeField(nullable=True),23        'uuid': objects.base.fields.UUIDField(),24        'text': objects.base.fields.StringField(nullable=True),25        'integer': objects.base.fields.IntegerField(nullable=True),26    }27    def obj_make_compatible(self, primitive, target_version):28        super(ChildObject, self).obj_make_compatible(primitive,29                                                     target_version)30        target_version = versionutils.convert_version_to_tuple(target_version)31        if target_version < (1, 1):32            primitive.pop('text', None)33        if target_version < (1, 2):34            primitive.pop('integer', None)35@objects.base.CinderObjectRegistry.register_if(False)36class ParentObject(objects.base.CinderObject):37    VERSION = '1.1'38    fields = {39        'uuid': objects.base.fields.UUIDField(),40        'child': objects.base.fields.ObjectField('ChildObject', nullable=True),41        'scheduled_at': objects.base.fields.DateTimeField(nullable=True),42    }43    def obj_make_compatible(self, primitive, target_version):44        super(ParentObject, self).obj_make_compatible(primitive,45                                                      target_version)46        target_version = versionutils.convert_version_to_tuple(target_version)47        if target_version < (1, 1):48            primitive.pop('scheduled_at', None)49@objects.base.CinderObjectRegistry.register_if(False)50class ParentObjectList(objects.base.CinderObject, objects.base.ObjectListBase):51    VERSION = ParentObject.VERSION52    fields = {53        'objects': objects.base.fields.ListOfObjectsField('ParentObject'),54    }55class MyHistory(objects.base.CinderObjectVersionsHistory):56    linked_objects = {'ParentObject': 'ParentObjectList'}57    def __init__(self):58        self.versions = ['1.0']59        self['1.0'] = {'ChildObject': '1.0'}60        self.add('1.1', {'ChildObject': '1.1'})61        self.add('1.2', {'ParentObject': '1.0'})62        self.add('1.3', {'ParentObjectList': '1.0'})63        self.add('1.4', {'ParentObject': '1.1'})...

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.