KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use import_instance method in localstack

Best Python code snippet using localstack_python

test_import.py

Source:test_import.py

1# coding=utf-82from django.core.files.uploadedfile import SimpleUploadedFile3from django.db import models, transaction4from djangoplicity.contacts.admin import ImportAdmin5from djangoplicity.contacts.api.serializers import ImportSerializer6from djangoplicity.contacts.models import Contact, ContactGroup, ImportTemplate, ImportMapping, \7    ImportSelector, ImportGroupMapping, DataImportError, Import, Deduplication8from tests.base import BasicTestCase, TestDeduplicationBase9from tests.factories import factory_import_selector, factory_request_data, factory_deduplication10from djangoplicity.contacts.importer import CSVImporter, ExcelImporter11from djangoplicity.contacts.tasks import prepare_import12import json13from django.core import mail14try:15    from mock import patch, MagicMock16except ImportError:17    from unittest.mock import patch, MagicMock18class TestImportTemplate(BasicTestCase):19    """20    Test template defines how a CSV or Excel file should be imported into the contacts model.21    It supports mapping columns to contacts fields.22    """23    def test_template_get_data_from_csv_file(self):24        """25        Test open csv import file26        """27        filepath = './tests/data_sources/contacts.csv'28        template = ImportTemplate.objects.get(name='TEST Contacts all')29        importer = template.get_importer('./tests/data_sources/contacts.csv')30        self.assertIsInstance(importer, CSVImporter)31        self.assertIsNotNone(importer)32        self.assertEqual(len(importer), 100)33        self.assertEqual(len(importer.cols), 17)34        self.assertEqual(str(template), 'TEST Contacts all')35        row_0_email = importer.row(0)['Email']36        row_1_email = importer.row(1)['Email']37        row_2_email = importer.row(2)['Email']38        row_3_email = importer.row(3)['Email']39        row_4_email = importer.row(4)['Email']40        self.assertEqual(row_0_email, 'adam70@yahoo.com')41        self.assertEqual(row_1_email, 'pgrant@green-shaw.biz')42        self.assertEqual(row_2_email, 'bradley00@lopez.com')43        self.assertEqual(row_3_email, 'joneskristen@hotmail.com')44        self.assertEqual(row_4_email, 'greenryan@hernandez.com')45    def test_template_get_data_from_xls_file(self):46        """47        Test open xls import file48        """49        filepath = './tests/data_sources/contacts.xls'50        template = ImportTemplate.objects.get(name='TEST Contacts all')51        template.clear_selector_cache()52        # clear caches53        template.clear_mapping_cache()54        importer = template.get_importer(filepath)55        self.assertIsInstance(importer, ExcelImporter)56        self.assertIsNotNone(importer)57        self.assertEqual(len(importer), 100)58        self.assertEqual(len(importer.cols), 17)59        self.assertEqual(len(importer.header()), 17)60        self.assertEqual(str(template), 'TEST Contacts all')61        # Get specific row62        row = importer.row(0)63        self.assertEqual(row, {64            'City': 'Lauratown',65            'Country': 'Germany',66            'Department': 'LLC',67            'Email': 'michael11@cox-ayala.com',68            'First name': 'Christopher',69            'Groups': 'Public NL, Public NL-invite, Unknown Group',70            'Language': 'en',71            'Last name': 'Wade',72            'Organization': 'Johnson, Smith and Mclean',73            'Phone': '890.067.9460',74            'Position': 'Product designer',75            'Region/State': 'Schleswig-Holstein',76            'Street 1': '36310 Juan Fords\nNew Sabrina, NY 54546',77            'Street 2': '35304 Carlson Branch\nMeyertown, NJ 71989',78            'Title': 'Mrs.',79            'Website': 'https://www.smith.org/',80            'Zipcode': 15418.081        })82        # test import data83        data = template.extract_data(filepath)84        first_row = data.next()85        second_row = data.next()86        self.assertEqual(type(first_row), dict)87        self.assertEqual(second_row, {88            'city': u'Francisburgh',89            'country': 7,90            'department': u'Ltd',91            'email': u'tracy77@hotmail.com',92            'first_name': u'Angela',93            'groups': [1, 2],94            'language': u'en',95            'last_name': u'Spencer',96            'organisation': u'Alexander, Rivera and Jones',97            'phone': u'752.462.1748x89014',98            'position': u'Accommodation manager',99            'region': 550,100            'street_1': u'005 Huber Crest\nPort Marychester, NJ 69633',101            'street_2': u'31342 Katrina Village\nWest Cathybury, WA 38502',102            'title': u'Mrs.',103            'website': u'http://www.carter-king.com/',104            'zip': 78583105        })106        # Test preview table107        mappings_rows, data_table = template.preview_data(filepath)108        # test table header109        self.assertEqual(len(data_table), 100)110        self.assertEqual(len(mappings_rows), 18)111        # test the first table row112        self.assertEqual(type(data_table[0]), list)113        self.assertEqual(len(data_table[0]), 18)114    def test_bad_file_format(self):115        # Raise exception with invalid file format116        filepath = './tests/data_sources/contacts.pdf'117        template = ImportTemplate.objects.get(name='TEST Contacts all')118        self.assertRaises(DataImportError, template.get_importer, filepath)119    def test_bad_column_in_file(self):120        # Raise bad column exception121        filepath = './tests/data_sources/bad_column.xls'122        template = ImportTemplate.objects.get(name='TEST Contacts all')123        row, data = template.preview_data(filepath)124        count_errors = 0125        for row in data:126            if 'ERROR' in row:127                count_errors += 1128        self.assertEqual(10, count_errors)129        self.assertIsNotNone(data)130    def test_template_review_data(self):131        """132        Review template data133        """134        filepath = './tests/data_sources/contacts.xls'135        template = ImportTemplate.objects.get(name='TEST Contacts all')136        # Review data137        mapping, imported, new, duplicates = template.review_data(filepath, {}, {})138        self.assertEqual(len(mapping), 17)139        self.assertEqual(imported, [])140        self.assertEqual(len(new), 100)141        self.assertEqual(duplicates, [])142    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)143    def test_direct_import_data_from_xls(self, contact_group_check_mock):144        """145        Import data from xls file146        """147        with self.settings(SITE_ENVIRONMENT='prod'):148            before_contacts = Contact.objects.count()149            filepath = './tests/data_sources/contacts.xls'150            template = ImportTemplate.objects.get(name='TEST Contacts all')151            template.direct_import_data(filepath)152            after_contacts = Contact.objects.count()153            self.assertEqual(before_contacts, 0)154            self.assertEqual(after_contacts, 100)155            self.assertEqual(contact_group_check_mock.call_count, 100)156    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)157    def test_import_duplication_contacts_from_xml(self, contact_group_check_mock):158        """159        Import data from xls file160        """161        with self.settings(SITE_ENVIRONMENT='prod'):162            # Creating contacts for the first time163            filepath = './tests/data_sources/contacts.xls'164            template = ImportTemplate.objects.get(name='TEST Contacts all')165            template.direct_import_data(filepath)166            # Preparing to import duplicate contacts167            duplication_filepath = './tests/data_sources/duplicates.xls'168            duplicates = template.prepare_import(duplication_filepath)169            self.assertEqual(len(duplicates), 10)170class TestImportMapping(BasicTestCase):171    def test_import_mapping_methods(self):172        """173        Testing mapping fields174        """175        template = ImportTemplate.objects.get(name='TEST Contacts all')176        title_mapping = template.importmapping_set.first()177        title_mapping.clear_groupmap_cache()178        title_mapping.group_separator = ';'179        title_mapping.save()180        django_field = title_mapping.get_django_field()181        mapping = ImportMapping.objects.get(field= 'title')182        self.assertIsInstance(django_field, models.CharField)183        self.assertIsNone(title_mapping._groupmap_cache,)184        self.assertEqual(mapping.group_separator, ';')185        self.assertEqual(title_mapping.get_field(), 'title')186        self.assertEqual(template.importmapping_set.all().count(), 17)187    def test_import_mapping_country_and_regions(self):188        template = ImportTemplate.objects.get(name='TEST Contacts all')189        country_mapping = template.importmapping_set.get(field='country')190        # Test unicode cast191        # find country by id192        result1 = country_mapping.get_country_value(b'de')193        result2 = country_mapping.get_country_value('de')194        # find country by similar text and ISO EXPANSIONS195        result3 = country_mapping.get_country_value('ermany')196        result4 = country_mapping.get_country_value('united states')197        # Not find country and regions198        result5 = country_mapping.get_country_value('gArmany')199        result6 = country_mapping.get_country_value('DI')200        unknown = country_mapping.get_region_value(None)201        self.assertIsNotNone(result1)202        self.assertIsNotNone(result2)203        self.assertIsNotNone(result3)204        self.assertIsNotNone(result4)205        self.assertIsNone(result5)206        self.assertIsNone(result6)207        self.assertIsNone(unknown)208class TestImportSelectorModel(BasicTestCase):209    def test_import_selector_creation(self):210        template = ImportTemplate.objects.get(name='TEST Contacts all')211        count_before = ImportSelector.objects.count()212        for header, value in [('Country', 'Germany'), ('Region', 'Berlin')]:213            selector = factory_import_selector(template, {214                'header': header,215                'value': value,216                'case_sensitive': False217            })218            selector.save()219        count_after = ImportSelector.objects.count()220        self.assertEqual(count_before + 2, count_after)221    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)222    def test_import_direct_with_selectors(self, contact_group_check_mock):223        """224        Test import with column selector, where Country is Germany225        """226        with self.settings(SITE_ENVIRONMENT='prod'):227            template = ImportTemplate.objects.get(name='TEST Contacts all')228            selector = factory_import_selector(template, {229                'header': 'Country',230                'value': 'Germany',231                'case_sensitive': False232            })233            selector.save()234            filepath = './tests/data_sources/contacts.xls'235            template = ImportTemplate.objects.get(name='TEST Contacts all')236            template.direct_import_data(filepath)237            contact_count = Contact.objects.count()238            self.assertEqual(contact_count, 51)239class TestImportGroupMapping(BasicTestCase):240    def test_Group_mapping_creation(self):241        """242        Add a new group mapping to template243        """244        template = ImportTemplate.objects.get(name='TEST Contacts all')245        mapping_groups = template.importmapping_set.get(field='groups')246        contact_group = ContactGroup.objects.get(name='Public NL-unsubscribed')247        g_mapping = ImportGroupMapping(mapping=mapping_groups,248                                       value='Public NL-unsubscribed', group=contact_group)249        g_mapping.save()250        count_g_mapping = ImportGroupMapping.objects.count()251        self.assertEqual(count_g_mapping, 3)252class TestImportModel(BasicTestCase):253    """254    Test stores an excel file and selects which import template to use when importing255    the data.256    """257    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)258    def test_import_creation(self, contactgroup_change_check_mock):259        # Create an import model instance260        with self.settings(SITE_ENVIRONMENT='prod'), open("./tests/data_sources/contacts.xls") as contacts_file:261            import_count_before = Import.objects.count()262            contacts_count_before = Contact.objects.count()263            template = ImportTemplate.objects.get(name='TEST Contacts all')264            data = {265                "template": template,266                "data_file": SimpleUploadedFile(contacts_file.name, bytes(contacts_file.read())),267            }268            instance = Import(**data)269            instance.save()270            imported = instance.direct_import_data()271            import_count_after = Import.objects.count()272            contacts_count_after = Contact.objects.count()273            columns, rows = instance.preview_data()274            self.assertTrue(imported)275            self.assertEqual(contacts_count_after, contacts_count_before + 100)276            self.assertEqual(import_count_before+1, import_count_after)277            self.assertEqual(18, len(columns))278            self.assertEqual(100, len(rows))279            self.assertIsInstance(instance, Import)280    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)281    def test_import_serializer(self, contactgroup_change_check_mock):282        with self.settings(SITE_ENVIRONMENT='prod'), open("./tests/data_sources/contacts.xls") as contacts_file:283            template = ImportTemplate.objects.get(name='TEST Contacts all')284            data = {285                "template": template,286                "data_file": SimpleUploadedFile(contacts_file.name, bytes(contacts_file.read())),287            }288            instance = Import(**data)289            instance.save()290            instance.direct_import_data()291            # Serializer methods292            serializer = ImportSerializer(instance)293            obj_serialize = serializer.data294            self.assertEqual(obj_serialize['status'], 'imported')295    def test_import_deletion(self):296        with self.settings(SITE_ENVIRONMENT='prod'), open("./tests/data_sources/contacts.xls") as contacts_file:297            template = ImportTemplate.objects.get(name='TEST Contacts all')298            data = {299                "template": template,300                "data_file": SimpleUploadedFile(contacts_file.name, bytes(contacts_file.read())),301            }302            instance = Import(**data)303            instance.save()304            instance.delete()305            with self.assertRaises(Import.DoesNotExist):306                instance.refresh_from_db()307class TestImportMethodsModel(BasicTestCase):308    """309    Test Import methods in import model310    """311    import_instance = None312    response = None313    template = None314    def setUp(self):315        super(TestImportMethodsModel, self).setUp()316        # set initial contacts to test duplicates behavior317        self.template = ImportTemplate.objects.get(name='TEST Contacts all')318        self._import_initial_contacts()319        self._import_contacts()320    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)321    def _import_initial_contacts(self, contactgroup_change_check):322        # Create an import model instance323        with self.settings(SITE_ENVIRONMENT='prod'), open("./tests/data_sources/duplicates.xls") as contacts_file:324            data = {325                "template": self.template,326                "data_file": SimpleUploadedFile(contacts_file.name, bytes(contacts_file.read())),327            }328            instance = Import(**data)329            instance.save()330            instance.direct_import_data()331    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)332    def _import_contacts(self, contactgroup_change_check):333        # Create an import model instance334        with self.settings(SITE_ENVIRONMENT='prod'), open("./tests/data_sources/contacts.xls") as contacts_file:335            data = {336                "template": self.template,337                "data_file": SimpleUploadedFile(contacts_file.name, bytes(contacts_file.read())),338            }339            self.import_instance = Import(**data)340            self.import_instance.save()341    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)342    def test_prepare_to_import(self, contactgroup_change_check_mock):343        with self.settings(SITE_ENVIRONMENT='prod'):344            prepare_import(self.import_instance.id, 'jhondoe@mail.com')345            response = self.import_instance.prepare_import()346            duplicates = json.loads(self.import_instance.duplicate_contacts)347            # Generate a preview of the data mapping for this import. The data348            # will be used as the basis for the import.349            headers, rows = self.import_instance.preview_data()350            self.assertEqual(len(mail.outbox), 1)351            self.assertEqual(mail.outbox[0].subject, 'Import %s (%s) ready for review' %352                             (self.import_instance.pk, self.import_instance.data_file))353            self.assertTrue(response)354            self.assertEqual(len(duplicates), 10)355            self.assertEqual(len(headers), 18)356            self.assertEqual(len(rows), 100)357    def test_review_data(self):358        with self.settings(SITE_ENVIRONMENT='prod'):359            mapping, imported, new, duplicates = self.import_instance.review_data()360            self.assertEqual(len(mapping), 17)361            self.assertEqual(len(new), 100)362    @patch('djangoplicity.contacts.tasks.contactgroup_change_check.apply_async', raw=True)363    def test_import_data(self, contactgroup_change_check_mock):364        with self.settings(SITE_ENVIRONMENT='prod'):365            data = factory_request_data()366            mapping, imported, new, duplicates = self.import_instance.review_data()367            import_contacts = ImportAdmin.clean_import_data(data)368            response = self.import_instance.import_data(import_contacts)369            self.import_instance.save()370            self.assertEqual(self.import_instance.status, 'new')371            self.assertTrue(response)372            # test template review data for mapping[1] - phone373            contact = Contact.objects.get(email='jhondoe@cox-ayala.com')374            field_data = self.template._get_review_form_data(mapping[1], contact)375            self.assertEqual(field_data['name'], 'phone')376            self.assertEqual(field_data['value'], '890.067.9460')377            # test template review data for mapping[2] - email378            field_data = self.template._get_review_form_data(mapping[2])379            self.assertEqual(field_data['name'], 'email')380            self.assertIsNone(field_data['value'])381class TestDeduplicationModel(TestDeduplicationBase):382    def test_deduplication_creation(self):383        instance = factory_deduplication({})384        instance.save()385        # Look for 10 duplicates contacts in the given groups386        response = instance.run()387        duplicates, total_duplicates = instance.review_data()388        duplicate_contacts = json.loads(instance.duplicate_contacts)389        self.assertTrue(response)390        self.assertEqual(len(duplicate_contacts), 10)391        self.assertEqual(len(duplicates), 10)392        self.assertEqual(total_duplicates, 10)...

