Share validated TGF parsing across training pipelines

.github/workflows/tests.yml

@@ -0,0 +1,22 @@
+name: Component tests
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: '3.11'
+      - run: python -m pip install pytest
+      - run: python -m pytest -q tests

README.md

@@ -26,6 +26,27 @@ The shell scripts document the experiment workflows. Review their data paths, mo
 
 [AFGCN](https://github.com/lmlearning/AFGCN) · [AFGraphLib](https://github.com/lmlearning/AFGraphLib) · [FastAFGCN](https://github.com/lmlearning/FastAFGCN)
 
-## License
+## TGF input validation
+
+`graph_io.py` provides the shared parser used by `pyg_train.py` and the refined
+training pipeline. It accepts the unlabelled TGF subset used here: one argument
+identifier per line, a single `#` separator, then two attack endpoints per line.
+Blank lines are ignored and spaces or tabs can separate endpoints. Argument and
+attack order, isolated arguments and self-attacks are preserved.
+
+Duplicate argument identifiers, malformed attacks, undeclared endpoints and missing
+or repeated separators raise `ValueError` with file context. Optional TGF node/edge
+labels are not supported. `train_refined_afgcn.parseTGF` retains its existing
+list-of-lists attack return type.
+
+```bash
+python -m pip install pytest
+python -m pytest tests
+```
+
+These component tests need no graph-learning libraries or checkpoints. They validate
+input handling; they do not reproduce training or evaluate model accuracy.
+
+## License
 
 See [LICENSE](LICENSE).

graph_io.py

@@ -0,0 +1,40 @@
+"""Dependency-free input parsing shared by the graph-training entry points."""
+
+
+def parse_tgf(path):
+    """Read the unlabelled TGF subset used by the argumentation datasets.
+
+    Preserve argument/attack order, ignore blank lines, and reject malformed
+    records before NetworkX can silently create undeclared arguments.
+    """
+    arguments, attacks = [], []
+    declared = set()
+    seen_separator = False
+    with open(path, encoding="utf-8") as source:
+        for line_number, line in enumerate(source, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            context = f"{path}:{line_number}: "
+            if line == "#":
+                if seen_separator:
+                    raise ValueError(context + "duplicate TGF separator")
+                seen_separator = True
+                continue
+            fields = line.split()
+            if not seen_separator:
+                if len(fields) != 1:
+                    raise ValueError(context + "expected one unlabelled argument identifier")
+                if line in declared:
+                    raise ValueError(context + "duplicate argument identifier")
+                arguments.append(line)
+                declared.add(line)
+            else:
+                if len(fields) != 2:
+                    raise ValueError(context + "expected two attack endpoints")
+                if any(endpoint not in declared for endpoint in fields):
+                    raise ValueError(context + "attack endpoint is not a declared argument")
+                attacks.append(tuple(fields))
+    if not seen_separator:
+        raise ValueError(f"{path}: missing TGF '#' separator")
+    return arguments, attacks

pyg_train.py

@@ -22,6 +22,7 @@
 import pickle
 from pathlib import Path
 from typing import Dict, List, Tuple
+from graph_io import parse_tgf as _parse_tgf
 
 import networkx as nx
 import numpy as np
@@ -45,16 +46,7 @@
 
 
 def parse_tgf(file: Path) -> Tuple[List[str], List[Tuple[str, str]]]:
-    args, atts, seen_hash = [], [], False
-    with file.open() as f:
-        for ln in f:
-            ln = ln.strip()
-            if ln == "#":
-                seen_hash = True
-                continue
-            (args if not seen_hash else atts).append(ln)
-    atts = [tuple(e.split()) for e in atts]
-    return args, atts
+    return _parse_tgf(file)
 
 
 def reindex_nodes(g: nx.DiGraph) -> Tuple[nx.DiGraph, Dict[str, int]]:

tests/test_graph_io.py

@@ -0,0 +1,39 @@
+import pytest
+
+from graph_io import parse_tgf
+
+
+def parse(tmp_path, text):
+    path = tmp_path / "graph.tgf"
+    path.write_text(text, encoding="utf-8")
+    return parse_tgf(path)
+
+
+def test_whitespace_does_not_create_nodes_or_attacks(tmp_path):
+    assert parse(tmp_path, "\nalpha\n\nbeta\nisolated\n # \n\nalpha\t beta\nbeta  beta\n\n") == (
+        ["alpha", "beta", "isolated"], [("alpha", "beta"), ("beta", "beta")]
+    )
+
+
+def test_argument_order_is_preserved(tmp_path):
+    assert parse(tmp_path, "10\n2\n1\n#\n2 10\n1 2\n") == (
+        ["10", "2", "1"], [("2", "10"), ("1", "2")]
+    )
+
+
+def test_empty_framework(tmp_path):
+    assert parse(tmp_path, "\n#\n") == ([], [])
+
+
+@pytest.mark.parametrize("text,message", [
+    ("a\n", "missing TGF"),
+    ("a\n#\n#\n", "duplicate TGF"),
+    ("a\na\n#\n", "duplicate argument"),
+    ("a label\n#\n", "unlabelled"),
+    ("a\n#\na\n", "two attack endpoints"),
+    ("a\n#\na a extra\n", "two attack endpoints"),
+    ("a\n#\na b\n", "not a declared argument"),
+])
+def test_invalid_graphs_fail_with_filename_context(tmp_path, text, message):
+    with pytest.raises(ValueError, match="graph.tgf.*" + message):
+        parse(tmp_path, text)

train_refined_afgcn.py

@@ -16,6 +16,7 @@
 """
 
 import os, random, argparse, pickle, math, time
+from graph_io import parse_tgf
 import numpy as np, networkx as nx
 from tqdm import tqdm
 from sklearn.preprocessing import StandardScaler
@@ -74,18 +75,9 @@ def categoriser_ranking(adj_matrix, max_iter=1000, tol=1e-9):
 
 
 def parseTGF(path):
-    with open(path) as f:
-        lines = [ln.strip() for ln in f]
-    args, atts, hash_seen = [], [], False
-    for ln in lines:
-        if ln == '#':
-            hash_seen = True
-            continue
-        if not hash_seen:
-            args.append(ln)
-        else:
-            atts.append(ln.split(' '))
-    return args, atts
+    # Preserve the list-of-lists return type used by existing callers.
+    arguments, attacks = parse_tgf(path)
+    return arguments, [list(attack) for attack in attacks]
 
 
 def parse_apx(path):
@@ -386,4 +378,4 @@ def evaluate(model, loader, loss_fn, device):
             best_mcc = mcc
             torch.save({'epoch':ep,'model_state_dict':model.state_dict()},
                        args.checkpoint)
-            print(f"   ↳ new best ({mcc:.4f}) saved to {args.checkpoint}")
+            print(f"   ↳ new best ({mcc:.4f}) saved to {args.checkpoint}")