Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_standalone_row_limit_in_range(self):
filepath = os.path.join(self.data_dir, 'row_limit_schema.csv')
schema = os.path.join(self.data_dir, 'test_schema.json')
with io.open(filepath) as stream:
validator = processors.SchemaProcessor(row_limit=2, schema=schema)
result, report, data = validator.run(stream)
self.assertEqual(len(report.generate()['results']), 0)
def test_standalone_schema_valid_simple(self):
data_filepath = os.path.join(self.data_dir, 'contacts', 'people.csv')
schema_filepath = os.path.join(self.data_dir, 'contacts',
'schema_valid.json')
with io.open(data_filepath) as data_stream, \
io.open(schema_filepath) as schema_stream:
schema = json.load(schema_stream)
validator = processors.SchemaProcessor(schema=schema)
result, report, data = validator.run(data_stream)
self.assertTrue(result)
def test_standalone_process_extra_fields(self):
filepath = os.path.join(self.data_dir, 'contacts', 'people.csv')
schema = os.path.join(self.data_dir, 'contacts', 'schema_incomplete_fields.json')
validator = processors.SchemaProcessor(schema=schema, case_insensitive_headers=True,
process_extra_fields=True, result_level='info')
result, report, data = validator.run(filepath)
reports = report.generate()
self.assertTrue(any('schema_008' in res.values() for res in reports['results']))
def test_standalone_report_limit_out_range(self):
limit = processors.SchemaProcessor.REPORT_LIMIT_MAX
validator = processors.SchemaProcessor(report_limit=(limit + 1))
self.assertEqual(validator.report_limit, limit)
def test_standalone_fail_fast_true(self):
filepath = os.path.join(self.data_dir, 'fail_fast_two_schema_errors.csv')
schema = os.path.join(self.data_dir, 'test_schema.json')
with io.open(filepath) as stream:
validator = processors.SchemaProcessor(fail_fast=True, schema=schema)
result, report, data = validator.run(stream)
self.assertEqual(len(report.generate()['results']), 1)
def test_standalone_case_insensitive_headers(self):
filepath = os.path.join(self.data_dir, 'case_insensitive_headers.csv')
schema = os.path.join(self.data_dir, 'test_schema.json')
validator = processors.SchemaProcessor(schema=schema, case_insensitive_headers=True)
result, report, data = validator.run(filepath)
self.assertEqual(len(report.generate()['results']), 0)