diff --git a/tests/functional/test_advanced_features.py b/tests/functional/test_advanced_features.py index 5ecb5676..e44f9fa2 100644 --- a/tests/functional/test_advanced_features.py +++ b/tests/functional/test_advanced_features.py @@ -3,6 +3,7 @@ from pyreason import Threshold import torch import torch.nn as nn +import networkx as nx import numba import numpy as np import pytest @@ -31,8 +32,10 @@ def probability_func(annotations, weights): return union_prob, 1 -def test_probability_func_consistency(): +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_probability_func_consistency(mode): """Ensure annotation function behaves the same with and without JIT.""" + setup_mode(mode) annotations = numba.typed.List() annotations.append(numba.typed.List([closed(0.01, 1.0)])) annotations.append(numba.typed.List([closed(0.2, 1.0)])) @@ -172,3 +175,71 @@ def test_classifier_integration(mode): print("\nGenerated PyReason Facts:") for fact in facts: print(fact) + + +@pytest.mark.skipif(True, reason="Reason again functionality not implemented for FP version") +@pytest.mark.parametrize("mode", ["regular"]) +def test_reason_again(mode): + """Test reasoning continuation functionality.""" + setup_mode(mode) + + # Modify the paths based on where you've stored the files we made above + graph_path = './tests/functional/friends_graph.graphml' + + # Load all the files into pyreason + pr.load_graphml(graph_path) + pr.add_rule(pr.Rule('popular(x) <-1 popular(y), Friends(x,y), owns(y,z), owns(x,z)', 'popular_rule')) + pr.add_fact(pr.Fact('popular(Mary)', 'popular_fact', 0, 1)) + + # Run the program for two timesteps to see the diffusion take place + faulthandler.enable() + interpretation = pr.reason(timesteps=1) + + # Now reason again + new_fact = pr.Fact('popular(Mary)', 'popular_fact2', 2, 4) + pr.add_fact(new_fact) + interpretation = pr.reason(timesteps=3, again=True, restart=False) + + # Display the changes in the interpretation for each timestep + dataframes = pr.filter_and_sort_nodes(interpretation, ['popular']) + for t, df in enumerate(dataframes): + print(f'TIMESTEP - {t}') + print(df) + print() + + assert len(dataframes[2]) == 1, 'At t=0 there should be one popular person' + assert len(dataframes[3]) == 2, 'At t=1 there should be two popular people' + assert len(dataframes[4]) == 3, 'At t=2 there should be three popular people' + + # Mary should be popular in all three timesteps + assert 'Mary' in dataframes[2]['component'].values and dataframes[2].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=0 timesteps' + assert 'Mary' in dataframes[3]['component'].values and dataframes[3].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=1 timesteps' + assert 'Mary' in dataframes[4]['component'].values and dataframes[4].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=2 timesteps' + + # Justin should be popular in timesteps 1, 2 + assert 'Justin' in dataframes[3]['component'].values and dataframes[3].iloc[1].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=1 timesteps' + assert 'Justin' in dataframes[4]['component'].values and dataframes[4].iloc[2].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=2 timesteps' + + # John should be popular in timestep 3 + assert 'John' in dataframes[4]['component'].values and dataframes[4].iloc[1].popular == [1, 1], 'John should have popular bounds [1,1] for t=2 timesteps' + + +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_reason_with_queries(mode): + """Test reasoning with query-based rule filtering""" + setup_mode(mode) + # Set up test scenario + graph = nx.DiGraph() + graph.add_edges_from([("A", "B"), ("B", "C")]) + pr.load_graph(graph) + + pr.add_rule(pr.Rule('popular(x) <-1 friend(x, y)', 'rule1')) + pr.add_rule(pr.Rule('friend(x, y) <-1 knows(x, y)', 'rule2')) + pr.add_fact(pr.Fact('knows(A, B)', 'fact1')) + + # Create query to filter rules + query = pr.Query('popular(A)') + pr.settings.verbose = False # Reduce output noise + + interpretation = pr.reason(timesteps=1, queries=[query]) + # Should complete and apply rule filtering logic diff --git a/tests/functional/test_annotation_function.py b/tests/functional/test_annotation_function.py deleted file mode 100644 index e69de29b..00000000 diff --git a/tests/functional/test_anyBurl_infer_edges_rules.py b/tests/functional/test_anyBurl_infer_edges_rules.py index c58c6ce3..fe9e9501 100644 --- a/tests/functional/test_anyBurl_infer_edges_rules.py +++ b/tests/functional/test_anyBurl_infer_edges_rules.py @@ -1,48 +1,23 @@ import pyreason as pr import pytest - -@pytest.mark.slow -def test_anyBurl_rule_1(): - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' +def setup_mode(mode): + """Configure PyReason settings for the specified mode.""" pr.reset() pr.reset_rules() pr.reset_settings() - # Modify pyreason settings to make verbose and to save the rule trace to a file pr.settings.verbose = True - pr.settings.atom_trace = True - pr.settings.memory_profile = False - pr.settings.canonical = True - pr.settings.inconsistency_check = False - pr.settings.static_graph_facts = False - pr.settings.output_to_file = False - pr.settings.store_interpretation_changes = True - pr.settings.save_graph_attributes_to_trace = True - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('isConnectedTo(A, Y) <-1 isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_1', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - # pr.save_rule_trace(interpretation) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 2 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Vnukovo_International_Airport', 'Riga_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Riga_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' + if mode == "fp": + pr.settings.fp_version = True + elif mode == "parallel": + pr.settings.parallel_computing = True @pytest.mark.slow -def test_anyBurl_rule_2(): +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_anyBurl_rule_1(mode): graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() + setup_mode(mode) # Modify pyreason settings to make verbose and to save the rule trace to a file pr.settings.verbose = True pr.settings.atom_trace = True @@ -53,119 +28,6 @@ def test_anyBurl_rule_2(): pr.settings.output_to_file = False pr.settings.store_interpretation_changes = True pr.settings.save_graph_attributes_to_trace = True - pr.settings.parallel_computing = False - # Load all the files into pyreason - pr.load_graphml(graph_path) - - pr.add_rule(pr.Rule('isConnectedTo(Y, A) <-1 isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_2', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - # pr.save_rule_trace(interpretation) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 2 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Riga_International_Airport', 'Vnukovo_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Riga_International_Airport, Vnukovo_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.slow -def test_anyBurl_rule_3(): - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() - # Modify pyreason settings to make verbose and to save the rule trace to a file - pr.settings.verbose = True - pr.settings.atom_trace = True - pr.settings.memory_profile = False - pr.settings.canonical = True - pr.settings.inconsistency_check = False - pr.settings.static_graph_facts = False - pr.settings.output_to_file = False - pr.settings.store_interpretation_changes = True - pr.settings.save_graph_attributes_to_trace = True - pr.settings.parallel_computing = False - # Load all the files into pyreason - pr.load_graphml(graph_path) - - pr.add_rule(pr.Rule('isConnectedTo(A, Y) <-1 isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_3', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - # pr.save_rule_trace(interpretation) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 1 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Vnukovo_International_Airport', 'Yali') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Yali) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.slow -def test_anyBurl_rule_4(): - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() - # Modify pyreason settings to make verbose and to save the rule trace to a file - pr.settings.verbose = True - pr.settings.atom_trace = True - pr.settings.memory_profile = False - pr.settings.canonical = True - pr.settings.inconsistency_check = False - pr.settings.static_graph_facts = False - pr.settings.output_to_file = False - pr.settings.store_interpretation_changes = True - pr.settings.save_graph_attributes_to_trace = True - pr.settings.parallel_computing = False - # Load all the files into pyreason - pr.load_graphml(graph_path) - - pr.add_rule(pr.Rule('isConnectedTo(Y, A) <-1 isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_4', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - # pr.save_rule_trace(interpretation) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 1 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Yali', 'Vnukovo_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Yali, Vnukovo_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.fp -@pytest.mark.slow -def test_anyBurl_rule_1_fp(): - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() - # Modify pyreason settings to make verbose and to save the rule trace to a file - pr.settings.verbose = True - pr.settings.fp_version = True # Use the FP version of the reasoner - pr.settings.atom_trace = True - pr.settings.memory_profile = False - pr.settings.canonical = True - pr.settings.inconsistency_check = False - pr.settings.static_graph_facts = False - pr.settings.output_to_file = False - pr.settings.store_interpretation_changes = True - pr.settings.save_graph_attributes_to_trace = True # Load all the files into pyreason pr.load_graphml(graph_path) pr.add_rule(pr.Rule('isConnectedTo(A, Y) <-1 isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_1', infer_edges=True)) @@ -185,16 +47,13 @@ def test_anyBurl_rule_1_fp(): assert ('Vnukovo_International_Airport', 'Riga_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Riga_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' -@pytest.mark.fp @pytest.mark.slow -def test_anyBurl_rule_2_fp(): +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_anyBurl_rule_2(mode): graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() + setup_mode(mode) # Modify pyreason settings to make verbose and to save the rule trace to a file pr.settings.verbose = True - pr.settings.fp_version = True # Use the FP version of the reasoner pr.settings.atom_trace = True pr.settings.memory_profile = False pr.settings.canonical = True @@ -224,16 +83,13 @@ def test_anyBurl_rule_2_fp(): assert ('Riga_International_Airport', 'Vnukovo_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Riga_International_Airport, Vnukovo_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' -@pytest.mark.fp @pytest.mark.slow -def test_anyBurl_rule_3_fp(): +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_anyBurl_rule_3(mode): graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() + setup_mode(mode) # Modify pyreason settings to make verbose and to save the rule trace to a file pr.settings.verbose = True - pr.settings.fp_version = True # Use the FP version of the reasoner pr.settings.atom_trace = True pr.settings.memory_profile = False pr.settings.canonical = True @@ -263,16 +119,13 @@ def test_anyBurl_rule_3_fp(): assert ('Vnukovo_International_Airport', 'Yali') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Yali) should have isConnectedTo bounds [1,1] for t=1 timesteps' -@pytest.mark.fp @pytest.mark.slow -def test_anyBurl_rule_4_fp(): +@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) +def test_anyBurl_rule_4(mode): graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - pr.reset() - pr.reset_rules() - pr.reset_settings() + setup_mode(mode) # Modify pyreason settings to make verbose and to save the rule trace to a file pr.settings.verbose = True - pr.settings.fp_version = True # Use the FP version of the reasoner pr.settings.atom_trace = True pr.settings.memory_profile = False pr.settings.canonical = True diff --git a/tests/functional/test_basic_reasoning.py b/tests/functional/test_basic_reasoning.py index 3c165cc2..b0b39d30 100644 --- a/tests/functional/test_basic_reasoning.py +++ b/tests/functional/test_basic_reasoning.py @@ -59,55 +59,6 @@ def test_hello_world(mode): # John should be popular in timestep 3 assert 'John' in dataframes[2]['component'].values and dataframes[2].iloc[1].popular == [1, 1], 'John should have popular bounds [1,1] for t=2 timesteps' - -@pytest.mark.slow -def test_hello_world_consistency(): - """Test consistency between JIT and pure Python implementations.""" - # Reset PyReason - pr.reset() - pr.reset_rules() - pr.reset_settings() - - # Modify the paths based on where you've stored the files we made above - graph_path = './tests/functional/friends_graph.graphml' - - # Modify pyreason settings to make verbose - pr.settings.verbose = True # Print info to screen - pr.settings.atom_trace = True # Print atom trace - pr.settings.test_inconsistency = True - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('popular(x) <-1 popular(y), Friends(x,y), owns(y,z), owns(x,z)', 'popular_rule')) - pr.add_fact(pr.Fact('popular(Mary)', 'popular_fact', 0, 2)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=2) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_nodes(interpretation, ['popular']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - - assert len(dataframes[0]) == 1, 'At t=0 there should be one popular person' - assert len(dataframes[1]) == 2, 'At t=1 there should be two popular people' - assert len(dataframes[2]) == 3, 'At t=2 there should be three popular people' - - # Mary should be popular in all three timesteps - assert 'Mary' in dataframes[0]['component'].values and dataframes[0].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=0 timesteps' - assert 'Mary' in dataframes[1]['component'].values and dataframes[1].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=1 timesteps' - assert 'Mary' in dataframes[2]['component'].values and dataframes[2].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=2 timesteps' - - # Justin should be popular in timesteps 1, 2 - assert 'Justin' in dataframes[1]['component'].values and dataframes[1].iloc[1].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=1 timesteps' - assert 'Justin' in dataframes[2]['component'].values and dataframes[2].iloc[2].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=2 timesteps' - - # John should be popular in timestep 3 - assert 'John' in dataframes[2]['component'].values and dataframes[2].iloc[1].popular == [1, 1], 'John should have popular bounds [1,1] for t=2 timesteps' - - @pytest.mark.parametrize("mode", ["regular", "fp"]) def test_reorder_clauses(mode): """Test clause reordering functionality.""" @@ -161,4 +112,4 @@ def test_reorder_clauses(mode): assert first_justin_rule['Clause-1'][0] == ('Justin', 'Mary') else: # Regular version: The second row, clause 1 should be the edge grounding ('Justin', 'Mary') - assert rule_trace_node.iloc[2]['Clause-1'][0] == ('Justin', 'Mary') \ No newline at end of file + assert rule_trace_node.iloc[2]['Clause-1'][0] == ('Justin', 'Mary') diff --git a/tests/functional/test_continuation.py b/tests/functional/test_continuation.py deleted file mode 100644 index 1b572ca1..00000000 --- a/tests/functional/test_continuation.py +++ /dev/null @@ -1,65 +0,0 @@ -# Tests for reasoning continuation functionality (reason again) -import pyreason as pr -import faulthandler -import pytest - - -def setup_mode(mode): - """Configure PyReason settings for the specified mode.""" - pr.reset() - pr.reset_rules() - pr.reset_settings() - pr.settings.verbose = True - pr.settings.atom_trace = True - - if mode == "fp": - pr.settings.fp_version = True - elif mode == "parallel": - pr.settings.parallel_computing = True - - -@pytest.mark.skipif(True, reason="Reason again functionality not implemented for FP version") -@pytest.mark.parametrize("mode", ["regular"]) -def test_reason_again(mode): - """Test reasoning continuation functionality.""" - setup_mode(mode) - - # Modify the paths based on where you've stored the files we made above - graph_path = './tests/functional/friends_graph.graphml' - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('popular(x) <-1 popular(y), Friends(x,y), owns(y,z), owns(x,z)', 'popular_rule')) - pr.add_fact(pr.Fact('popular(Mary)', 'popular_fact', 0, 1)) - - # Run the program for two timesteps to see the diffusion take place - faulthandler.enable() - interpretation = pr.reason(timesteps=1) - - # Now reason again - new_fact = pr.Fact('popular(Mary)', 'popular_fact2', 2, 4) - pr.add_fact(new_fact) - interpretation = pr.reason(timesteps=3, again=True, restart=False) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_nodes(interpretation, ['popular']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - - assert len(dataframes[2]) == 1, 'At t=0 there should be one popular person' - assert len(dataframes[3]) == 2, 'At t=1 there should be two popular people' - assert len(dataframes[4]) == 3, 'At t=2 there should be three popular people' - - # Mary should be popular in all three timesteps - assert 'Mary' in dataframes[2]['component'].values and dataframes[2].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=0 timesteps' - assert 'Mary' in dataframes[3]['component'].values and dataframes[3].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=1 timesteps' - assert 'Mary' in dataframes[4]['component'].values and dataframes[4].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=2 timesteps' - - # Justin should be popular in timesteps 1, 2 - assert 'Justin' in dataframes[3]['component'].values and dataframes[3].iloc[1].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=1 timesteps' - assert 'Justin' in dataframes[4]['component'].values and dataframes[4].iloc[2].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=2 timesteps' - - # John should be popular in timestep 3 - assert 'John' in dataframes[4]['component'].values and dataframes[4].iloc[1].popular == [1, 1], 'John should have popular bounds [1,1] for t=2 timesteps' diff --git a/tests/functional/test_edge_inference.py b/tests/functional/test_edge_inference.py deleted file mode 100644 index ad9eb969..00000000 --- a/tests/functional/test_edge_inference.py +++ /dev/null @@ -1,127 +0,0 @@ -# Edge inference rule tests for PyReason -import pyreason as pr -import pytest - - -def setup_mode(mode): - """Configure PyReason settings for the specified mode.""" - pr.reset() - pr.reset_rules() - pr.reset_settings() - - # Modify pyreason settings to make verbose and to save the rule trace to a file - pr.settings.verbose = True - pr.settings.atom_trace = True - pr.settings.memory_profile = False - pr.settings.canonical = True - pr.settings.inconsistency_check = False - pr.settings.static_graph_facts = False - pr.settings.output_to_file = False - pr.settings.store_interpretation_changes = True - pr.settings.save_graph_attributes_to_trace = True - pr.settings.parallel_computing = False - - if mode == "fp": - pr.settings.fp_version = True - elif mode == "parallel": - pr.settings.parallel_computing = True - - -@pytest.mark.slow -@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) -def test_anyBurl_rule_1(mode): - """Test anyBurl rule 1: isConnectedTo(A, Y) <- isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)""" - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - setup_mode(mode) - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('isConnectedTo(A, Y) <-1 isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_1', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 2 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Vnukovo_International_Airport', 'Riga_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Riga_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.slow -@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) -def test_anyBurl_rule_2(mode): - """Test anyBurl rule 2: isConnectedTo(Y, A) <- isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)""" - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - setup_mode(mode) - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('isConnectedTo(Y, A) <-1 isConnectedTo(Y, B), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_2', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 2 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Riga_International_Airport', 'Vnukovo_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Riga_International_Airport, Vnukovo_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.slow -@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) -def test_anyBurl_rule_3(mode): - """Test anyBurl rule 3: isConnectedTo(A, Y) <- isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)""" - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - setup_mode(mode) - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('isConnectedTo(A, Y) <-1 isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_3', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 1 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Vnukovo_International_Airport', 'Yali') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Vnukovo_International_Airport, Yali) should have isConnectedTo bounds [1,1] for t=1 timesteps' - - -@pytest.mark.slow -@pytest.mark.parametrize("mode", ["regular", "fp", "parallel"]) -def test_anyBurl_rule_4(mode): - """Test anyBurl rule 4: isConnectedTo(Y, A) <- isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)""" - graph_path = './tests/functional/knowledge_graph_test_subset.graphml' - setup_mode(mode) - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('isConnectedTo(Y, A) <-1 isConnectedTo(B, Y), Amsterdam_Airport_Schiphol(B), Vnukovo_International_Airport(A)', 'connected_rule_4', infer_edges=True)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=1) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_edges(interpretation, ['isConnectedTo']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - assert len(dataframes) == 2, 'Pyreason should run exactly 1 fixpoint operations' - assert len(dataframes[1]) == 1, 'At t=1 there should be only 1 new isConnectedTo atom' - assert ('Yali', 'Vnukovo_International_Airport') in dataframes[1]['component'].values.tolist() and dataframes[1]['isConnectedTo'].iloc[0] == [1, 1], '(Yali, Vnukovo_International_Airport) should have isConnectedTo bounds [1,1] for t=1 timesteps' diff --git a/tests/functional/test_hello_world_parallel.py b/tests/functional/test_hello_world_parallel.py deleted file mode 100644 index f9c04325..00000000 --- a/tests/functional/test_hello_world_parallel.py +++ /dev/null @@ -1,47 +0,0 @@ -# Test if the simple hello world program works. -import pyreason as pr - - -def test_hello_world_parallel(): - # Reset PyReason - pr.reset() - pr.reset_rules() - - # Modify the paths based on where you've stored the files we made above - graph_path = './tests/functional/friends_graph.graphml' - - # Modify pyreason settings to make verbose - pr.reset_settings() - pr.settings.verbose = True # Print info to screen - pr.settings.parallel_computing = True # Use parallel computing - - # Load all the files into pyreason - pr.load_graphml(graph_path) - pr.add_rule(pr.Rule('popular(x) <-1 popular(y), Friends(x,y), owns(y,z), owns(x,z)', 'popular_rule')) - pr.add_fact(pr.Fact('popular(Mary)', 'popular_fact', 0, 2)) - - # Run the program for two timesteps to see the diffusion take place - interpretation = pr.reason(timesteps=2) - - # Display the changes in the interpretation for each timestep - dataframes = pr.filter_and_sort_nodes(interpretation, ['popular']) - for t, df in enumerate(dataframes): - print(f'TIMESTEP - {t}') - print(df) - print() - - assert len(dataframes[0]) == 1, 'At t=0 there should be one popular person' - assert len(dataframes[1]) == 2, 'At t=1 there should be two popular people' - assert len(dataframes[2]) == 3, 'At t=2 there should be three popular people' - - # Mary should be popular in all three timesteps - assert 'Mary' in dataframes[0]['component'].values and dataframes[0].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=0 timesteps' - assert 'Mary' in dataframes[1]['component'].values and dataframes[1].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=1 timesteps' - assert 'Mary' in dataframes[2]['component'].values and dataframes[2].iloc[0].popular == [1, 1], 'Mary should have popular bounds [1,1] for t=2 timesteps' - - # Justin should be popular in timesteps 1, 2 - assert 'Justin' in dataframes[1]['component'].values and dataframes[1].iloc[1].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=1 timesteps' - assert 'Justin' in dataframes[2]['component'].values and dataframes[2].iloc[2].popular == [1, 1], 'Justin should have popular bounds [1,1] for t=2 timesteps' - - # John should be popular in timestep 3 - assert 'John' in dataframes[2]['component'].values and dataframes[2].iloc[1].popular == [1, 1], 'John should have popular bounds [1,1] for t=2 timesteps' diff --git a/tests/functional/test_pyreason_comprehensive.py b/tests/functional/test_pyreason_comprehensive.py deleted file mode 100644 index ea4b06ac..00000000 --- a/tests/functional/test_pyreason_comprehensive.py +++ /dev/null @@ -1,436 +0,0 @@ -""" -Comprehensive functional tests for pyreason.py to cover missing branches. -These tests focus on error conditions, settings validation, and edge cases. -""" - -import pytest -import tempfile -import os -import networkx as nx -from unittest import mock - -import pyreason as pr - - -class TestSettingsValidation: - """Test settings validation - covers all TypeError branches""" - - def setup_method(self): - """Reset settings before each test""" - pr.reset() - pr.reset_settings() - - def test_verbose_type_error(self): - """Test verbose setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.verbose = "not_bool" - - def test_output_to_file_type_error(self): - """Test output_to_file setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.output_to_file = 123 - - def test_output_file_name_type_error(self): - """Test output_file_name setter with invalid type""" - with pytest.raises(TypeError, match='file_name has to be a string'): - pr.settings.output_file_name = 123 - - def test_graph_attribute_parsing_type_error(self): - """Test graph_attribute_parsing setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.graph_attribute_parsing = "not_bool" - - def test_abort_on_inconsistency_type_error(self): - """Test abort_on_inconsistency setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.abort_on_inconsistency = 1.5 - - def test_memory_profile_type_error(self): - """Test memory_profile setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.memory_profile = [] - - def test_reverse_digraph_type_error(self): - """Test reverse_digraph setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.reverse_digraph = {} - - def test_atom_trace_type_error(self): - """Test atom_trace setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.atom_trace = "false" - - def test_save_graph_attributes_to_trace_type_error(self): - """Test save_graph_attributes_to_trace setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.save_graph_attributes_to_trace = 0 - - def test_canonical_type_error(self): - """Test canonical setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.canonical = "canonical" - - def test_persistent_type_error(self): - """Test persistent setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.persistent = None - - def test_inconsistency_check_type_error(self): - """Test inconsistency_check setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.inconsistency_check = 42 - - def test_static_graph_facts_type_error(self): - """Test static_graph_facts setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.static_graph_facts = [True] - - def test_store_interpretation_changes_type_error(self): - """Test store_interpretation_changes setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.store_interpretation_changes = 1 - - def test_parallel_computing_type_error(self): - """Test parallel_computing setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.parallel_computing = "True" - - def test_update_mode_type_error(self): - """Test update_mode setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a str'): - pr.settings.update_mode = True - - def test_allow_ground_rules_type_error(self): - """Test allow_ground_rules setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.allow_ground_rules = 3.14 - - def test_fp_version_type_error(self): - """Test fp_version setter with invalid type""" - with pytest.raises(TypeError, match='value has to be a bool'): - pr.settings.fp_version = "optimized" - - -class TestFileOperations: - """Test file loading operations and error conditions""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_load_graphml_nonexistent_file(self): - """Test loading non-existent GraphML file""" - with pytest.raises((FileNotFoundError, OSError)): - pr.load_graphml("non_existent_file.graphml") - - def test_load_ipl_nonexistent_file(self): - """Test loading non-existent IPL file""" - with pytest.raises((FileNotFoundError, OSError)): - pr.load_inconsistent_predicate_list("non_existent_ipl.yaml") - - def test_add_rules_from_nonexistent_file(self): - """Test adding rules from non-existent file""" - with pytest.raises((FileNotFoundError, OSError)): - pr.add_rules_from_file("non_existent_rules.txt") - - def test_add_rules_from_file_with_comments_and_empty_lines(self): - """Test rule file parsing handles comments and empty lines""" - with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f: - f.write("# This is a comment\n") - f.write("\n") # Empty line - f.write(" \n") # Whitespace-only line - f.write("test_rule(x) <-1 other_rule(x)\n") - f.write("# Another comment\n") - f.write("another_rule(y) <-1 test_rule(y)\n") - temp_path = f.name - - try: - pr.add_rules_from_file(temp_path) - rules = pr.get_rules() - assert len(rules) == 2 # Should only include the 2 actual rules - finally: - os.unlink(temp_path) - - def test_add_inconsistent_predicates(self): - """Test adding inconsistent predicate pairs""" - pr.add_inconsistent_predicate("pred1", "pred2") - pr.add_inconsistent_predicate("pred3", "pred4") - # Should not raise exceptions - - -class TestReasoningErrorConditions: - """Test reasoning function error conditions and edge cases""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_reason_without_rules_exception(self): - """Test reasoning without rules raises exception""" - # Load a graph but no rules - graph = nx.DiGraph() - graph.add_edge("A", "B") - pr.load_graph(graph) - - with pytest.raises(Exception, match='There are no rules'): - pr.reason() - - def test_reason_without_graph_uses_empty_graph(self): - """Test reasoning without graph uses empty graph and warns""" - pr.add_rule(pr.Rule('test(x) <- test2(x)', 'test_rule')) - - with pytest.warns(UserWarning, match='Graph not loaded'): - interpretation = pr.reason() - # Should complete without crashing - - def test_reason_auto_names_rules(self): - """Test that rules get auto-named when no name provided""" - pr.add_rule(pr.Rule('test1(x) <- test2(x)')) # No name - pr.add_rule(pr.Rule('test3(x) <- test4(x)')) # No name - - rules = pr.get_rules() - assert rules[0].get_rule_name() == 'rule_0' - assert rules[1].get_rule_name() == 'rule_1' - - def test_reason_auto_names_facts(self): - """Test that facts get auto-named when no name provided""" - fact1 = pr.Fact('test(node1)') # No name - fact2 = pr.Fact('test(node1, node2)') # No name - - pr.add_fact(fact1) - pr.add_fact(fact2) - - # Names should be auto-generated - assert fact1.name.startswith('fact_') - assert fact2.name.startswith('fact_') - - -class TestGraphAttributeParsing: - """Test graph attribute parsing branches""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_load_graph_with_attribute_parsing_enabled(self): - """Test loading graph with attribute parsing enabled""" - graph = nx.DiGraph() - graph.add_node("A", label="person", age=25) - graph.add_node("B", label="person", age=30) - graph.add_edge("A", "B", relation="knows", weight=0.8) - - pr.settings.graph_attribute_parsing = True - pr.load_graph(graph) - # Should complete without errors - - def test_load_graph_with_attribute_parsing_disabled(self): - """Test loading graph with attribute parsing disabled (lines 540-543, 562-565)""" - graph = nx.DiGraph() - graph.add_node("A", label="person") - graph.add_edge("A", "B", relation="knows") - - pr.settings.graph_attribute_parsing = False - pr.load_graph(graph) - # Should complete without errors and use empty collections - - -class TestOutputFunctionAssertions: - """Test output functions when store_interpretation_changes=False""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_save_rule_trace_assertion(self): - """Test save_rule_trace assertion when store_interpretation_changes=False""" - pr.settings.store_interpretation_changes = False - - with pytest.raises(AssertionError, match='store interpretation changes setting is off'): - pr.save_rule_trace(mock.MagicMock(), './test/') - - def test_get_rule_trace_assertion(self): - """Test get_rule_trace assertion when store_interpretation_changes=False""" - pr.settings.store_interpretation_changes = False - - with pytest.raises(AssertionError, match='store interpretation changes setting is off'): - pr.get_rule_trace(mock.MagicMock()) - - def test_filter_and_sort_nodes_assertion(self): - """Test filter_and_sort_nodes assertion when store_interpretation_changes=False""" - pr.settings.store_interpretation_changes = False - - with pytest.raises(AssertionError, match='store interpretation changes setting is off'): - pr.filter_and_sort_nodes(mock.MagicMock(), ['test']) - - def test_filter_and_sort_edges_assertion(self): - """Test filter_and_sort_edges assertion when store_interpretation_changes=False""" - pr.settings.store_interpretation_changes = False - - with pytest.raises(AssertionError, match='store interpretation changes setting is off'): - pr.filter_and_sort_edges(mock.MagicMock(), ['test']) - - -class TestReasoningModes: - """Test different reasoning modes and settings""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_reason_with_memory_profiling(self): - """Test reasoning with memory profiling enabled""" - # Set up minimal working example - graph = nx.DiGraph() - graph.add_edge("A", "B") - pr.load_graph(graph) - pr.add_rule(pr.Rule('test(x) <- test(y)', 'test_rule')) - pr.add_fact(pr.Fact('test(A)', 'test_fact')) - - pr.settings.memory_profile = True - pr.settings.verbose = False # Reduce output noise - - # Should complete without errors - interpretation = pr.reason(timesteps=1) - - def test_reason_with_output_to_file(self): - """Test reasoning with output_to_file enabled""" - # Set up minimal working example - graph = nx.DiGraph() - graph.add_edge("A", "B") - pr.load_graph(graph) - pr.add_rule(pr.Rule('test(x) <- test(y)', 'test_rule')) - pr.add_fact(pr.Fact('test(A)', 'test_fact')) - - pr.settings.output_to_file = True - pr.settings.output_file_name = "test_output" - - interpretation = pr.reason(timesteps=1) - - # Check if output file was created (and clean up) - import glob - output_files = glob.glob("test_output_*.txt") - for f in output_files: - os.unlink(f) - - def test_reason_again_functionality(self): - """Test reason again functionality (lines 688-693, 788-799)""" - # Set up initial reasoning - graph = nx.DiGraph() - graph.add_edge("A", "B") - pr.load_graph(graph) - pr.add_rule(pr.Rule('test(x) <- test(y)', 'test_rule')) - pr.add_fact(pr.Fact('test(A)', 'test_fact', 0, 1)) - - # First reasoning - interpretation1 = pr.reason(timesteps=1) - - # Add new fact and reason again - pr.add_fact(pr.Fact('test(B)', 'test_fact2', 2, 3)) - interpretation2 = pr.reason(timesteps=2, again=True, restart=False) - - # Should complete without errors - - -class TestAnnotationFunctions: - """Test annotation function management""" - - def test_add_annotation_function(self): - """Test adding annotation function""" - def test_func(annotations, weights): - return sum(w * a[0].lower for w, a in zip(weights, annotations)), 1.0 - - pr.add_annotation_function(test_func) - # Should complete without errors - - -class TestTorchIntegrationHandling: - """Test torch integration state consistency""" - - def test_torch_integration_consistency(self): - """Test that torch integration variables are consistent""" - # Just verify the current state is consistent - if hasattr(pr, 'LogicIntegratedClassifier'): - if pr.LogicIntegratedClassifier is None: - # If LogicIntegratedClassifier is None, ModelInterfaceOptions should also be None - assert pr.ModelInterfaceOptions is None - else: - # If LogicIntegratedClassifier exists, ModelInterfaceOptions should also exist - assert pr.ModelInterfaceOptions is not None - - -class TestQueryFiltering: - """Test query-based rule filtering""" - - def setup_method(self): - """Reset state before each test""" - pr.reset() - pr.reset_settings() - - def test_reason_with_queries(self): - """Test reasoning with query-based rule filtering""" - # Set up test scenario - graph = nx.DiGraph() - graph.add_edges_from([("A", "B"), ("B", "C")]) - pr.load_graph(graph) - - pr.add_rule(pr.Rule('popular(x) <-1 friend(x, y)', 'rule1')) - pr.add_rule(pr.Rule('friend(x, y) <-1 knows(x, y)', 'rule2')) - pr.add_fact(pr.Fact('knows(A, B)', 'fact1')) - - # Create query to filter rules - query = pr.Query('popular(A)') - pr.settings.verbose = False # Reduce output noise - - interpretation = pr.reason(timesteps=1, queries=[query]) - # Should complete and apply rule filtering logic - - -@pytest.mark.fp -class TestFixedPointVersions: - """Test key functionality with FP version enabled""" - - def setup_method(self): - """Reset settings before each test""" - pr.reset() - pr.reset_settings() - pr.settings.fp_version = True # Enable FP version for all tests in this class - - @pytest.mark.fp - @pytest.mark.skip(reason="Pure edge-to-edge transitive reasoning not supported in PyReason architecture. Both regular and FP versions fail this test. PyReason requires node clause anchors for multi-clause rule grounding.") - def test_basic_reasoning_fp(self): - """Test basic reasoning functionality with FP version""" - # Create simple graph - graph = nx.DiGraph() - graph.add_edge("A", "B") - graph.add_edge("B", "C") - pr.load_graph(graph) - - # Add rule and fact - pr.add_rule(pr.Rule('connected(x, z) <-1 connected(x, y), connected(y, z)', 'transitive_rule')) - pr.add_fact(pr.Fact('connected(A, B)', 'fact1')) - pr.add_fact(pr.Fact('connected(B, C)', 'fact2')) - - # Reason - interpretation = pr.reason(timesteps=2) - - # Verify transitivity worked - assert interpretation.query(pr.Query('connected(A, C)')), 'Should infer connected(A, C) via transitivity' - - @pytest.mark.fp - def test_settings_validation_fp(self): - """Test that settings validation works with FP version""" - # Test that fp_version setting is properly set - assert pr.settings.fp_version == True, 'FP version should be enabled' - - # Test that other settings still work - pr.settings.verbose = True - assert pr.settings.verbose == True, 'Verbose setting should work' - - -if __name__ == '__main__': - pytest.main([__file__]) diff --git a/tests/unit/api_tests/test_pyreason_file_loading.py b/tests/unit/api_tests/test_pyreason_file_loading.py index 2767d915..09ef3f63 100644 --- a/tests/unit/api_tests/test_pyreason_file_loading.py +++ b/tests/unit/api_tests/test_pyreason_file_loading.py @@ -703,22 +703,23 @@ def test_add_rules_from_file_simple_rules(self): finally: os.unlink(tmp_path) - def test_add_rules_from_file_with_comments(self): - """Test loading rules from file with comments.""" - rules_content = """# This is a comment -friend(A, B) <- knows(A, B) -# Another comment -enemy(A, B) <- ~friend(A, B) -# Final comment""" - - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as tmp: - tmp.write(rules_content) - tmp_path = tmp.name + def test_add_rules_from_file_with_comments_and_empty_lines(self): + """Test rule file parsing handles comments and empty lines""" + with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f: + f.write("# This is a comment\n") + f.write("\n") # Empty line + f.write(" \n") # Whitespace-only line + f.write("test_rule(x) <-1 other_rule(x)\n") + f.write("# Another comment\n") + f.write("another_rule(y) <-1 test_rule(y)\n") + temp_path = f.name try: - pr.add_rules_from_file(tmp_path) + pr.add_rules_from_file(temp_path) + rules = pr.get_rules() + assert len(rules) == 2 # Should only include the 2 actual rules finally: - os.unlink(tmp_path) + os.unlink(temp_path) def test_add_rules_from_file_with_empty_lines(self): """Test loading rules from file with empty lines.""" @@ -842,6 +843,13 @@ def test_add_rules_from_file_after_existing_rules(self): os.unlink(tmp_path) + def test_add_inconsistent_predicates(self): + """Test adding inconsistent predicate pairs""" + pr.add_inconsistent_predicate("pred1", "pred2") + pr.add_inconsistent_predicate("pred3", "pred4") + # Should not raise exceptions + + class TestRuleTrace: """Test save_rule_trace() and get_rule_trace() functions.""" diff --git a/tests/unit/api_tests/test_pyreason_reasoning.py b/tests/unit/api_tests/test_pyreason_reasoning.py index ba371aa4..20a2ec49 100644 --- a/tests/unit/api_tests/test_pyreason_reasoning.py +++ b/tests/unit/api_tests/test_pyreason_reasoning.py @@ -22,25 +22,13 @@ def setup_method(self): pr.reset() pr.reset_settings() - def test_reason_with_no_graph_loads_empty_graph(self): - """Test reasoning without loading a graph (should load empty graph with warning).""" - # Don't load any graph - pr.add_rule(Rule("test(A) <- test(A)", "test_rule", False)) + def test_reason_without_graph_uses_empty_graph(self): + """Test reasoning without graph uses empty graph and warns""" + pr.add_rule(pr.Rule('test(x) <- test2(x)', 'test_rule')) - # Capture stdout to check for warning - captured_output = StringIO() - original_stdout = sys.stdout - pr.settings.verbose = True - - try: - sys.stdout = captured_output - interpretation = pr.reason(timesteps=1) - output = captured_output.getvalue() - - # Should contain warning about no graph - assert "Graph not loaded" in output or interpretation is not None - finally: - sys.stdout = original_stdout + with pytest.warns(UserWarning, match='Graph not loaded'): + interpretation = pr.reason() + # Should complete without crashing def test_reason_with_no_rules_raises_exception(self): """Test reasoning without any rules raises exception.""" @@ -51,6 +39,27 @@ def test_reason_with_no_rules_raises_exception(self): with pytest.raises(Exception, match="There are no rules"): pr.reason(timesteps=1) + def test_reason_auto_names_rules(self): + """Test that rules get auto-named when no name provided""" + pr.add_rule(pr.Rule('test1(x) <- test2(x)')) # No name + pr.add_rule(pr.Rule('test3(x) <- test4(x)')) # No name + + rules = pr.get_rules() + assert rules[0].get_rule_name() == 'rule_0' + assert rules[1].get_rule_name() == 'rule_1' + + def test_reason_auto_names_facts(self): + """Test that facts get auto-named when no name provided""" + fact1 = pr.Fact('test(node1)') # No name + fact2 = pr.Fact('test(node1, node2)') # No name + + pr.add_fact(fact1) + pr.add_fact(fact2) + + # Names should be auto-generated + assert fact1.name.startswith('fact_') + assert fact2.name.startswith('fact_') + def test_reason_with_output_to_file(self): """Test reasoning with output_to_file setting.""" graph = nx.DiGraph() diff --git a/tests/unit/api_tests/test_pyreason_state_management.py b/tests/unit/api_tests/test_pyreason_state_management.py index c500be72..91c709f8 100644 --- a/tests/unit/api_tests/test_pyreason_state_management.py +++ b/tests/unit/api_tests/test_pyreason_state_management.py @@ -173,6 +173,17 @@ def setup_method(self): pr.reset() pr.reset_settings() + def test_torch_integration_consistency(self): + """Test that torch integration variables are consistent""" + # Just verify the current state is consistent + if hasattr(pr, 'LogicIntegratedClassifier'): + if pr.LogicIntegratedClassifier is None: + # If LogicIntegratedClassifier is None, ModelInterfaceOptions should also be None + assert pr.ModelInterfaceOptions is None + else: + # If LogicIntegratedClassifier exists, ModelInterfaceOptions should also exist + assert pr.ModelInterfaceOptions is not None + def test_state_isolation_between_operations(self): """Test that state is properly isolated between operations.""" # This test verifies that subsequent operations don't interfere