Add CLI tests

Author: IlyaMuravjov

test/cfpq_decomposer_cli/test_decompose_cflr_matrix.py

@@ -0,0 +1,34 @@
+import os
+from unittest.mock import patch
+
+import pytest
+
+from cfpq_decomposer_cli.decompose_cflr_matrix import main as decompose_main
+from src.utils.useful_paths import POCR_FORMAT_DATA
+from test.cfpq_decomposer_cli.utils import parse_decomposer_cli_output, CorruptDecomposer
+from test.utils import find_graph_file, find_grammar_file
+
+EXPECTED_SEDGES = 3_968_276
+DATA_PATH = os.path.join(POCR_FORMAT_DATA, 'leela')
+GRAPH_PATH = find_graph_file(DATA_PATH)
+GRAMMAR_PATH = find_grammar_file(DATA_PATH)
+
+
+@pytest.mark.CI
+@pytest.mark.parametrize("args", [[], ["--prototype"]], ids=["default", "prototype"])
+def test_cli_decompose_cflr_matrix_modes(args, capsys):
+    decompose_main([GRAPH_PATH, GRAMMAR_PATH] + args)
+    captured = capsys.readouterr()
+    metrics = parse_decomposer_cli_output(captured.out)
+
+    assert metrics.s_edges == EXPECTED_SEDGES, f"#SEdges mismatch for args={args}"
+    assert metrics.compression_factor > 10, f"Compression factor too low for args={args}"
+    assert metrics.is_valid, f"Compression invalid for args={args}"
+
+@pytest.mark.CI
+@patch('cfpq_decomposer_cli.decompose_cflr_matrix.HighPerformanceDecomposer', new=CorruptDecomposer)
+def test_cli_decompose_cflr_matrix_corrupted(monkeypatch, capsys):
+    decompose_main([GRAPH_PATH, GRAMMAR_PATH])
+    captured = capsys.readouterr()
+    metrics = parse_decomposer_cli_output(captured.out)
+    assert not metrics.is_valid, "Expected compression to be invalid when decomposer is corrupted"

test/cfpq_decomposer_cli/utils.py

@@ -0,0 +1,44 @@
+from dataclasses import dataclass
+from typing import Tuple
+
+from graphblas.core.matrix import Matrix
+
+from cfpq_decomposer.high_performance_decomposer import HighPerformanceDecomposer
+from cfpq_matrix.matrix_utils import stack
+
+
+@dataclass
+class DecomposerCliMetrics:
+    s_edges: int
+    compression_factor: float
+    is_valid: bool
+
+def parse_decomposer_cli_output(output: str) -> DecomposerCliMetrics:
+    lines = output.strip().splitlines()
+    sedges_line = next(l for l in lines if l.startswith('#SEdges'))
+    s_edges = int(sedges_line.split()[1])
+    cf_line = next(l for l in lines if l.startswith('Compression factor'))
+    compression_factor = float(cf_line.split()[2])
+    valid_line = next(l for l in lines if l.startswith('Is compression valid'))
+    is_valid = valid_line.split()[3] == 'True'
+    return DecomposerCliMetrics(
+        s_edges=s_edges,
+        compression_factor=compression_factor,
+        is_valid=is_valid
+    )
+
+class CorruptDecomposer(HighPerformanceDecomposer):
+    def decompose(self, matrix: Matrix) -> Tuple[Matrix, Matrix]:
+        left, right = super().decompose(matrix)
+        nrows, _ = left.shape
+        _, ncols = right.shape
+
+        rows = list(range(nrows))
+        cols = [0] * nrows
+        extra_col = Matrix.from_coo(rows, cols, True, nrows=nrows, ncols=1)
+
+        rows2 = [0] * ncols
+        cols2 = list(range(ncols))
+        extra_row = Matrix.from_coo(rows2, cols2, True, nrows=1, ncols=ncols)
+
+        return stack([[left, extra_col]]), stack([[right], [extra_row]])

test/pocr_data/leela/aa.cnf

@@ -0,0 +1,17 @@
+M	DV	d
+DV	dbar	V
+V	A_r	V
+V	V	A
+V	FV_i	f_i
+V	M
+V
+FV_i	fbar_i	V
+A	a	M
+A	a
+A
+A_r	M	abar
+A_r	abar
+A_r
+
+Count:
+V