How to use get_reporters method in Testify

Best Python code snippet using Testify_python

miner.py

Source:miner.py

...7from miner.models import News, Log8class Command(BaseCommand):9    def handle(self, *args, **options):10        sys.path = ['../reporters'] + sys.path11        def get_reporters(self):12            reporters = []13            reporters_files = os.listdir('../reporters')14            print(reporters_files)15            for r in reporters_files:16                print(r)17                try:18                    if '__' not in r and r not in ('Credentials.py',19                                                   'utils.py',20                                                   'ReporterError.py'):21                        reporters.append(__import__(r[:-3]))22                except ModuleNotFoundError as e:23                    make_log(self, 'ERROR', e)24            return reporters25        def make_log(self, level, message):26            lo = Log(level_name=level, message=message)27            lo.save()28        print(1)29        reporters = get_reporters(self)30        print(2)31        make_log(self, 'INFO', 'start mining')32        with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:33            future_to_name = {executor.submit(r.get_news):34                              r.name for r in reporters}35            for future in concurrent.futures.as_completed(future_to_name):36                name = future_to_name[future]37                try:38                    data = future.result()39                    if isinstance(data, dict):40                        if len(News.objects.filter(text=data['text'])) == 0:41                            n = News(reporter_name=name,42                                     title=data['title'],43                                     text=data['text'])...

test_reporter.py

Source:test_reporter.py

...9        self.assertEqual(len(ReporterManager.get_reporter_list()), 1)10        11    def test_update_should_register_new_reporter_with_key(self):12        ReporterManager.update("Karma", "Projectx", success=0, failures=0, errors=20)13        self.assertEqual(ReporterManager.get_reporters().keys()[0], "Karma_Projectx")14    def test_get_reporters_should_return_all_reporter_with_keys(self):15        ReporterManager.update("Karma", "Projectx", success=0, failures=0, errors=20)16        self.assertEqual(ReporterManager.get_reporters().keys()[0], "Karma_Projectx")17    def test_REPORTER_LIST_should_contains_reporter_instance(self):18        ReporterManager.update("Karma", "Projectx", success=0, failures=0, errors=20)19        self.assertEqual(ReporterManager.get_reporters().values()[0].runner, "Karma")20        self.assertTrue(isinstance(ReporterManager.get_reporters().values()[0], Reporter))21    def test_REPORTER_LIST_should_contains_total_of_all_reporter(self):22        ReporterManager.update("Karma", "Projectx", success=3, failures=0, errors=20)23        ReporterManager.update("Karma", "Projecty", success=4, failures=5, errors=1)24        self.assertEqual(ReporterManager.total_success(), 7)25        self.assertEqual(ReporterManager.total_failures(), 5)26        self.assertEqual(ReporterManager.total_errors(), 21)27class ReporterTest(TestCase):28    def test_parser_should_register_key_value_element_from_stream(self):29        rep = Reporter("Karma", "Acme", None, None, 0 )30        self.assertFalse(rep.is_valid())31        del rep32        33        rep = Reporter("Karma", "Acme", 3, None, 0 )34        self.assertFalse(rep.is_valid())...

newsmaker.py

Source:newsmaker.py

...7reporters_location = '/reporters'8string_path = str(package_root_directory) + reporters_location9sys.path.append(string_path)10reporter_files = os.listdir(string_path)11def get_reporters():12	reporters = {}13	for i in reporter_files:14		if os.path.isfile(string_path+'/'+i) and '__' not in i:15			file_without_extension = Path(i).stem16			m = __import__(file_without_extension)17			if file_without_extension == 'covid_reporter' and file_without_extension\18												not in reporters:19				reporters['covid_reporter'] = m	20			else:21				reporters['currency_reporter'] = m22			print(m)23	return reporters24async def newsmaker_run():25	"""26	here all reporters will be run asynchroniously27	"""28	reporters = get_reporters()29	covid_data = reporters['covid_reporter'].get_news()30	currency_data = reporters['currency_reporter'].get_news()31	from flask_conf import REPORTERS_CONNECTION_ERROR as error32	if covid_data!=error and currency_data!=error:33		task1 = asyncio.create_task(reporters['covid_reporter'].covid_news(covid_data), name = 'covid_news')34		task2 = asyncio.create_task(reporters['currency_reporter'].currency_news(currency_data), name = 'currency_news')35		results = await asyncio.gather(task1,task2)36	else:37		results = error38		return results...

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.