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

How to use sanity_checks method in molecule

Best Python code snippet using molecule_python

data_sanity_checks.py

Source:data_sanity_checks.py

1from dataclasses import dataclass2import pandas as pd3from data_validation.data_validation_base_utils import Base4import grequests5import os6@dataclass7class DataSanitization(Base):8    valid_urls = []9    invalid_urls = []10    missing_dates = []11    file_list = os.listdir('../data/')12    invalid_csv_list = list()13    def start_sanitization(self):14        if not self.validate_all_csv_files():15            return self.sanity_checks16        for file in self.file_list:17            print(f'\n >>> PERFORMING SANITY CHECKS ON {file} <<<')18            df = pd.read_csv(f'../data/{file}')19            self.validate_urls(dataframe=df)20            self.validate_all_dates(dataframe=df)21            self.validate_if_any_year_missing(dataframe=df)22        for checks, results in self.sanity_checks.items():23            if isinstance(results, dict):24                self.sanity_passed = False25        return self.sanity_checks26    def validate_all_csv_files(self):27        print('- - Performing Sanity Checks - -')28        for file in self.file_list:29            try:30                df = pd.read_csv(f'../data/{file}')31                # check if any df is empty32                if df.empty:33                    self.invalid_csv_list.append(file)34                    self.sanity_checks['CSV_VALIDATION'] = {'STATUS': 'FAILURE',35                                                            'INVALID_CSV': self.invalid_csv_list}36                    return False37            except Exception as e:38                print(f"Error parsing file: {file} -> {e}")39                self.invalid_csv_list.append(file)40                self.sanity_checks['CSV_VALIDATION'] = {'STATUS': 'FAILURE',41                                                        'INVALID_CSV': self.invalid_csv_list}42                return False43        # return only those files who have valid data in them44        self.sanity_checks['CSV_VALIDATION'] = 'SUCCESS'45        return [x for x in self.file_list if x not in self.invalid_csv_list]46    def validate_urls(self, dataframe):47        current_url = None48        if 'url' in dataframe.columns:49            try:50                rs = (grequests.get(u) for u in dataframe['url'])51                requests = grequests.map(rs)52                for response, url in zip(requests, dataframe['url']):53                    current_url = url54                    # Validate if all URLS returns status code of 20055                    if response.status_code != 200:56                        self.invalid_urls.append(url)57                    else:58                        self.valid_urls.append(url)59            except Exception as e:60                self.invalid_urls.append(current_url)61        if len(self.invalid_urls) > 0:62            self.sanity_checks['URL_VALIDATION'] = {'STATUS': 'FAILED', 'INVALID_URL': self.invalid_urls}63        else:64            self.sanity_checks['URL_VALIDATION'] = 'SUCCESS'65            return True66    def validate_all_dates(self, dataframe):67        if 'date' in dataframe.columns:68            # Validate if dates match the specified format69            if pd.to_datetime(dataframe['date'], format='%Y-%m-%d', errors='coerce').notnull().all():70                self.sanity_checks['DATE_FORMAT_VALIDATION'] = 'SUCCESS'71            else:72                self.sanity_checks['DATE_FORMAT_VALIDATION'] = 'FAILED'73    def validate_if_any_year_missing(self, dataframe):74        if 'year' in dataframe.columns:75            # Get all unique years from the dataframes76            years = dataframe['year'].unique()77            # find all the years missing in the series78            missing_years = [y for y in range(min(years), max(years) + 1) if y not in years]79            # keep appending to avoid any issues in new files80            for year in missing_years:81                if year not in self.missing_dates:82                    self.missing_dates.append(year)83            if len(self.missing_dates) > 0:84                self.sanity_checks['RACE YEAR VALIDATION'] = {'Status': 'FAILED', 'MISSING YEARS': self.missing_dates}85            else:...

test_sanity_check.py

Source:test_sanity_check.py

