feature: support table reader functionality

Signed-off-by: Timothée Mazzucotelli <dev@pawamoy.fr>

python/tests/unit/extensions/test_macros.py

@@ -25,13 +25,18 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
+import pandas
 import pytest
 from jinja2.exceptions import TemplateSyntaxError, UndefinedError
 
 from zensical.extensions.context import ContextPreprocessor
 from zensical.extensions.macros import (
     MacroEnv,
+    _add_indentation,
+    _convert_to_md_table,
     _fix_url,
+    _get_fake_table_readers,
+    _get_table_readers,
     _load_module,
     _load_one_yaml,
     _merge_include_yaml,
@@ -42,6 +47,7 @@ if TYPE_CHECKING:
     from pathlib import Path
 
     from markdown import Markdown
+    from pandas import DataFrame
 
 
 # ---------------------------------------------------------------------------
@@ -443,3 +449,271 @@ class TestPreprocessor:
         result = md.convert("{{ code_snippet('python', 'x = 1 + 1') }}")
         assert "<code>python" not in result
         assert "x = 1 + 1" not in result  # we expect spans
+
+
+# ---------------------------------------------------------------------------
+# Table helpers
+# ---------------------------------------------------------------------------
+
+
+# Unit tests for the table conversion helper functions.
+class TestTableHelpers:
+    def test_add_indentation_spaces(self) -> None:
+        result = _add_indentation("line1\nline2", spaces=4)
+        assert result == "    line1\n    line2"
+
+    def test_add_indentation_tabs(self) -> None:
+        result = _add_indentation("line1\nline2", tabs=2)
+        assert result == "\t\tline1\n\t\tline2"
+
+    def test_add_indentation_none_returns_unchanged(self) -> None:
+        assert _add_indentation("hello") == "hello"
+
+    def test_add_indentation_raises_when_both_specified(self) -> None:
+        with pytest.raises(ValueError, match="spaces or tabs"):
+            _add_indentation("x", spaces=2, tabs=1)
+
+    def test_convert_to_md_table_basic(self) -> None:
+        df: DataFrame = pandas.DataFrame(
+            {"Name": ["Alice", "Bob"], "Age": [30, 25]}
+        )
+        result = _convert_to_md_table(df)
+        assert "|" in result
+        assert "Name" in result
+        assert "Age" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_convert_to_md_table_escapes_pipes_in_cells(self) -> None:
+        df: DataFrame = pandas.DataFrame({"Col": ["a|b", "c"]})
+        result = _convert_to_md_table(df)
+        assert r"a\|b" in result
+
+    def test_convert_to_md_table_escapes_pipes_in_column_names(self) -> None:
+        df: DataFrame = pandas.DataFrame({"Na|me": ["Alice"]})
+        result = _convert_to_md_table(df)
+        assert r"Na\|me" in result
+
+    def test_convert_to_md_table_omits_index_by_default(self) -> None:
+        # Custom index values must not leak into the output.
+        # Verifies that the `index=False` default is applied.
+        df: DataFrame = pandas.DataFrame({"X": [1, 2]}, index=[100, 200])
+        result = _convert_to_md_table(df)
+        assert "100" not in result
+        assert "200" not in result
+
+
+# ---------------------------------------------------------------------------
+# Table readers
+# ---------------------------------------------------------------------------
+
+
+class TestTableReaders:
+    def test_fake_readers_raise_runtime_error_when_pandas_missing(
+        self,
+    ) -> None:
+        readers = _get_fake_table_readers()
+        for reader in readers.values():
+            with pytest.raises(RuntimeError, match="table reading requires"):
+                reader("irrelevant.csv")
+
+    # CSV
+    def test_read_csv(self, tmp_path: Path) -> None:
+        (tmp_path / "data.csv").write_text(
+            "Name,Age\nAlice,30\nBob,25\n", encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_csv"]("data.csv")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_csv_returns_dataframe(self, tmp_path: Path) -> None:
+        (tmp_path / "data.csv").write_text("X,Y\n1,2\n3,4\n", encoding="utf-8")
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_csv"]("data.csv")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # JSON
+    def test_read_json(self, tmp_path: Path) -> None:
+        (tmp_path / "data.json").write_text(
+            '[{"Name": "Alice", "Age": 30}, {"Name": "Bob", "Age": 25}]',
+            encoding="utf-8",
+        )
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_json"]("data.json")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_json_returns_dataframe(self, tmp_path: Path) -> None:
+        (tmp_path / "data.json").write_text(
+            '[{"X": 1, "Y": 2}, {"X": 3, "Y": 4}]', encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_json"]("data.json")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # YAML
+    def test_read_yaml(self, tmp_path: Path) -> None:
+        (tmp_path / "data.yaml").write_text(
+            "- Name: Alice\n  Age: 30\n- Name: Bob\n  Age: 25\n",
+            encoding="utf-8",
+        )
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_yaml"]("data.yaml")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_yaml_returns_dataframe(self, tmp_path: Path) -> None:
+        (tmp_path / "data.yaml").write_text(
+            "- X: 1\n  Y: 2\n- X: 3\n  Y: 4\n", encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_yaml"]("data.yaml")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # Table (tab-separated)
+    def test_read_table(self, tmp_path: Path) -> None:
+        (tmp_path / "data.tsv").write_text(
+            "Name\tAge\nAlice\t30\nBob\t25\n", encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_table"]("data.tsv")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_table_returns_dataframe(self, tmp_path: Path) -> None:
+        (tmp_path / "data.tsv").write_text(
+            "X\tY\n1\t2\n3\t4\n", encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_table"]("data.tsv")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # FWF (fixed-width format)
+    def test_read_fwf(self, tmp_path: Path) -> None:
+        content = "Name     Age\nAlice    30\nBob      25\n"
+        (tmp_path / "data.fwf").write_text(content, encoding="utf-8")
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_fwf"]("data.fwf")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_fwf_returns_dataframe(self, tmp_path: Path) -> None:
+        (tmp_path / "data.fwf").write_text(
+            "X    Y\n1    2\n3    4\n", encoding="utf-8"
+        )
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_fwf"]("data.fwf")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # Excel (.xlsx)
+    def test_read_excel(self, tmp_path: Path) -> None:
+        pytest.importorskip("openpyxl")
+        df: DataFrame = pandas.DataFrame(
+            {"Name": ["Alice", "Bob"], "Age": [30, 25]}
+        )
+        df.to_excel(tmp_path / "data.xlsx", index=False)
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_excel"]("data.xlsx")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_excel_returns_dataframe(self, tmp_path: Path) -> None:
+        pytest.importorskip("openpyxl")
+        df_in: DataFrame = pandas.DataFrame({"X": [1, 3], "Y": [2, 4]})
+        df_in.to_excel(tmp_path / "data.xlsx", index=False)
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_excel"]("data.xlsx")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # Feather
+    def test_read_feather(self, tmp_path: Path) -> None:
+        pytest.importorskip("pyarrow")
+        df: DataFrame = pandas.DataFrame(
+            {"Name": ["Alice", "Bob"], "Age": [30, 25]}
+        )
+        df.to_feather(tmp_path / "data.feather")
+        readers = _get_table_readers(tmp_path)
+        result = readers["read_feather"]("data.feather")
+        assert "Name" in result
+        assert "Alice" in result
+        assert "Bob" in result
+
+    def test_pd_read_feather_returns_dataframe(self, tmp_path: Path) -> None:
+        pytest.importorskip("pyarrow")
+        df_in: DataFrame = pandas.DataFrame({"X": [1, 3], "Y": [2, 4]})
+        df_in.to_feather(tmp_path / "data.feather")
+        readers = _get_table_readers(tmp_path)
+        df: DataFrame = readers["pd_read_feather"]("data.feather")
+        assert list(df.columns) == ["X", "Y"]
+        assert len(df) == 2
+
+    # End-to-end: CSV rendered through the Jinja2 / Markdown pipeline
+    @pytest.mark.parametrize(
+        "md",
+        [
+            pytest.param(
+                {
+                    "config": {
+                        "markdown_extensions": {
+                            "zensical.extensions.macros": {
+                                "render_by_default": True,
+                            },
+                        },
+                    }
+                },
+                id="end_to_end_csv",
+            ),
+        ],
+        indirect=["md"],
+    )
+    def test_end_to_end_csv_via_template(
+        self, md: Markdown, tmp_path: Path
+    ) -> None:
+        (tmp_path / "scores.csv").write_text(
+            "Player,Score\nAlice,100\nBob,80\n", encoding="utf-8"
+        )
+        result = md.convert("{{ read_csv('scores.csv') }}")
+        assert "Player" in result
+        assert "Alice" in result
+        assert "Score" in result
+
+    @pytest.mark.parametrize(
+        "md",
+        [
+            pytest.param(
+                {
+                    "config": {
+                        "markdown_extensions": {
+                            "zensical.extensions.macros": {
+                                "render_by_default": True,
+                            },
+                        },
+                    }
+                },
+                id="end_to_end_pd_filter",
+            ),
+        ],
+        indirect=["md"],
+    )
+    def test_end_to_end_pd_read_csv_with_convert_filter(
+        self, md: Markdown, tmp_path: Path
+    ) -> None:
+        (tmp_path / "nums.csv").write_text("X,Y\n1,2\n3,4\n", encoding="utf-8")
+        result = md.convert(
+            "{{ pd_read_csv('nums.csv') | convert_to_md_table }}"
+        )
+        assert "X" in result
+        assert "Y" in result