Improve CSV loader tests

Author: ColtonPayne

tests/api_tests/test_files/example_facts.csv

@@ -1,9 +1,19 @@
 fact_text,name,start_time,end_time,static
 Viewed(Zach),seen-fact-zach,0,3,False
-Viewed(Justin),seen-fact-justin,0,3,False
-Viewed(Michelle),seen-fact-michelle,1,3,False
-Viewed(Amy),seen-fact-amy,2,3,False
+Viewed(Justin),seen-fact-justin,0,3,true
+Viewed(Michelle),seen-fact-michelle,1,3,FALSE
+Viewed(Amy),seen-fact-amy,2,3,0
 "HaveAccess(Zach,TextMessage)",access-zach,0,5,True
-"HaveAccess(Justin,TextMessage)",access-justin,0,5,True
-"HaveAccess(Michelle,TextMessage)",access-michelle,0,5,True
-"HaveAccess(Amy,TextMessage)",access-amy,0,5,True
+"HaveAccess(Justin,TextMessage)",access-justin,0,5,1
+"HaveAccess(Michelle,TextMessage)",access-michelle,0,5,yes
+"HaveAccess(Amy,TextMessage)",access-amy,0,5,no
+"Processed(Node1):[0.5,0.8]",interval-node,0,10,False
+"Knows(Person1,Person2):[0.7,0.9]",interval-edge,0,10,True
+Viewed(Valid),valid-fact,0,3,False
+,empty-fact-text,0,3,False
+InvalidSyntax,bad-syntax,0,3,False
+"Viewed(OutOfRange):[2.5,3.0]",out-of-range,0,3,False
+Viewed(BadStartTime),bad-start,abc,5,False
+Viewed(BadEndTime),bad-end,0,xyz,False
+Viewed(BadStaticValue),bad-static,0,5,invalid
+Viewed(EmptyOptionals),,,,

tests/api_tests/test_pyreason_file_loading.py

@@ -858,21 +858,58 @@ def setup_method(self):
         pr.reset()
         pr.reset_settings()
 
-    def test_add_fact_in_bulk_with_header(self):
-        """Test loading facts from CSV with header row."""
-        csv_content = """fact_text,name,start_time,end_time,static
-Viewed(Zach),seen-fact-zach,0,3,False
-Viewed(Justin),seen-fact-justin,0,3,False
-"HaveAccess(Amy,TextMessage)",access-fact,0,5,True"""
+    def test_add_fact_in_bulk_comprehensive(self):
+        """Test loading facts from CSV with various valid and invalid scenarios.
+
+        This test uses example_facts.csv which contains:
+        - Valid node facts with various boolean formats
+        - Valid edge facts with various boolean formats
+        - Node and edge facts with interval bounds
+        - Empty fact_text (should warn)
+        - Invalid syntax (should warn)
+        - Out of range intervals (should warn)
+        - Invalid start_time (should warn)
+        - Invalid end_time (should warn)
+        - Invalid static value (should warn)
+        - Empty optional fields
+        """
+        csv_path = 'pyreason/tests/api_tests/test_files/example_facts.csv'
+
+        # Expect warnings for rows with invalid data:
+        # - Row with empty fact_text
+        # - Row with invalid syntax (missing parentheses)
+        # - Row with out-of-range interval values
+        # - Row with invalid start_time
+        # - Row with invalid end_time
+        # - Row with invalid static value
+        with pytest.warns(UserWarning) as warning_list:
+            pr.add_fact_in_bulk(csv_path)
 
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp:
-            tmp.write(csv_content)
-            tmp_path = tmp.name
+        # Verify that we got warnings (at least 6 from the invalid rows)
+        assert len(warning_list) >= 6, f"Expected at least 6 warnings, got {len(warning_list)}"
 
-        try:
-            pr.add_fact_in_bulk(tmp_path)
-        finally:
-            os.unlink(tmp_path)
+        # Check that specific warning messages appear
+        warning_messages = [str(w.message) for w in warning_list]
+
+        # Verify warning for empty fact_text
+        assert any("Missing required 'fact_text'" in msg for msg in warning_messages), \
+            "Expected warning about missing fact_text"
+
+        # Verify warning for invalid syntax
+        assert any("Failed to parse fact" in msg and "bad-syntax" in msg for msg in warning_messages), \
+            "Expected warning about invalid syntax"
+
+        # Verify warning for invalid start_time
+        assert any("Invalid start_time" in msg for msg in warning_messages), \
+            "Expected warning about invalid start_time"
+
+        # Verify warning for invalid end_time
+        assert any("Invalid end_time" in msg for msg in warning_messages), \
+            "Expected warning about invalid end_time"
+
+        # Verify warning for invalid static value
+        assert any("Invalid static value" in msg for msg in warning_messages), \
+            "Expected warning about invalid static value"
 
     def test_add_fact_in_bulk_without_header(self):
         """Test loading facts from CSV without header row."""
@@ -920,8 +957,8 @@ def test_add_fact_in_bulk_empty_optional_fields(self):
         finally:
             os.unlink(tmp_path)
 
