def test_validation_parallelize_error(self):
    self.monkeypatch.setattr(dataactvalidator.validation_handlers.validationManager, 'MULTIPROCESSING_POOLS', 2)
    self.validator.submission_id = 1
    self.validator.file_type = self.session.query(FileType).filter_by(file_type_id=FILE_TYPE_DICT['appropriations']).one()
    self.validator.file_name = APPROP_FILE
    self.setup_csv_record_validation(APPROP_FILE, 'appropriations')
    self.validator.is_fabs = False
    self.validator.reader = CsvReader()
    self.validator.error_list = {}
    self.validator.error_rows = []
    self.validator.total_rows = 1
    self.validator.total_data_rows = 0
    self.validator.short_rows = []
    self.validator.long_rows = []
    self.validator.has_data = False
    self.validator.model = Appropriation
    self.validator.error_file_name = report_file_name(self.validator.submission_id, False, self.validator.file_type.name)
    self.validator.error_file_path = ''.join([CONFIG_SERVICES['error_report_path'], self.validator.error_file_name])
    self.validator.warning_file_name = report_file_name(self.validator.submission_id, True, self.validator.file_type.name)
    self.validator.warning_file_path = ''.join([CONFIG_SERVICES['error_report_path'], self.validator.warning_file_name])
    self.validator.fields = self.session.query(FileColumn).filter((FileColumn.file_id == FILE_TYPE_DICT[self.validator.file_type.name])).order_by(FileColumn.daims_name.asc()).all()
    (self.validator.expected_headers, self.validator.parsed_fields) = parse_fields(self.session, self.validator.fields)
    self.validator.csv_schema = {row.name_short: row for row in self.validator.fields}
    with open(self.validator.error_file_path, 'w', newline='') as error_file, open(self.validator.warning_file_path, 'w', newline='') as warning_file:
        error_csv = csv.writer(error_file, delimiter=',', quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
        warning_csv = csv.writer(warning_file, delimiter=',', quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
        error_csv.writerow(self.validator.report_headers)
        warning_csv.writerow(self.validator.report_headers)
    self.validator.filename = self.validator.reader.get_filename(None, None, self.validator.file_name)
    self.validator.reader.open_file(None, None, self.validator.file_name, self.validator.fields, None, self.validator.get_file_name(self.validator.error_file_name), self.validator.daims_to_short_dict[self.validator.file_type.file_type_id], self.validator.short_to_daims_dict[self.validator.file_type.file_type_id], is_local=self.validator.is_local)
    self.validator.reader.file.seek(0)
    reader_obj = pd.read_csv(self.validator.reader.file, dtype=str, delimiter=',', error_bad_lines=False, na_filter=False, chunksize=2, warn_bad_lines=False)
    self.validator.flex_fields = ['flex_field_a', 'flex_field_b']
    self.validator.header_dict = self.validator.reader.header_dict
    self.validator.file_type_name = self.validator.file_type.name
    self.validator.file_type_id = self.validator.file_type.file_type_id
    self.validator.job_id = 2
    normal_chunks = list(reader_obj)
    broken_chunks = [normal_chunks[0], 'BREAK', normal_chunks[1], normal_chunks[2]]
    with self.assertRaises(Exception) as val_except:
        self.validator.parallel_data_loading(self.session, broken_chunks)
    self.assertTrue((type(val_except.exception) == AttributeError))
    self.assertTrue((str(val_except.exception) == "'str' object has no attribute 'empty'"))
    job = ps.Process(os.getpid())
    assert (len(job.children(recursive=True)) == 0)
def test_validation_parallelize_error(self):
    self.monkeypatch.setattr(dataactvalidator.validation_handlers.validationManager, 'MULTIPROCESSING_POOLS', 2)
    self.validator.submission_id = 1
    self.validator.file_type = self.session.query(FileType).filter_by(file_type_id=FILE_TYPE_DICT['appropriations']).one()
    self.validator.file_name = APPROP_FILE
    self.setup_csv_record_validation(APPROP_FILE, 'appropriations')
    self.validator.is_fabs = False
    self.validator.reader = CsvReader()
    self.validator.error_list = {}
    self.validator.error_rows = []
    self.validator.total_rows = 1
    self.validator.total_data_rows = 0
    self.validator.short_rows = []
    self.validator.long_rows = []
    self.validator.has_data = False
    self.validator.model = Appropriation
    self.validator.error_file_name = report_file_name(self.validator.submission_id, False, self.validator.file_type.name)
    self.validator.error_file_path = .join([CONFIG_SERVICES['error_report_path'], self.validator.error_file_name])
    self.validator.warning_file_name = report_file_name(self.validator.submission_id, True, self.validator.file_type.name)
    self.validator.warning_file_path = .join([CONFIG_SERVICES['error_report_path'], self.validator.warning_file_name])
    self.validator.fields = self.session.query(FileColumn).filter((FileColumn.file_id == FILE_TYPE_DICT[self.validator.file_type.name])).order_by(FileColumn.daims_name.asc()).all()
    (self.validator.expected_headers, self.validator.parsed_fields) = parse_fields(self.session, self.validator.fields)
    self.validator.csv_schema = {row.name_short: row for row in self.validator.fields}
    with open(self.validator.error_file_path, 'w', newline=) as error_file, open(self.validator.warning_file_path, 'w', newline=) as warning_file:
        error_csv = csv.writer(error_file, delimiter=',', quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
        warning_csv = csv.writer(warning_file, delimiter=',', quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
        error_csv.writerow(self.validator.report_headers)
        warning_csv.writerow(self.validator.report_headers)
    self.validator.filename = self.validator.reader.get_filename(None, None, self.validator.file_name)
    self.validator.reader.open_file(None, None, self.validator.file_name, self.validator.fields, None, self.validator.get_file_name(self.validator.error_file_name), self.validator.daims_to_short_dict[self.validator.file_type.file_type_id], self.validator.short_to_daims_dict[self.validator.file_type.file_type_id], is_local=self.validator.is_local)
    self.validator.reader.file.seek(0)
    reader_obj = pd.read_csv(self.validator.reader.file, dtype=str, delimiter=',', error_bad_lines=False, na_filter=False, chunksize=2, warn_bad_lines=False)
    self.validator.flex_fields = ['flex_field_a', 'flex_field_b']
    self.validator.header_dict = self.validator.reader.header_dict
    self.validator.file_type_name = self.validator.file_type.name
    self.validator.file_type_id = self.validator.file_type.file_type_id
    self.validator.job_id = 2
    normal_chunks = list(reader_obj)
    broken_chunks = [normal_chunks[0], 'BREAK', normal_chunks[1], normal_chunks[2]]
    with self.assertRaises(Exception) as val_except:
        self.validator.parallel_data_loading(self.session, broken_chunks)
    self.assertTrue((type(val_except.exception) == AttributeError))
    self.assertTrue((str(val_except.exception) == "'str' object has no attribute 'empty'"))
    job = ps.Process(os.getpid())
    assert (len(job.children(recursive=True)) == 0)