main.py

Source:main.py

1# Import used to run pytest, without complete path, test doesn't work2# Please comment the last 4 imports34# from quake_log_parser.reports.by_death_mean_report import ByDeathMeanReport5# from quake_log_parser.reports.complete_report import CompleteReport6# from quake_log_parser.reports.default_report import DefaultReport7# from quake_log_parser.importer.txt_importer import TxtImporter89# Import used to run the program (main.py), with complete path, program10# doesn't work11# Please comment the first 4 imports1213from reports.by_death_mean_report import ByDeathMeanReport14from reports.complete_report import CompleteReport15from reports.default_report import DefaultReport16from importer.txt_importer import TxtImporter1718import sys192021menu = """22Select one of the options below:23 0 - Default report (total kills, players, kills ranking);24 1 - Report by death mean (kills by means);25 2 - Complete report;26 3 - Dictionary/JSON of Default Report;27 4 - Dictionary/JSON of By Death Mean Report;28 5 - Exit.\n29 -> """3031import_instance = TxtImporter3233file_log = import_instance.import_data("./quake_log_parser/data/qgames.log")34joined_info_to_divide = import_instance.prepare_to_divide_by_match(file_log)35init_lines = joined_info_to_divide[0]36total_matches = joined_info_to_divide[1]37log_by_match = import_instance.divide_log_by_match(file_log, init_lines, total_matches)38ranking = import_instance.create_ranking_dict(log_by_match, total_matches)39import_instance.add_score_to_ranking(ranking, log_by_match, total_matches)40ranking_by_death = import_instance.create_ranking_by_death_dict(log_by_match, total_matches)414243def option_0():44    report_instance = DefaultReport45    default_report = report_instance.generate(ranking)46    return print(default_report)474849def option_1():50    report_instance = ByDeathMeanReport51    by_death_report = report_instance.generate(ranking_by_death)52    return print(by_death_report)535455def option_2():56    report_instance = CompleteReport57    complete_report = report_instance.generate(ranking, ranking_by_death)58    return print(complete_report)596061def option_3():62    return(print(f"\n{ranking}\n"))636465def option_4():66    return(print(f"\n{ranking_by_death}\n"))676869def option_5():70    return print("\nTerminating script!\n")717273options = {74    "0": option_0,75    "1": option_1,76    "2": option_2,77    "3": option_3,78    "4": option_4,79    "5": option_5,80}818283if __name__ == "__main__":84    sys.stdout.write(menu)85    option = input()86    if option in options.keys():87        options[option]()88    else:89        sys.stderr.write("Invalid option\n")909192# Method used only for testing (tests/test_main.py)93def show_menu():94    sys.stdout.write(menu)95    option = input()96    if option in options.keys():97        options[option]()98    else:
...

vm_import.py

Source:vm_import.py

...8    def describe_conversion_tasks(self):9        raise NotImplementedError(10            "VMImport.describe_conversion_tasks is not yet implemented"11        )12    def import_instance(self):13        raise NotImplementedError("VMImport.import_instance is not yet implemented")14    def import_volume(self):...

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.