diff --git a/src/ariadnepy/plot/_custom.py b/src/ariadnepy/plot/_custom.py index 6676796..4d6eeaa 100644 --- a/src/ariadnepy/plot/_custom.py +++ b/src/ariadnepy/plot/_custom.py @@ -86,6 +86,15 @@ def add_resource( df = df.iloc[:, :2] from_col, to_col = df.columns[0], df.columns[1] + # Validate that at least one feature is already in the graph (mirrors R's addResource check). + existing = set(graph.vs["name"]) if graph.vcount() > 0 else set() + new_vars = {col for col in (from_col, to_col) if col not in existing} + if len(new_vars) == 2: + raise AriadneError( + f"At least one feature must already be in the graph. " + f"Neither {from_col!r} nor {to_col!r} was found." + ) + # Cache the linkmap as parquet cache_dir = init_cache() safe = "".join(c if c.isalnum() else "_" for c in str(file))[:80] @@ -105,15 +114,23 @@ def add_resource( graph = graph.copy() # Add vertices if missing - existing = set(graph.vs["name"]) if graph.vcount() > 0 else set() for node in (from_col, to_col): if node not in existing: graph.add_vertex(name=node) existing.add(node) - # Add edge + # Check for duplicate edge (same node pair + same res_name) before adding. src_idx = graph.vs.find(name=from_col).index tgt_idx = graph.vs.find(name=to_col).index + if not force: + for e in graph.es: + pair = {graph.vs[e.source]["name"], graph.vs[e.target]["name"]} + if pair == {from_col, to_col} and e.attributes().get("source") == res_name: + raise AriadneError( + f"A '{res_name}' edge between {from_col!r} and {to_col!r} already exists. " + "Set force=True to overwrite it." + ) + graph.add_edge(src_idx, tgt_idx) for k, v in edge_attrs.items(): graph.es[-1][k] = v diff --git a/src/ariadnepy/plot/_draw.py b/src/ariadnepy/plot/_draw.py index 5e39e6e..44c899c 100644 --- a/src/ariadnepy/plot/_draw.py +++ b/src/ariadnepy/plot/_draw.py @@ -75,7 +75,8 @@ def plot_path( path_edges.append((row["from"], row["to"])) path_edge_labels[(row["from"], row["to"])] = row.get("source", "") - path_edge_set = set(path_edges) + # Use frozensets so edge lookup is direction-agnostic, matching R's sort(c(from,to)) approach. + path_edge_set = {frozenset(e) for e in path_edges} # prune/focus both restrict layout to the path subgraph. if path_nodes and (focus or prune): @@ -88,8 +89,8 @@ def plot_path( else: draw_graph = graph - # Fruchterman-Reingold layout — equivalent to R ariadne's "stress" layout - layout = draw_graph.layout("fr") + # Kamada-Kawai layout — closest available equivalent to R ariadne's "stress" layout + layout = draw_graph.layout("kk") all_names = draw_graph.vs["name"] pos = {name: tuple(layout[i]) for i, name in enumerate(all_names)} @@ -111,7 +112,7 @@ def plot_path( for u, v, src in all_edges_raw: if u not in pos or v not in pos: continue - is_path_edge = (u, v) in path_edge_set or (v, u) in path_edge_set + is_path_edge = frozenset((u, v)) in path_edge_set color = _C_EDGE_PATH if is_path_edge else _C_EDGE_BG lw = 2.0 if is_path_edge else 1.0 x0, y0 = pos[u] diff --git a/tests/test_core/test_graph/test_add_resource.py b/tests/test_core/test_graph/test_add_resource.py new file mode 100644 index 0000000..52c1b99 --- /dev/null +++ b/tests/test_core/test_graph/test_add_resource.py @@ -0,0 +1,130 @@ +"""Tests for plot/_custom.py::add_resource. + +add_resource connects CSV *column names* as graph nodes (not row values). +The fixture CSV has columns 'known_node' and 'new_node', so those two strings +become the candidate vertex names. The base graph below contains 'known_node' +as a pre-existing vertex so that the "at least one feature must already be in +the graph" invariant is satisfied for all positive-path tests. + +All tests use a synthetic offline undirected graph — no network calls. +""" +from __future__ import annotations + +import igraph as ig +import pandas as pd +import pytest + +from ariadnepy.exceptions import AriadneError +from ariadnepy.plot._custom import add_resource + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def base_graph() -> ig.Graph: + """Minimal undirected graph that already contains 'known_node'. + + Having 'known_node' pre-existing means a CSV whose first column is named + 'known_node' will pass the "at least one feature in graph" guard. + """ + g = ig.Graph(directed=False) + g.add_vertex(name="known_node") + return g + + +@pytest.fixture +def tmp_linkmap(tmp_path) -> str: + """CSV with columns 'known_node' and 'new_node' — two rows of sample data.""" + df = pd.DataFrame({"known_node": ["x", "y"], "new_node": ["p", "q"]}) + p = tmp_path / "map.csv" + df.to_csv(p, index=False) + return str(p) + + +@pytest.fixture +def both_new_linkmap(tmp_path) -> str: + """CSV whose column names ('alpha', 'beta') are both absent from the base graph. + + Used to verify that add_resource raises AriadneError when neither column + name is already present as a graph vertex. + """ + df = pd.DataFrame({"alpha": ["a", "b"], "beta": ["c", "d"]}) + p = tmp_path / "both_new.csv" + df.to_csv(p, index=False) + return str(p) + + +# ── Return-type test ────────────────────────────────────────────────────────── + + +def test_add_resource_returns_igraph(base_graph, tmp_linkmap): + """add_resource must return an igraph.Graph instance.""" + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB") + assert isinstance(result, ig.Graph) + + +# ── Edge addition tests ─────────────────────────────────────────────────────── + + +def test_add_resource_adds_new_edge(base_graph, tmp_linkmap): + """A new edge must be present in the returned graph.""" + before = base_graph.ecount() + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB") + assert result.ecount() == before + 1 + + +def test_add_resource_edge_has_source_attribute(base_graph, tmp_linkmap): + """The newly added edge must carry the 'source' attribute set to res_name.""" + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB") + assert result.es[-1]["source"] == "TestDB" + + +# ── Node addition test ──────────────────────────────────────────────────────── + + +def test_add_resource_adds_new_node(base_graph, tmp_linkmap): + """The column name 'new_node' (absent from base_graph) must appear as a vertex.""" + before_names = set(base_graph.vs["name"]) + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB") + after_names = set(result.vs["name"]) + assert "new_node" in after_names - before_names + + +# ── Validation-gap test — both columns new ─────────────────────────────────── + + +def test_add_resource_raises_when_both_columns_new(base_graph, both_new_linkmap): + """add_resource must raise AriadneError when neither CSV column name is in + the graph. + + This guards against silently grafting a disconnected subgraph — mirroring + R's addResource check that at least one feature type must already exist. + """ + with pytest.raises(AriadneError, match="At least one feature"): + add_resource(base_graph, both_new_linkmap, res_name="Disconnected") + + +# ── One-column-known test ───────────────────────────────────────────────────── + + +def test_add_resource_succeeds_when_one_column_known(base_graph, tmp_linkmap): + """add_resource must succeed (not raise) when one column name is already in + the graph as a vertex — 'known_node' is pre-existing in base_graph.""" + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB") + assert "known_node" in result.vs["name"] + + +# ── Duplicate-edge guard (force=False) ─────────────────────────────────────── + + +def test_add_resource_force_false_raises_on_duplicate(base_graph, tmp_linkmap): + """Calling add_resource twice with force=False must raise AriadneError on the + second call to prevent silently creating duplicate edges for the same resource. + + This test documents the intended validation behaviour; the implementation + must enforce it. + """ + result = add_resource(base_graph, tmp_linkmap, res_name="TestDB", force=False) + with pytest.raises(AriadneError): + add_resource(result, tmp_linkmap, res_name="TestDB", force=False) diff --git a/tests/test_core/test_graph/test_plot.py b/tests/test_core/test_graph/test_plot.py new file mode 100644 index 0000000..455ac06 --- /dev/null +++ b/tests/test_core/test_graph/test_plot.py @@ -0,0 +1,108 @@ +"""Tests for plot/_draw.py::plot_path. + +All tests use a synthetic offline undirected graph — no network calls. +Matplotlib is forced into the non-interactive Agg backend so the suite +runs headlessly in CI. +""" +from __future__ import annotations + +import igraph as ig +import matplotlib +import pytest + +matplotlib.use("Agg") # headless — must be set before importing pyplot + +from matplotlib.figure import Figure + +from ariadnepy.plot._draw import plot_path + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def simple_graph() -> ig.Graph: + """Minimal undirected graph: A -- B -- C with a single resource label.""" + g = ig.Graph(directed=False) + for n in ("A", "B", "C"): + g.add_vertex(name=n) + g.add_edge(0, 1) + g.es[0]["source"] = "DB1" + g.add_edge(1, 2) + g.es[1]["source"] = "DB1" + return g + + +# ── Return-type tests ───────────────────────────────────────────────────────── + + +def test_plot_path_no_by_returns_figure(simple_graph): + """plot_path(graph) with no 'by' formula must return a matplotlib Figure.""" + fig = plot_path(simple_graph) + assert isinstance(fig, Figure) + + +def test_plot_path_by_k1_returns_figure(simple_graph): + """plot_path(graph, 'A ~ C', k=1) must return a matplotlib Figure.""" + fig = plot_path(simple_graph, "A ~ C", k=1) + assert isinstance(fig, Figure) + + +def test_plot_path_prune_returns_figure(simple_graph): + """plot_path with prune=True must return a matplotlib Figure.""" + fig = plot_path(simple_graph, "A ~ C", k=1, prune=True) + assert isinstance(fig, Figure) + + +def test_plot_path_focus_returns_figure(simple_graph): + """plot_path with focus=True must return a matplotlib Figure.""" + fig = plot_path(simple_graph, "A ~ C", k=1, focus=True) + assert isinstance(fig, Figure) + + +# ── Title tests ─────────────────────────────────────────────────────────────── + + +def test_plot_path_no_by_title_is_resource_graph(simple_graph): + """When no path formula is given the figure title must be 'Resource Graph'.""" + fig = plot_path(simple_graph) + ax = fig.axes[0] + assert ax.get_title() == "Resource Graph" + + +def test_plot_path_by_k2_title_contains_path_2(simple_graph): + """With k=2 the figure title must contain 'Path 2' (mirrors R plotPath behaviour).""" + # Build a branching graph so two distinct shortest paths exist. + g = ig.Graph(directed=False) + for n in ("A", "M1", "M2", "C"): + g.add_vertex(name=n) + g.add_edge(g.vs.find(name="A").index, g.vs.find(name="M1").index) + g.es[-1]["source"] = "DB1" + g.add_edge(g.vs.find(name="M1").index, g.vs.find(name="C").index) + g.es[-1]["source"] = "DB1" + g.add_edge(g.vs.find(name="A").index, g.vs.find(name="M2").index) + g.es[-1]["source"] = "DB1" + g.add_edge(g.vs.find(name="M2").index, g.vs.find(name="C").index) + g.es[-1]["source"] = "DB1" + + fig = plot_path(g, "A ~ C", k=2) + ax = fig.axes[0] + assert "Path 2" in ax.get_title() + + +# ── Validation tests ────────────────────────────────────────────────────────── + + +def test_plot_path_prune_without_by_raises(simple_graph): + """prune=True without a 'by' formula must raise ValueError.""" + with pytest.raises(ValueError, match="prune"): + plot_path(simple_graph, prune=True) + + +# ── Axes structure tests ────────────────────────────────────────────────────── + + +def test_plot_path_returns_exactly_one_axes(simple_graph): + """The returned Figure must contain exactly one Axes object.""" + fig = plot_path(simple_graph) + assert len(fig.axes) == 1