From eed215c03b41a68152ee4a60b3e28c7c3c2d21ee Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 5 Sep 2025 16:17:35 +0100 Subject: [PATCH] Tests --- .../test_cli/test_load_structured_data.py | 312 +++++------------- 1 file changed, 86 insertions(+), 226 deletions(-) diff --git a/tests/unit/test_cli/test_load_structured_data.py b/tests/unit/test_cli/test_load_structured_data.py index f84e1e5c..cfe730c2 100644 --- a/tests/unit/test_cli/test_load_structured_data.py +++ b/tests/unit/test_cli/test_load_structured_data.py @@ -14,13 +14,7 @@ from io import StringIO import asyncio # Import the function we're testing -from trustgraph.cli.load_structured_data import ( - load_structured_data, - parse_csv_data, - parse_json_data, - parse_xml_data, - apply_transformations -) +from trustgraph.cli.load_structured_data import load_structured_data class TestLoadStructuredDataUnit: @@ -68,228 +62,94 @@ Bob Johnson,bob@company.org,42,UK""" } } - # CSV Parsing Tests - def test_csv_parsing_with_header(self): - """Test CSV parsing with header row""" - format_info = {"type": "csv", "encoding": "utf-8", "options": {"header": True, "delimiter": ","}} + # CLI Dry-Run Tests - Test CLI behavior without actual connections + def test_csv_dry_run_processing(self): + """Test CSV processing in dry-run mode""" + input_file = self.create_temp_file(self.test_csv_data, '.csv') + descriptor_file = self.create_temp_file(json.dumps(self.test_descriptor), '.json') - records = parse_csv_data(self.test_csv_data, format_info) - - assert len(records) == 3 - assert records[0]["name"] == "John Smith" - assert records[0]["email"] == "john@email.com" - assert records[1]["country"] == "CA" - - def test_csv_parsing_without_header(self): - """Test CSV parsing without header row""" - csv_data = """John Smith,john@email.com,35,US -Jane Doe,jane@email.com,28,CA""" - format_info = {"type": "csv", "encoding": "utf-8", "options": {"header": False, "delimiter": ","}} - - records = parse_csv_data(csv_data, format_info) - - assert len(records) == 2 - # Should use column indices as keys - assert records[0]["0"] == "John Smith" - assert records[0]["1"] == "john@email.com" - - def test_csv_parsing_custom_delimiter(self): - """Test CSV parsing with custom delimiter""" - csv_data = """name;email;age -John Smith;john@email.com;35 -Jane Doe;jane@email.com;28""" - format_info = {"type": "csv", "encoding": "utf-8", "options": {"header": True, "delimiter": ";"}} - - records = parse_csv_data(csv_data, format_info) - - assert len(records) == 2 - assert records[0]["name"] == "John Smith" - assert records[1]["email"] == "jane@email.com" - - # JSON Parsing Tests - def test_json_parsing_array_format(self): - """Test JSON parsing with array format""" - json_data = json.dumps(self.test_json_data) - format_info = {"type": "json", "encoding": "utf-8"} - - records = parse_json_data(json_data, format_info) - - assert len(records) == 2 - assert records[0]["name"] == "John Smith" - assert records[1]["country"] == "CA" - - def test_json_parsing_newline_delimited(self): - """Test JSON parsing with newline-delimited format""" - json_data = '{"name": "John", "age": 35}\n{"name": "Jane", "age": 28}' - format_info = {"type": "json", "encoding": "utf-8", "options": {"newline_delimited": True}} - - records = parse_json_data(json_data, format_info) - - assert len(records) == 2 - assert records[0]["name"] == "John" - assert records[1]["age"] == 28 - - def test_json_parsing_malformed_data(self): - """Test JSON parsing with malformed data""" - json_data = '{"name": "John", "age": 35' # Missing closing brace - format_info = {"type": "json", "encoding": "utf-8"} - - with pytest.raises(json.JSONDecodeError): - parse_json_data(json_data, format_info) - - # XML Parsing Tests - def test_xml_parsing_xpath_expressions(self): - """Test XML parsing with XPath expressions""" - format_info = { - "type": "xml", - "encoding": "utf-8", - "options": { - "record_path": "/ROOT/data/record", - "field_attribute": "name" - } - } - - records = parse_xml_data(self.test_xml_data, format_info) - - assert len(records) == 2 - assert records[0]["name"] == "John Smith" - assert records[0]["email"] == "john@email.com" - assert records[1]["name"] == "Jane Doe" - - def test_xml_parsing_field_attributes(self): - """Test XML parsing with field attributes""" - xml_data = """ - - - -""" - - format_info = { - "type": "xml", - "encoding": "utf-8", - "options": {"record_path": "//item"} - } - - records = parse_xml_data(xml_data, format_info) - - assert len(records) == 2 - assert records[0]["id"] == "1" - assert records[0]["name"] == "Product A" - assert records[1]["price"] == "29.99" - - # Data Transformation Tests - def test_field_mappings_application(self): - """Test field mapping application""" - record = {"source_name": "John Smith", "source_email": "JOHN@EMAIL.COM"} - mappings = [ - {"source_field": "source_name", "target_field": "name", "transforms": []}, - {"source_field": "source_email", "target_field": "email", "transforms": []} - ] - - result = apply_transformations(record, mappings) - - assert result["name"] == "John Smith" - assert result["email"] == "JOHN@EMAIL.COM" - - def test_data_transforms_trim(self): - """Test trim transformation""" - record = {"name": " John Smith ", "email": "john@email.com"} - mappings = [ - {"source_field": "name", "target_field": "name", "transforms": [{"type": "trim"}]} - ] - - result = apply_transformations(record, mappings) - - assert result["name"] == "John Smith" - - def test_data_transforms_case_conversion(self): - """Test case conversion transformations""" - record = {"name": "John Smith", "email": "JOHN@EMAIL.COM"} - mappings = [ - {"source_field": "name", "target_field": "name", "transforms": [{"type": "upper"}]}, - {"source_field": "email", "target_field": "email", "transforms": [{"type": "lower"}]} - ] - - result = apply_transformations(record, mappings) - - assert result["name"] == "JOHN SMITH" - assert result["email"] == "john@email.com" - - def test_data_transforms_type_conversion(self): - """Test type conversion transformations""" - record = {"age": "35", "price": "19.99", "active": "true"} - mappings = [ - {"source_field": "age", "target_field": "age", "transforms": [{"type": "to_int"}]}, - {"source_field": "price", "target_field": "price", "transforms": [{"type": "to_float"}]}, - {"source_field": "active", "target_field": "active", "transforms": [{"type": "to_bool"}]} - ] - - result = apply_transformations(record, mappings) - - # All values should be converted to strings for ExtractedObject compatibility - assert result["age"] == "35" - assert result["price"] == "19.99" - assert result["active"] == "True" - - # Batching Functionality Tests - def test_batch_processing_single_object(self): - """Test batch processing with single object""" - records = [{"name": "John", "age": "35"}] - batch_size = 10 - schema_name = "customer" - - # Simulate batching logic - batches = [] - for i in range(0, len(records), batch_size): - batch_records = records[i:i + batch_size] - batch_values = [record for record in batch_records] - batches.append({ - "schema_name": schema_name, - "values": batch_values - }) - - assert len(batches) == 1 - assert len(batches[0]["values"]) == 1 - assert batches[0]["values"][0]["name"] == "John" - - def test_batch_processing_multiple_objects(self): - """Test batch processing with multiple objects""" - records = [{"name": f"User{i}", "age": str(20+i)} for i in range(5)] - batch_size = 2 - schema_name = "customer" - - # Simulate batching logic - batches = [] - for i in range(0, len(records), batch_size): - batch_records = records[i:i + batch_size] - batch_values = [record for record in batch_records] - batches.append({ - "schema_name": schema_name, - "values": batch_values - }) - - assert len(batches) == 3 # 5 records with batch_size=2 -> 3 batches - assert len(batches[0]["values"]) == 2 - assert len(batches[1]["values"]) == 2 - assert len(batches[2]["values"]) == 1 - assert batches[2]["values"][0]["name"] == "User4" - - def test_batch_size_configuration(self): - """Test batch size configuration""" - records = [{"id": str(i)} for i in range(10)] - - for batch_size in [1, 3, 5, 10, 15]: - batches = [] - for i in range(0, len(records), batch_size): - batch_records = records[i:i + batch_size] - batches.append(batch_records) + try: + # Dry run should complete without errors + result = load_structured_data( + api_url="http://localhost:8088", + input_file=input_file, + descriptor_file=descriptor_file, + dry_run=True, + batch_size=2 + ) - expected_batches = (len(records) + batch_size - 1) // batch_size - assert len(batches) == expected_batches + # Dry run returns None + assert result is None - # Check all records are included - total_records = sum(len(batch) for batch in batches) - assert total_records == len(records) + finally: + self.cleanup_temp_file(input_file) + self.cleanup_temp_file(descriptor_file) + + def test_parse_only_mode(self): + """Test parse-only mode functionality""" + input_file = self.create_temp_file(self.test_csv_data, '.csv') + descriptor_file = self.create_temp_file(json.dumps(self.test_descriptor), '.json') + output_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) + output_file.close() + + try: + result = load_structured_data( + api_url="http://localhost:8088", + input_file=input_file, + descriptor_file=descriptor_file, + parse_only=True, + output_file=output_file.name + ) + + # Check output file was created + assert os.path.exists(output_file.name) + + # Check it contains parsed data + with open(output_file.name, 'r') as f: + parsed_data = json.load(f) + assert isinstance(parsed_data, list) + assert len(parsed_data) > 0 + + finally: + self.cleanup_temp_file(input_file) + self.cleanup_temp_file(descriptor_file) + self.cleanup_temp_file(output_file.name) + + def test_batch_size_parameter(self): + """Test batch size parameter is accepted""" + input_file = self.create_temp_file(self.test_csv_data, '.csv') + descriptor_file = self.create_temp_file(json.dumps(self.test_descriptor), '.json') + + try: + # Should accept batch_size parameter without error + result = load_structured_data( + api_url="http://localhost:8088", + input_file=input_file, + descriptor_file=descriptor_file, + batch_size=5, + dry_run=True + ) + + assert result is None + + finally: + self.cleanup_temp_file(input_file) + self.cleanup_temp_file(descriptor_file) + + def create_temp_file(self, content, suffix='.txt'): + """Create a temporary file with given content""" + temp_file = tempfile.NamedTemporaryFile(mode='w', suffix=suffix, delete=False) + temp_file.write(content) + temp_file.flush() + temp_file.close() + return temp_file.name + + def cleanup_temp_file(self, file_path): + """Clean up temporary file""" + try: + os.unlink(file_path) + except: + pass # Schema Suggestion Tests @patch('trustgraph.cli.load_structured_data.TrustGraphAPI')