How to use _cat method in fMBT

Best Python code snippet using fMBT_python

test_catalogs.py

Source:test_catalogs.py

1from ...data_sources.local import TEST_ROOT2from .. import LcCatalog3from ...lc_resource import LcResource4from ...catalog_query import CatalogQuery, READONLY_INTERFACE_TYPES5from lcatools.archives.tests import basic_archive_src6import os7import unittest8from shutil import rmtree9uslci_fg = LcResource('test.uslci', '/data/LCI/USLCI/USLCI_Processes_ecospold1.zip', 'EcospoldV1Archive',10                      interfaces='inventory',11                      priority=40,12                      static=False,13                      prefix='USLCI_Processes_ecospold1/USLCI_Processes_ecospold1')14uslci_fg_dup = LcResource('test.uslci', '/data/LCI/USLCI/USLCI_Processes_ecospold1.zip', 'EcospoldV1Archive',15                          interfaces='inventory',16                          priority=40,17                          static=False,18                          prefix='USLCI_Processes_ecospold1/USLCI_Processes_ecospold1')19uslci_fg_bad = LcResource('test.uslci', '/data/LCI/USLCI/junk.zip', 'EcospoldV1Archive',20                          interfaces='inventory',21                          priority=40,22                          static=False,23                          prefix='USLCI_Processes_ecospold1/USLCI_Processes_ecospold1')24uslci_bg = LcResource('test.uslci.allocated', '/data/GitHub/lca-tools-datafiles/catalogs/uslci_clean_allocated.json.gz',25                      'json',26                      interfaces=READONLY_INTERFACE_TYPES,27                      priority=90,28                      static=True)29test_resource = LcResource('test.basic', basic_archive_src, 'json',30                           interfaces=READONLY_INTERFACE_TYPES)31class LcCatalogFixture(unittest.TestCase):32    @classmethod33    def setUpClass(cls):34        cls._cat = LcCatalog.make_tester()35        cls._cat.add_resource(uslci_fg)36        cls._cat.add_resource(uslci_bg)37        cls._cat.add_resource(test_resource)38    @classmethod39    def tearDownClass(cls):40        rmtree(TEST_ROOT)41    def test_resolver_index(self):42        self.assertSetEqual({r for r in self._cat.references}, {'local.qdb', 'test.uslci', 'test.uslci.allocated',43                                                                'test.basic'})44    def test_priority(self):45        q = CatalogQuery('test.uslci', catalog=self._cat)46        p = q.get('Acetic acid, at plant')47        self.assertEqual(p.origin, 'test.uslci')48    def test_inventory(self):49        q = self._cat.query('test.uslci')50        inv = [x for x in q.inventory('Acetic acid, at plant')]51        self.assertEqual(len(inv), 21)52    @unittest.skip53    def test_find_source(self):54        """55        Need to determine a set of testing conditions that ensure the resolver.get_resource() works properly56        :return:57        """58        pass59    def test_add_delete_resource_1(self):60        """61        This adds a resource62        :return:63        """64        r = self._cat.new_resource('test.my.dummy', '/dev/null', 'LcArchive')65        self.assertIn('basic', r.interfaces)66        self.assertIn('test.my.dummy', self._cat.references)67        self.assertNotIn('test.my.doofus', self._cat.references)68    def test_add_delete_resource_2(self):69        """70        This deletes the resource71        :return:72        """73        r = self._cat.get_resource('test.my.dummy')74        self.assertEqual(r.source, '/dev/null')75        self._cat.delete_resource(r)76        self.assertNotIn('test.my.dummy', self._cat.references)77        self.assertFalse(os.path.exists(os.path.join(self._cat.resource_dir, r.reference)))78    def test_has_resource(self):79        """80        If a resource matches one that exists, has_resource should return True81        :return:82        """83        self.assertTrue(self._cat.has_resource(uslci_fg_dup))84        self.assertFalse(self._cat.has_resource(uslci_fg_bad))85    def test_local_resource(self):86        inx = self._cat.index_ref(test_resource.reference)87        self.assertIn(inx, self._cat.references)88        res = self._cat.get_resource(inx)89        self.assertTrue(self._cat.has_resource(res))90        self.assertTrue(res.source.startswith('$CAT_ROOT'))91        abs_path = self._cat.abs_path(res.source)92        self.assertEqual(self._cat._localize_source(abs_path), res.source)93        self.assertEqual(self._cat._index_file(test_resource.source), abs_path)94        self.assertTrue(os.path.exists(abs_path))95        self._cat.delete_resource(res, delete_source=True)96        self.assertFalse(os.path.exists(abs_path))97if __name__ == '__main__':...

ElasticSearchEnumerator.py

Source:ElasticSearchEnumerator.py

1#!/usr/bin/env python32import requests3import sys4import json5import re6import itertools7def elk():8    ip =  sys.argv[1]9    port = sys.argv[2]10    protocol = sys.argv[3]11    f_host = f'{protocol}://{ip}:{port}'12    url = re.sub('\?|\!|\'|\n|\'|\;', '', f_host)13    catinfo = [14    '/_cat/segments',15    '/_cat/shards',16    '/_cat/repositories',17    '/_cat/recovery',18    '/_cat/plugins',19    '/_cat/pending_tasks',20    '/_cat/nodes',21    '/_cat/tasks',22    '/_cat/templates',23    '/_cat/thread_pool',24    '/_cat/ml/trained_models',25    '/_cat/transforms/_all',26    '/_cat/aliases',27    '/_cat/allocation',28    '/_cat/ml/anomaly_detectors',29    '/_cat/count',30    '/_cat/ml/data_frame/analytics',31    '/_cat/ml/datafeeds',32    '/_cat/fielddata',33    '/_cat/health',34    '/_cat/indices',35    '/_cat/master',36    '/_cat/nodeattrs',37    '/_cat/nodes']38    clusterinfo = [39    '/_cluster/allocation/explain',40    '/_cluster/settings','/_cluster/health',41    '/_cluster/state',42    '/_cluster/stats',43    '/_cluster/pending_tasks',44    '/_nodes',45    '/_nodes/usage',46    '/_nodes/hot_threads',47    '/_nodes/stats','/_tasks',48    '/_remote/info']49    securityinfo = [50    '/_security/user',51    '/_security/privilege',52    '/_security/role_mapping',53    '/_security/role',54    '/_security/api_key']55    paths = list(itertools.chain(catinfo,clusterinfo,securityinfo))56    for path in paths:57        try:58            f_url = f'{url}{path}?format=json&pretty'59            r = requests.request('GET',f_url, verify=False)60            decoded_r = json.loads(r.text)61            print('\n\n', f_url,'\n\n', json.dumps(decoded_r, sort_keys=True, indent=3), end='\n\n========================================================\n\n')62        except:63            print('Could not execute the following request.')64            sys.exit()65if __name__ == '__main__':            ...

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.