Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _get_all method in fMBT

Best Python code snippet using fMBT_python

test_sqlalchemytypes.py

Source:test_sqlalchemytypes.py

...38    def _get_test_table(self, meta):39        """Returns a new sa.Table() object for this test case."""40    def _add_row(self, **kargs):41        self.engine.execute(self.test_table.insert().values(**kargs))42    def _get_all(self):43        rows_select = self.test_table.select()44        return self.engine.execute(rows_select).fetchall()45    def _update_row(self, **kargs):46        self.engine.execute(self.test_table.update().values(**kargs))47    def _delete_rows(self):48        self.engine.execute(self.test_table.delete())49    def _validate_crud(self, data_field_name, expected=None):50        objs = self._get_all()51        self.assertEqual(len(expected) if expected else 0, len(objs))52        if expected:53            for obj in objs:54                name = obj['id']55                self.assertEqual(expected[name], obj[data_field_name])56class IPAddressTestCase(SqlAlchemyTypesBaseTestCase):57    def _get_test_table(self, meta):58        return sa.Table(59            'fakeipaddressmodels',60            meta,61            sa.Column('id', sa.String(36), primary_key=True, nullable=False),62            sa.Column('ip', sqlalchemytypes.IPAddress))63    def _validate_ip_address(self, data_field_name, expected=None):64        objs = self._get_all()65        self.assertEqual(len(expected) if expected else 0, len(objs))66        if expected:67            for obj in objs:68                name = obj['id']69                self.assertEqual(expected[name], obj[data_field_name])70    def _test_crud(self, ip_addresses):71        ip = netaddr.IPAddress(ip_addresses[0])72        self._add_row(id='fake_id', ip=ip)73        self._validate_ip_address(data_field_name='ip',74                                  expected={'fake_id': ip})75        ip2 = netaddr.IPAddress(ip_addresses[1])76        self._update_row(ip=ip2)77        self._validate_ip_address(data_field_name='ip',78                                  expected={'fake_id': ip2})79        self._delete_rows()80        self._validate_ip_address(data_field_name='ip', expected=None)81    def test_crud(self):82        ip_addresses = ["10.0.0.1", "10.0.0.2"]83        self._test_crud(ip_addresses)84        ip_addresses = [85                        "2210::ffff:ffff:ffff:ffff",86                        "2120::ffff:ffff:ffff:ffff"87                       ]88        self._test_crud(ip_addresses)89    def test_wrong_type(self):90        self.assertRaises(exception.DBError, self._add_row,91                          id='fake_id', ip="")92        self.assertRaises(exception.DBError, self._add_row,93                          id='fake_id', ip="10.0.0.5")94    def _test_multiple_create(self, entries):95        reference = {}96        for entry in entries:97            ip = netaddr.IPAddress(entry['ip'])98            name = entry['name']99            self._add_row(id=name, ip=ip)100            reference[name] = ip101        self._validate_ip_address(data_field_name='ip', expected=reference)102        self._delete_rows()103        self._validate_ip_address(data_field_name='ip', expected=None)104    def test_multiple_create(self):105        ip_addresses = [106                        {'name': 'fake_id1', 'ip': "10.0.0.5"},107                        {'name': 'fake_id2', 'ip': "10.0.0.1"},108                        {'name': 'fake_id3',109                         'ip': "2210::ffff:ffff:ffff:ffff"},110                        {'name': 'fake_id4',111                         'ip': "2120::ffff:ffff:ffff:ffff"}112                       ]113        self._test_multiple_create(ip_addresses)114class CIDRTestCase(SqlAlchemyTypesBaseTestCase):115    def _get_test_table(self, meta):116        return sa.Table(117            'fakecidrmodels',118            meta,119            sa.Column('id', sa.String(36), primary_key=True, nullable=False),120            sa.Column('cidr', sqlalchemytypes.CIDR)121        )122    def _get_one(self, value):123        row_select = self.test_table.select().\124            where(self.test_table.c.cidr == value)125        return self.engine.execute(row_select).first()126    def _update_row(self, key, cidr):127        self.engine.execute(128            self.test_table.update().values(cidr=cidr).129            where(self.test_table.c.cidr == key))130    def test_crud(self):131        cidrs = ["10.0.0.0/24", "10.123.250.9/32", "2001:db8::/42",132                 "fe80::21e:67ff:fed0:56f0/64"]133        for cidr_str in cidrs:134            cidr = netaddr.IPNetwork(cidr_str)135            self._add_row(id=uuidutils.generate_uuid(), cidr=cidr)136            obj = self._get_one(cidr)137            self.assertEqual(cidr, obj['cidr'])138            random_cidr = netaddr.IPNetwork(tools.get_random_cidr())139            self._update_row(cidr, random_cidr)140            obj = self._get_one(random_cidr)141            self.assertEqual(random_cidr, obj['cidr'])142        objs = self._get_all()143        self.assertEqual(len(cidrs), len(objs))144        self._delete_rows()145        objs = self._get_all()146        self.assertEqual(0, len(objs))147    def test_wrong_cidr(self):148        wrong_cidrs = ["10.500.5.0/24", "10.0.0.1/40", "10.0.0.10.0/24",149                       "cidr", "", '2001:db8:5000::/64', '2001:db8::/130']150        for cidr in wrong_cidrs:151            self.assertRaises(exception.DBError, self._add_row,152                              id=uuidutils.generate_uuid(), cidr=cidr)153class MACAddressTestCase(SqlAlchemyTypesBaseTestCase):154    def _get_test_table(self, meta):155        return sa.Table(156            'fakemacaddressmodels',157            meta,158            sa.Column('id', sa.String(36), primary_key=True, nullable=False),159            sa.Column('mac', sqlalchemytypes.MACAddress)160        )161    def _get_one(self, value):162        row_select = self.test_table.select().\163            where(self.test_table.c.mac == value)164        return self.engine.execute(row_select).first()165    def _get_all(self):166        rows_select = self.test_table.select()167        return self.engine.execute(rows_select).fetchall()168    def _update_row(self, key, mac):169        self.engine.execute(170            self.test_table.update().values(mac=mac).171            where(self.test_table.c.mac == key))172    def _delete_row(self):173        self.engine.execute(174            self.test_table.delete())175    def test_crud(self):176        mac_addresses = ['FA:16:3E:00:00:01', 'FA:16:3E:00:00:02']177        for mac in mac_addresses:178            mac = netaddr.EUI(mac)179            self._add_row(id=uuidutils.generate_uuid(), mac=mac)180            obj = self._get_one(mac)181            self.assertEqual(mac, obj['mac'])182            random_mac = netaddr.EUI(net.get_random_mac(183                ['fe', '16', '3e', '00', '00', '00']))184            self._update_row(mac, random_mac)185            obj = self._get_one(random_mac)186            self.assertEqual(random_mac, obj['mac'])187        objs = self._get_all()188        self.assertEqual(len(mac_addresses), len(objs))189        self._delete_rows()190        objs = self._get_all()191        self.assertEqual(0, len(objs))192    def test_wrong_mac(self):193        wrong_macs = ["fake", "", -1,194                      "FK:16:3E:00:00:02",195                      "FA:16:3E:00:00:020"]196        for mac in wrong_macs:197            self.assertRaises(exception.DBError, self._add_row,198                              id=uuidutils.generate_uuid(), mac=mac)199class TruncatedDateTimeTestCase(SqlAlchemyTypesBaseTestCase):200    def _get_test_table(self, meta):201        return sa.Table(202            'timetable',203            meta,204            sa.Column('id', sa.String(36), primary_key=True, nullable=False),205            sa.Column('thetime', sqlalchemytypes.TruncatedDateTime)206        )207    def test_microseconds_truncated(self):208        tstamp = timeutils.utcnow()209        tstamp_low = tstamp.replace(microsecond=111111)210        tstamp_high = tstamp.replace(microsecond=999999)211        self._add_row(id=1, thetime=tstamp_low)212        self._add_row(id=2, thetime=tstamp_high)213        rows = self._get_all()214        self.assertEqual(2, len(rows))...

