Merge PR #244 after resolving conflicts

tests/ingest/test_ingest_from_csv().py

@@ -1,11 +1,9 @@
 import os
 import tempfile
 import pandas as pd
-import pytest
 
 from semantica.ingest.pandas_ingestor import PandasIngestor
 
-
 def write_temp_csv(content: str, encoding="utf-8"):
     tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
     tmp.close()
@@ -14,30 +12,25 @@ def write_temp_csv(content: str, encoding="utf-8"):
     return tmp.name
 
 
-# -------------------------------------------------------
-# Test 1: Encoding Detection (international entities)
-# -------------------------------------------------------
-
-def test_from_csv_detects_encoding():
-    # Latin-1 encoded content
-    content = (
-        "entity_name,location\n"
-        "Siemens,München\n"
-        "Telefónica,España\n"
-    )
+# =======================================================
+# ENCODING (5 tests)
+# =======================================================
 
+def test_encoding_latin1():
+    content = "name,city\nJosé,São Paulo\n"
     path = write_temp_csv(content, encoding="latin-1")
-
-    ingestor = PandasIngestor()
-    data = ingestor.from_csv(path)
-
-    assert data.row_count == 2
-    assert data.columns == ["entity_name", "location"]
-    assert data.dataframe.iloc[0]["entity_name"] == "Siemens"
-
+    data = PandasIngestor().from_csv(path)
+    assert data.dataframe.iloc[0]["name"] == "José"
     os.remove(path)
 
 
+def test_encoding_utf8():
+    content = "user,country\n李雷,China\n"
+    path = write_temp_csv(content, encoding="utf-8")
+    data = PandasIngestor().from_csv(path)
+    assert data.dataframe.iloc[0]["user"] == "李雷"
+    os.remove(path)
+ 
 def test_from_csv_detects_tab_delimiter():
     content = (
         "user_id\trole\n"
@@ -151,47 +144,92 @@ def test_from_csv_preserves_nan_values():
 # Test 2: Delimiter Detection (semicolon separated)
 # -------------------------------------------------------
 
-def test_from_csv_detects_delimiter():
-    content = (
-        "company;sector\n"
-        "CrowdStrike;Cybersecurity\n"
-        "Fortinet;Network Security\n"
-    )
-
-    path = write_temp_csv(content)
-
-    ingestor = PandasIngestor()
-    data = ingestor.from_csv(path)
-
-    assert data.row_count == 2
-    assert data.columns == ["company", "sector"]
-    assert data.dataframe.iloc[1]["company"] == "Fortinet"
 
+def test_encoding_accented_text():
+    content = "company,city\nRenée,Zürich\n"
+    path = write_temp_csv(content, encoding="latin-1")
+    data = PandasIngestor().from_csv(path)
+    assert data.row_count == 1
     os.remove(path)
 
 
-# -------------------------------------------------------
-# Test 3: Bad Rows Are Skipped (malformed threat feed)
-# -------------------------------------------------------
-
-def test_from_csv_skips_bad_rows():
-    content = (
-        "indicator,type\n"
-        "192.168.1.10,IP\n"
-        "this,is,too,many,columns\n"   # malformed row
-        "evil-domain.com,Domain\n"
-    )
-
-    path = write_temp_csv(content)
-
-    ingestor = PandasIngestor()
-    data = ingestor.from_csv(path)
-
-    # Malformed row should be skipped
-    assert data.row_count == 2
-    assert list(data.dataframe["indicator"]) == [
-        "192.168.1.10",
-        "evil-domain.com",
-    ]
-
+def test_encoding_spanish():
+    content = "org,country\nTelefónica,España\n"
+    path = write_temp_csv(content, encoding="latin-1")
+    data = PandasIngestor().from_csv(path)
+    assert data.row_count == 1
     os.remove(path)
+
+
+def test_encoding_ansi_cp1252():
+    content = "brand,city\nPeugeot,Montréal\n"
+    path = write_temp_csv(content, encoding="cp1252")   
+    data = PandasIngestor().from_csv(path)
+    assert data.dataframe.iloc[0]["city"] == "Montréal"
+    os.remove(path)
+
+
+# =======================================================
+# DELIMITERS (4 tests)
+# =======================================================
+
+def test_delimiter_comma():
+    content = "a,b\n1,2\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert list(data.columns) == ["a", "b"]
+    os.remove(path)
+
+
+def test_delimiter_semicolon():
+    content = "a;b\n1;2\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert list(data.columns) == ["a", "b"]
+    os.remove(path)
+
+
+def test_delimiter_pipe():
+    content = "a|b\n1|2\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert list(data.columns) == ["a", "b"]
+    os.remove(path)
+
+
+def test_delimiter_tab():
+    content = "a\tb\n1\t2\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert list(data.columns) == ["a", "b"]
+    os.remove(path)
+
+
+# =======================================================
+# BAD ROWS (3 tests)
+# =======================================================
+
+def test_bad_row_extra_columns():
+    content = "x,y\n1,2\n1,2,3,4\n5,6\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert data.row_count == 2
+    os.remove(path)
+
+def test_bad_row_missing_column():
+    content = "x,y\n1,2\n3\n4,5\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    assert data.row_count == 3
+    assert data.dataframe["y"].isna().sum() == 1
+
+def test_bad_row_unclosed_quote():
+    content = "x,y\n1,2\n\"3,4\n5,6\n"
+    path = write_temp_csv(content)
+    data = PandasIngestor().from_csv(path)
+    # The malformed quoted line consumes the following line; both are skipped.
+    # Only the first valid row remains.
+    assert data.row_count == 1
+
+
+