diff --git a/tests/ingest/test_ingest_from_csv().py b/tests/ingest/test_ingest_from_csv().py index 3810d94f..33e46ba5 100644 --- a/tests/ingest/test_ingest_from_csv().py +++ b/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 + + +