apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-24 12:41:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Fix tests

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-05 17:24:27 +01:00
parent 2192756a17
commit d7145f58fb
4 changed files with 24 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

4
tests/integration/test_load_structured_data_integration.py
View file

@ -382,8 +382,8 @@ Bob Johnson,bob@company.org,not_a_number"""
result = load_structured_data( result = load_structured_data(
api_url="http://invalid-url:9999", api_url="http://invalid-url:9999",
input_file=input_file, input_file=input_file,
descriptor_file=descriptor_file, suggest_schema=True, # Use suggest_schema mode to trigger API connection and propagate errors
flow='obj-ex' flow='obj-ex'
) )
finally: finally:

27
tests/unit/test_cli/test_error_handling_edge_cases.py
View file

@ -63,11 +63,19 @@ class TestErrorHandlingEdgeCases:
# File Access Error Tests # File Access Error Tests
def test_nonexistent_input_file(self): def test_nonexistent_input_file(self):
"""Test handling of nonexistent input file""" """Test handling of nonexistent input file"""
with pytest.raises(FileNotFoundError): # Create a dummy descriptor file for parse_only mode
load_structured_data( descriptor_file = self.create_temp_file('{"format": {"type": "csv"}, "mappings": []}', '.json')
api_url=self.api_url,
input_file="/nonexistent/path/file.csv" try:
) with pytest.raises(FileNotFoundError):
load_structured_data(
api_url=self.api_url,
input_file="/nonexistent/path/file.csv",
descriptor_file=descriptor_file,
parse_only=True # Use parse_only which will propagate FileNotFoundError
)
finally:
self.cleanup_temp_file(descriptor_file)
def test_nonexistent_descriptor_file(self): def test_nonexistent_descriptor_file(self):
"""Test handling of nonexistent descriptor file""" """Test handling of nonexistent descriptor file"""
@ -78,7 +86,8 @@ class TestErrorHandlingEdgeCases:
load_structured_data( load_structured_data(
api_url=self.api_url, api_url=self.api_url,
input_file=input_file, input_file=input_file,
descriptor_file="/nonexistent/descriptor.json" descriptor_file="/nonexistent/descriptor.json",
parse_only=True # Use parse_only since we have a descriptor_file
) )
finally: finally:
self.cleanup_temp_file(input_file) self.cleanup_temp_file(input_file)
@ -118,7 +127,8 @@ class TestErrorHandlingEdgeCases:
load_structured_data( load_structured_data(
api_url=self.api_url, api_url=self.api_url,
input_file=input_file, input_file=input_file,
descriptor_file=descriptor_file descriptor_file=descriptor_file,
parse_only=True # Use parse_only since we have a descriptor_file
) )
finally: finally:
self.cleanup_temp_file(input_file) self.cleanup_temp_file(input_file)
@ -160,7 +170,8 @@ class TestErrorHandlingEdgeCases:
load_structured_data( load_structured_data(
api_url=self.api_url, api_url=self.api_url,
input_file=input_file, input_file=input_file,
descriptor_file=descriptor_file descriptor_file=descriptor_file,
parse_only=True # Use parse_only since we have a descriptor_file
) )
finally: finally:
self.cleanup_temp_file(input_file) self.cleanup_temp_file(input_file)

4
tests/unit/test_cli/test_load_structured_data.py
View file

@ -168,7 +168,9 @@ Bob Johnson,bob@company.org,42,UK"""
with pytest.raises(FileNotFoundError): with pytest.raises(FileNotFoundError):
load_structured_data( load_structured_data(
api_url="http://localhost:8088", api_url="http://localhost:8088",
input_file="/nonexistent/file.csv" input_file="/nonexistent/file.csv",
descriptor_file=self.create_temp_file(json.dumps(self.test_descriptor), '.json'),
parse_only=True # Use parse_only mode which will propagate FileNotFoundError
) )
def test_invalid_descriptor_format(self): def test_invalid_descriptor_format(self):

5
trustgraph-cli/trustgraph/cli/load_structured_data.py
View file

@ -582,11 +582,6 @@ def load_structured_data(
print(f"- Records processed: {len(output_records)}") print(f"- Records processed: {len(output_records)}")
print(f"- Target schema: {schema_name}") print(f"- Target schema: {schema_name}")
print(f"- Field mappings: {len(mappings)}") print(f"- Field mappings: {len(mappings)}")
if args.verbose:
import traceback
traceback.print_exc()
sys.exit(1)
# Helper functions for auto mode # Helper functions for auto mode