-    def test_add_fact_in_bulk_with_invalid_facts(self):
-        """Test that invalid facts show warnings but don't crash."""
+    def test_add_fact_in_bulk_with_invalid_facts_warnings(self):
+        """Test that invalid facts produce specific warnings but don't crash."""
         csv_content = """fact_text,name,start_time,end_time,static
 Viewed(Valid),valid-fact,0,3,False
 ,empty-fact-text,0,3,False
@@ -933,8 +970,18 @@ def test_add_fact_in_bulk_with_invalid_facts(self):
             tmp_path = tmp.name
 
         try:
-            # Should not raise exception, only warnings
-            pr.add_fact_in_bulk(tmp_path)
+            # Should produce warnings for the 3 invalid rows
+            with pytest.warns(UserWarning) as warning_list:
+                pr.add_fact_in_bulk(tmp_path)
+
+            # Verify we got at least 3 warnings (one for each invalid row)
+            assert len(warning_list) >= 3, f"Expected at least 3 warnings, got {len(warning_list)}"
+
+            warning_messages = [str(w.message) for w in warning_list]
+
+            # Verify specific warnings appear
+            assert any("Missing required 'fact_text'" in msg for msg in warning_messages)
+            assert any("Failed to parse fact" in msg for msg in warning_messages)
         finally:
             os.unlink(tmp_path)
 
@@ -956,77 +1003,9 @@ def test_add_fact_in_bulk_empty_file(self):
         finally:
             os.unlink(tmp_path)
 
-    def test_add_fact_in_bulk_node_facts(self):
-        """Test loading node facts (no commas in components)."""
-        csv_content = """fact_text,name,start_time,end_time,static
-Viewed(Node1),fact1,0,5,False
-Viewed(Node2),fact2,1,5,True
-"Processed(Node3):[0.5,0.8]",fact3,0,10,False"""
-
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp:
-            tmp.write(csv_content)
-            tmp_path = tmp.name
-
-        try:
-            pr.add_fact_in_bulk(tmp_path)
-        finally:
-            os.unlink(tmp_path)
-
-    def test_add_fact_in_bulk_edge_facts(self):
-        """Test loading edge facts (with commas, properly quoted)."""
-        csv_content = """fact_text,name,start_time,end_time,static
-"Connected(A,B)",edge1,0,5,False
-"Related(X,Y):True",edge2,1,5,True
-"Knows(Person1,Person2):[0.7,0.9]",edge3,0,10,False"""
-
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp:
-            tmp.write(csv_content)
-            tmp_path = tmp.name
-
-        try:
-            pr.add_fact_in_bulk(tmp_path)
-        finally:
-            os.unlink(tmp_path)
-
-    def test_add_fact_in_bulk_mixed_fact_types(self):
-        """Test loading both node and edge facts in same file."""
-        csv_content = """fact_text,name,start_time,end_time,static
-Viewed(Alice),node-fact,0,3,False
-"Connected(Alice,Bob)",edge-fact,0,3,False
-"Processed(Charlie):[0.5,0.8]",node-interval,1,5,True
-"Knows(X,Y):[0.3,0.7]",edge-interval,2,5,False"""
-
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp:
-            tmp.write(csv_content)
-            tmp_path = tmp.name
-
-        try:
-            pr.add_fact_in_bulk(tmp_path)
-        finally:
-            os.unlink(tmp_path)
-
-    def test_add_fact_in_bulk_various_static_values(self):
-        """Test loading facts with various static value formats."""
-        csv_content = """fact_text,name,start_time,end_time,static
-Viewed(A),fact1,0,5,True
-Viewed(B),fact2,0,5,true
-Viewed(C),fact3,0,5,FALSE
-Viewed(D),fact4,0,5,0
-Viewed(E),fact5,0,5,1
-Viewed(F),fact6,0,5,yes
-Viewed(G),fact7,0,5,no"""
-
-        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp:
-            tmp.write(csv_content)
-            tmp_path = tmp.name
-
-        try:
-            pr.add_fact_in_bulk(tmp_path)
-        finally:
-            os.unlink(tmp_path)
 
     def test_add_fact_in_bulk_invalid_start_end_times(self):
-        """Test loading facts with invalid start/end time values."""
+        """Test loading facts with invalid start/end time values produce warnings."""
         csv_content = """fact_text,name,start_time,end_time,static
 Viewed(A),fact1,abc,5,False
 Viewed(B),fact2,0,xyz,False"""
@@ -1037,7 +1016,19 @@ def test_add_fact_in_bulk_invalid_start_end_times(self):
 
         try:
             # Should load facts with warnings about invalid times (using defaults)
-            pr.add_fact_in_bulk(tmp_path)
+            with pytest.warns(UserWarning) as warning_list:
+                pr.add_fact_in_bulk(tmp_path)
+
+            # Verify we got warnings for both invalid time values
+            assert len(warning_list) >= 2, f"Expected at least 2 warnings, got {len(warning_list)}"
+
+            warning_messages = [str(w.message) for w in warning_list]
+
+            # Verify specific warnings appear
+            assert any("Invalid start_time" in msg for msg in warning_messages), \
+                "Expected warning about invalid start_time"
+            assert any("Invalid end_time" in msg for msg in warning_messages), \
+                "Expected warning about invalid end_time"
         finally:
             os.unlink(tmp_path)
 
@@ -1061,21 +1052,6 @@ def test_add_fact_in_bulk_multiple_calls(self):
             os.unlink(tmp1_path)
             os.unlink(tmp2_path)
 
-    def test_add_fact_in_bulk_example_file_with_header(self):
-        """Test loading facts from example CSV file with header."""
-        csv_path = 'tests/api_tests/test_files/example_facts.csv'
-
-        # Only run if example file exists
-        if os.path.exists(csv_path):
-            pr.add_fact_in_bulk(csv_path)
-
-    def test_add_fact_in_bulk_example_file_without_header(self):
-        """Test loading facts from example CSV file without header."""
-        csv_path = 'tests/api_tests/test_files/example_facts_no_header.csv'
-
-        # Only run if example file exists
-        if os.path.exists(csv_path):
-            pr.add_fact_in_bulk(csv_path)
 
 
 class TestRuleTrace: