Coverage for tests / e2e / test_e2e_from_cli.py: 78%

1"""End-to-end tests for the merge command.""" 

2 

3import shutil 

4import subprocess 

5from pathlib import Path 

6 

7import pytest 

8from rdflib import Dataset, IdentifiedNode 

9from rdflib.compare import graph_diff, isomorphic 

10 

11UV_PATH = shutil.which("uv") 

12if UV_PATH is None: 

13 msg = "uv command not found in PATH" 

14 raise RuntimeError(msg) 

15UV_PATH = Path(UV_PATH) 

16PROJECT_ROOT = Path(__file__).parent.parent.parent 

17 

18 

19@pytest.mark.parametrize( 

20 ("project_name", "command"), 

21 [ 

22 ("eg0-basic", "merge"), 

23 ("eg0-basic", "infer"), 

24 ("eg1-ancestors", "merge"), 

25 ("eg1-ancestors", "infer"), 

26 ], 

27) 

28def test_cli_command( 

29 project_name: str, 

30 command: str, 

31) -> None: 

32 """Test that CLI commands produce expected output for example projects.""" 

33 project_dir = PROJECT_ROOT / "example_projects" / project_name 

34 

35 # Ensure the example directory exists 

36 assert project_dir.exists(), f"Project directory not found: {project_dir}" 

37 

38 actual_file = "merged.trig" if (command == "merge") else "inferred_owlrl.trig" 

39 expected_file = "expected_" + actual_file 

40 

41 # Path to expected and actual output files 

42 expected_file_path = project_dir / "derived" / expected_file 

43 actual_file_path = project_dir / "derived" / actual_file 

44 

45 # Ensure expected file exists 

46 assert expected_file_path.exists(), ( 

47 f"Expected file not found: `{expected_file_path}`" 

48 ) 

49 

50 # Remove actual output if it exists from previous runs 

51 if actual_file_path.exists(): 

52 actual_file_path.unlink() 

53 

54 # Run the command from the project directory 

55 # This is safe as no user-input is passed. 

56 result = subprocess.run( # noqa: S603 

57 [str(UV_PATH), "run", "pythinfer", command], 

58 cwd=project_dir, 

59 capture_output=True, 

60 text=True, 

61 check=False, 

62 ) 

63 

64 # Check the command succeeded 

65 assert result.returncode == 0, ( 

66 f"`{command}` command failed:\n" 

67 f"STDOUT:\n{result.stdout}\n" 

68 f"STDERR:\n{result.stderr}" 

69 ) 

70 

71 # Verify the output file was created 

72 assert actual_file_path.exists(), f"Output file not created: {actual_file_path}" 

73 

74 # Load both graphs and compare them 

75 expected_ds = Dataset() 

76 expected_ds.parse(expected_file_path, format="trig") 

77 

78 actual_ds = Dataset() 

79 actual_ds.parse(actual_file_path, format="trig") 

80 

81 # Compare datasets by checking each named graph individually (handles blank nodes) 

82 # First check we have the same graph identifiers 

83 expected_graphs = {g.identifier for g in expected_ds.graphs()} 

84 actual_graphs = {g.identifier for g in actual_ds.graphs()} 

85 assert expected_graphs == actual_graphs, ( 

86 f"Graph identifiers don't match:\n" 

87 f"Expected: {expected_graphs}\n" 

88 f"Actual: {actual_graphs}" 

89 ) 

90 

91 # Then check each named graph is isomorphic 

92 for graph_id in expected_graphs: 

93 expected_graph = expected_ds.graph(graph_id) 

94 actual_graph = actual_ds.graph(graph_id) 

95 

96 if not isomorphic(expected_graph, actual_graph): 

97 # Compute the difference to show what's missing/extra 

98 in_both, in_expected_only, in_actual_only = graph_diff( 

99 expected_graph, actual_graph 

100 ) 

101 

102 error_msg = [ 

103 f"Graphs with identifier {graph_id} are not isomorphic:", 

104 f"\nTriples in both graphs: {len(in_both)}", 

105 f"\nTriples only in expected ({len(in_expected_only)}):", 

106 ] 

107 if len(in_expected_only) > 0: 

108 error_msg.append(in_expected_only.serialize(format="turtle")) 

109 

110 error_msg.append(f"\nTriples only in actual ({len(in_actual_only)}):") 

111 if len(in_actual_only) > 0: 

112 error_msg.append(in_actual_only.serialize(format="turtle")) 

113 

114 pytest.fail("\n".join(error_msg))