1import os2import tempfile3import portend4from django.db.utils import OperationalError5from django.test import TestCase6from mock import patch7from kolibri.utils import sanity_checks8from kolibri.utils.sanity_checks import DatabaseNotMigrated9from kolibri.utils.server import NotRunning10from kolibri.utils.tests.helpers import override_option11class SanityCheckTestCase(TestCase):12    @patch("kolibri.utils.sanity_checks.logging.error")13    @patch("kolibri.utils.sanity_checks.get_status", return_value={1, 2, 3})14    def test_other_kolibri_running(self, status_mock, logging_mock):15        with self.assertRaises(SystemExit):16            sanity_checks.check_other_kolibri_running("8080")17            logging_mock.assert_called()18    @patch("kolibri.utils.sanity_checks.logging.error")19    @patch("kolibri.utils.sanity_checks.portend.free")20    @patch("kolibri.utils.sanity_checks.get_status")21    def test_port_occupied(self, status_mock, portend_mock, logging_mock):22        status_mock.side_effect = NotRunning("Kolibri not running")23        portend_mock.side_effect = portend.Timeout24        with self.assertRaises(SystemExit):25            sanity_checks.check_port_availability("0.0.0.0", "8080")26            logging_mock.assert_called()27    @patch("kolibri.utils.sanity_checks.logging.error")28    @patch("kolibri.utils.sanity_checks.portend.free")29    def test_socket_activation_support(self, portend_mock, logging_mock):30        portend_mock.side_effect = portend.Timeout31        # LISTEN_PID environment variable would be set if using socket activation32        with patch.dict(os.environ, {"LISTEN_PID": "1234"}):33            sanity_checks.check_port_availability("0.0.0.0", "8080")34            logging_mock.assert_not_called()35    @patch("kolibri.utils.cli.get_version", return_value="")36    @patch("kolibri.utils.sanity_checks.logging.error")37    @override_option("Paths", "CONTENT_DIR", "/dir_dne")38    def test_content_dir_dne(self, logging_mock, get_version):39        with self.assertRaises(SystemExit):40            sanity_checks.check_content_directory_exists_and_writable()41            logging_mock.assert_called()42    @patch("kolibri.utils.sanity_checks.logging.error")43    @patch("kolibri.utils.sanity_checks.os.access", return_value=False)44    @override_option("Paths", "CONTENT_DIR", tempfile.mkdtemp())45    def test_content_dir_not_writable(self, access_mock, logging_mock):46        with self.assertRaises(SystemExit):47            sanity_checks.check_content_directory_exists_and_writable()48            logging_mock.assert_called()49    @patch("kolibri.utils.cli.get_version", return_value="")50    @patch("kolibri.utils.sanity_checks.shutil.move")51    @patch(52        "kolibri.utils.sanity_checks.os.path.exists",53        # This requires an additional return value at the end54        # to prevent a StopIteration exception during test55        # execution, but the first three values are the ones56        # that make the difference to the assert count below.57        side_effect=[True, False, True, False],58    )59    def test_old_log_file_exists(self, path_exists_mock, move_mock, get_version):60        sanity_checks.check_log_file_location()61        # Check if the number of calls to shutil.move equals to the number of times62        # os.path.exists returns True63        self.assertEqual(move_mock.call_count, 2)64    def test_check_database_is_migrated(self):65        from morango.models import InstanceIDModel66        with patch.object(67            InstanceIDModel, "get_or_create_current_instance"68        ) as get_or_create_current_instance:69            get_or_create_current_instance.side_effect = OperationalError("Test")70            with self.assertRaises(DatabaseNotMigrated):...

test_sanity_checks.py

Source:test_sanity_checks.py

...15        cfg = compose(16            config_name="config_test",17        )18    return cfg19def test_sanity_checks(cfg):20    """Integration test to ensure main launcher works"""21    sanity_checks.main(cfg)22@pytest.mark.parametrize("similarity_type", ["resnet18", "resnet18_no_aug"])23def test_run_within_v_across(cfg, similarity_type):24    image_net_dir = sanity_checks.get_image_net_dir()25    class_labels = sanity_checks.get_class_labels(cfg, image_net_dir)26    across_v_within = sanity_checks.run_within_v_across(27        image_net_dir,28        class_labels,29        similarity_type,30        max_n_image_pairs=cfg.sanity_checks.max_n_image_pairs,31        max_class_label_combinations=cfg.sanity_checks.max_class_label_combinations,32        randomly_rotate=False,33    )...

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.