kubernetes.py

Source:kubernetes.py

...20        config.update({21            'method':   'Threaded',22        })23        return config24    def _get_all(self, kind, apigroup='api/v1'):25        """26        Get a list of all kubernetes objects of kind, across namespaces27        """28        # FIXME: Make the URL configurable29        url = 'http://localhost:8080/{apigroup}/{kind}'.format(kind=kind, apigroup=apigroup)30        return requests.get(31            url,32            headers={'User-Agent': 'Diamond Kubernetes Collector/1.0'}33        ).json()['items']34    def collect(self):35        # Number of nodes36        nodes = self._get_all('nodes')37        self.publish('nodes.all', len(nodes))38        # Number of deployments39        deployments = self._get_all('deployments', 'apis/extensions/v1beta1')40        self.publish('deployments.all', len(deployments))41        # Number of services42        services = self._get_all('services')43        self.publish('services.all', len(services))44        # Number of namespaces45        namespaces = self._get_all('namespaces')46        self.publish('namespaces.all', len(namespaces))47        # Pod stats:48        #  - Total number of pods49        #  - Pods in various states50        #  - Number of namespaces with at least one pod in them51        pods = self._get_all('pods')52        pod_phases = {}53        active_namespaces = set()54        for pod in pods:55            phase = pod['status']['phase'].lower()56            namespace = pod['metadata']['namespace']57            if namespace not in active_namespaces:58                active_namespaces.add(namespace)59            pod_phases[phase] = pod_phases.get(phase, 0) + 160        self.publish('pods.all', len(pods))61        for phase, count in pod_phases.items():62            self.publish('pods.' + phase, count)...

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.