Improve decomposer tests

IlyaMuravjov · IlyaMuravjov · commit 8ad194a2386c · 2025-05-04T00:31:08.000+03:00
diff --git a/test/cfpq_decomposer/measure_compression_factors.py b/test/cfpq_decomposer/measure_compression_factors.py
@@ -0,0 +1,43 @@
+# helper script needed to calculate safe values for
+# compression factor test asserts
+
+import numpy as np
+import graphblas
+from tqdm import tqdm
+import graphblas as gb
+from graphblas.core.matrix import Matrix
+from graphblas.core.dtypes import BOOL
+
+from cfpq_decomposer.high_performance_decomposer import HighPerformanceDecomposer
+from cfpq_decomposer.prototype_decomposer import PrototypeDecomposer
+from test.cfpq_decomposer.synthetic_data import similar_rows_matrix, multiple_patterns_matrix, double_threshold_matrix, \
+    similar_columns_matrix, random_matrix_with_patterns
+
+MATRIX_FNS = {
+    "similar_rows": similar_rows_matrix,
+    "multiple_patterns": multiple_patterns_matrix,
+    "double_threshold": double_threshold_matrix,
+    "similar_columns": similar_columns_matrix,
+    "random_patterns": random_matrix_with_patterns,
+}
+
+def compute_case_quantile(gen_fn, runs=2500):
+    values = []
+    for cls in (HighPerformanceDecomposer, PrototypeDecomposer):
+        desc = f"{gen_fn.__name__}::{cls.__name__}"
+        for _ in tqdm(range(runs), desc=desc, leave=False):
+            M = gen_fn()
+            left, right = cls().decompose(M)
+            left_right = left.mxm(right, op=graphblas.semiring.any_pair).new(dtype=BOOL)
+            rem = M.dup(mask=~left_right.S).nvals
+            values.append(M.nvals / (left.nvals + right.nvals + rem))
+    return np.percentile(values, 0.1)
+
+if __name__ == "__main__":
+    quantiles = {}
+    for key, fn in tqdm(MATRIX_FNS.items(), desc="Cases"):
+        quantiles[key] = compute_case_quantile(fn)
+    print("MIN_COMPRESSION_FACTORS = {")
+    for key, val in quantiles.items():
+        print(f'    "{key}": {val:.3f},')
+    print("}")
diff --git a/test/cfpq_decomposer/synthetic_data.py b/test/cfpq_decomposer/synthetic_data.py
@@ -0,0 +1,109 @@
+import numpy as np
+from graphblas.core.dtypes import BOOL
+from graphblas.core.matrix import Matrix
+
+MIN_COMPRESSION_FACTORS = {
+    "similar_rows": 1.5,
+    "multiple_patterns": 11,
+    "double_threshold": 11,
+    "similar_columns": 3.5,
+    "random_patterns": 2.5,
+}
+
+def similar_rows_matrix():
+    nrows, ncols = 15, 15
+    M = Matrix(BOOL, nrows=nrows, ncols=ncols)
+    base_row_indices = [0, 1, 2, 3, 4, 5]
+    for i in range(10):
+        if i % 3 == 0:
+            M[i, 3] = True
+        for j in base_row_indices:
+            M[i, j] = True
+    for i in range(10, nrows):
+        M[i, i % ncols] = True
+    return M
+
+def multiple_patterns_matrix():
+    nrows, ncols = 300, 100
+    M = Matrix(BOOL, nrows=nrows, ncols=ncols)
+    for i in range(100):
+        for j in range(20):
+            M[i, j] = True
+        if i % 10 == 0:
+            M[i, 25] = True
+    for i in range(100, 200):
+        for j in range(30, 50):
+            M[i, j] = True
+        if i % 15 == 0:
+            M[i, 55] = True
+    for i in range(200, 300):
+        for j in range(60, 80):
+            M[i, j] = True
+        if i % 20 == 0:
+            M[i, 85] = True
+    return M
+
+def double_threshold_matrix():
+    nrows, ncols = 100, 50
+    M = Matrix(BOOL, nrows=nrows, ncols=ncols)
+    for i in range(nrows):
+        for j in range(10):
+            M[i, j] = True
+    for i in range(75):
+        for j in range(10, 20):
+            M[i, j] = True
+    for i in range(74):
+        for j in range(20, 30):
+            M[i, j] = True
+    for i in range(76):
+        for j in range(30, 40):
+            M[i, j] = True
+    for i in range(80):
+        for j in range(40, 50):
+            M[i, j] = True
+    return M
+
+def similar_columns_matrix():
+    nrows, ncols = 100, 200
+    M = Matrix(BOOL, nrows=nrows, ncols=ncols)
+    for j in range(50):
+        for i in range(80):
+            M[i, j] = True
+        if j % 10 == 0:
+            for i in range(80, 85):
+                M[i, j] = True
+    for i in range(nrows):
+        for _ in range(5):
+            j = np.random.randint(50, ncols)
+            M[i, j] = True
+    return M
+
+def random_matrix_with_patterns():
+    nrows, ncols = 500, 500
+    M = Matrix(BOOL, nrows=nrows, ncols=ncols)
+    for group in range(5):
+        rows = range(group * 100, group * 100 + 100)
+        cols = np.random.choice(ncols, size=50, replace=False)
+        for i in rows:
+            for j in cols:
+                M[i, j] = True
+            if i % 25 == 0:
+                extra = np.random.choice(ncols, size=5, replace=False)
+                for j in extra:
+                    M[i, j] = True
+    for group in range(5):
+        cols = range(group * 100, group * 100 + 100)
+        rows = np.random.choice(nrows, size=50, replace=False)
+        for j in cols:
+            for i in rows:
+                M[i, j] = True
+            if j % 25 == 0:
+                extra = np.random.choice(nrows, size=5, replace=False)
+                for i in extra:
+                    M[i, j] = True
+    noise = int(M.nvals * 0.05)
+    for _ in range(noise):
+        i = np.random.randint(0, nrows)
+        j = np.random.randint(0, ncols)
+        M[i, j] = True
+    return M
diff --git a/test/cfpq_decomposer/test_abstract_decomposer.py b/test/cfpq_decomposer/test_abstract_decomposer.py
@@ -1,162 +1,41 @@
 from abc import ABC, abstractmethod
 from typing import Tuple
 
-from graphblas.core.matrix import Matrix
-from graphblas.core.dtypes import BOOL
-import numpy as np
 import graphblas
+import pytest
+from graphblas.core.dtypes import BOOL
+from graphblas.core.matrix import Matrix
 
 from cfpq_decomposer.decomposer import Decomposer
-
-class TestAbstractDecomposer(ABC):
+from test.cfpq_decomposer.synthetic_data import (
+    similar_rows_matrix,
+    multiple_patterns_matrix,
+    double_threshold_matrix,
+    similar_columns_matrix,
+    random_matrix_with_patterns, MIN_COMPRESSION_FACTORS
+)
+
+class AbstractDecomposerTest(ABC):
     @abstractmethod
     def create_decomposer(self) -> Decomposer:
         pass
 
     def decompose(self, matrix: Matrix) -> Tuple[Matrix, Matrix]:
         return self.create_decomposer().decompose(matrix)
 
-    def test_decompose_similar_rows_matrix(self):
-        nrows, ncols = 15, 15
-        M = Matrix(BOOL, nrows=nrows, ncols=ncols)
-        base_row_indices = [0, 1, 2, 3, 4, 5]
-        for i in range(10):
-            if i % 3 == 0:
-                M[i, 3] = True
-            for j in base_row_indices:
-                M[i, j] = True
-        for i in range(10, nrows):
-            M[i, i % ncols] = True
-        LEFT, RIGHT = self.decompose(M)
-        LEFT_RIGHT: Matrix = LEFT.mxm(RIGHT, op=graphblas.semiring.any_pair).new(dtype=BOOL)
-
-        assert LEFT_RIGHT.dup(mask=~M.S).nvals == 0
-        assert LEFT_RIGHT.nvals >= 18
-
-    def test_decompose_multiple_patterns(self):
-        nrows, ncols = 300, 100
-        M = Matrix(BOOL, nrows=nrows, ncols=ncols)
-
-        pattern1_cols = set(range(20))
-        for i in range(100):
-            for j in pattern1_cols:
-                M[i, j] = True
-            if i % 10 == 0:
-                M[i, 25] = True
-
-        pattern2_cols = set(range(30, 50))
-        for i in range(100, 200):
-            for j in pattern2_cols:
-                M[i, j] = True
-            if i % 15 == 0:
-                M[i, 55] = True
-
-        pattern3_cols = set(range(60, 80))
-        for i in range(200, 300):
-            for j in pattern3_cols:
-                M[i, j] = True
-            if i % 20 == 0:
-                M[i, 85] = True
-
-        LEFT, RIGHT = self.decompose(M)
-        LEFT_RIGHT = LEFT.mxm(RIGHT, op=graphblas.semiring.any_pair).new(dtype=BOOL)
-
-        assert M.nvals == 6022
-        assert LEFT_RIGHT.nvals >= 6000
-        assert (LEFT_RIGHT | LEFT_RIGHT).new(mask=~M.S).nvals == 0
-
-    def test_decompose_double_thresholding(self):
-        nrows, ncols = 100, 50
-        M = Matrix(BOOL, nrows=nrows, ncols=ncols)
-
-        for i in range(nrows):
-            for j in range(10):
-                M[i, j] = True
-
-        for i in range(75):
-            for j in range(10, 20):
-                M[i, j] = True
-
-        for i in range(74):
-            for j in range(20, 30):
-                M[i, j] = True
-
-        for i in range(76):
-            for j in range(30, 40):
-                M[i, j] = True
-
-        for i in range(80):
-            for j in range(40, 50):
-                M[i, j] = True
-
-        # Call the decompose function
-        LEFT, RIGHT = self.decompose(M)
-        LEFT_RIGHT = LEFT.mxm(RIGHT, op=graphblas.semiring.any_pair).new(dtype=BOOL)
-
-        assert M.nvals == 4050
-        assert LEFT_RIGHT.nvals >= 4000
-        assert (LEFT_RIGHT | LEFT_RIGHT).new(mask=~M.S).nvals == 0
-
-    def test_decompose_similar_columns_without_transpose(self):
-        nrows, ncols = 100, 200
-        M: Matrix = Matrix(BOOL, nrows=nrows, ncols=ncols)
-
-        for j in range(50):
-            for i in range(80):
-                M[i, j] = True
-            if j % 10 == 0:
-                for i in range(80, 85):
-                    M[i, j] = True
-
-        for i in range(nrows):
-            for _ in range(5):
-                j = np.random.randint(50, ncols)
-                M[i, j] = True
-
-        LEFT, RIGHT = self.decompose(M)
-        LEFT_RIGHT = LEFT.mxm(RIGHT, op=graphblas.semiring.any_pair).new(dtype=BOOL)
-
-        assert M.nvals in range(4400, 4600)
-        assert LEFT_RIGHT.nvals >= 3900
-        assert (LEFT_RIGHT | LEFT_RIGHT).new(mask=~M.S).nvals == 0
-
-    def test_decompose_random_matrix_with_patterns(self):
-        nrows, ncols = 500, 500
-        M = Matrix(BOOL, nrows=nrows, ncols=ncols)
-
-        for group in range(5):
-            row_start = group * 100
-            row_end = row_start + 100
-            cols = np.random.choice(ncols, size=50, replace=False)
-            for i in range(row_start, row_end):
-                for j in cols:
-                    M[i, j] = True
-                if i % 25 == 0:
-                    extra_cols = np.random.choice(ncols, size=5, replace=False)
-                    for j in extra_cols:
-                        M[i, j] = True
-
-        for group in range(5):
-            col_start = group * 100
-            col_end = col_start + 100
-            rows = np.random.choice(nrows, size=50, replace=False)
-            for j in range(col_start, col_end):
-                for i in rows:
-                    M[i, j] = True
-                if j % 25 == 0:
-                    extra_rows = np.random.choice(nrows, size=5, replace=False)
-                    for i in extra_rows:
-                        M[i, j] = True
-
-        num_noise_entries = int(M.nvals * 0.05)
-        for _ in range(num_noise_entries):
-            i = np.random.randint(0, nrows)
-            j = np.random.randint(0, ncols)
-            M[i, j] = True
-
-        LEFT, RIGHT = self.decompose(M)
-        LEFT_RIGHT = LEFT.mxm(RIGHT, op=graphblas.semiring.any_pair).new(dtype=BOOL)
-
-        assert M.nvals in range(48_000, 52_000)
-        assert LEFT_RIGHT.nvals >= 33_000
-        assert (LEFT_RIGHT | LEFT_RIGHT).new(mask=~M.S).nvals == 0
+    @pytest.mark.parametrize("matrix_fn,key", [
+        pytest.param(similar_rows_matrix, "similar_rows", id="similar_rows"),
+        pytest.param(multiple_patterns_matrix, "multiple_patterns", id="multiple_patterns"),
+        pytest.param(double_threshold_matrix, "double_threshold", id="double_threshold"),
+        pytest.param(similar_columns_matrix, "similar_columns", id="similar_columns"),
+        pytest.param(random_matrix_with_patterns, "random_patterns", id="random_patterns"),
+    ])
+    def test_decomposition(self, matrix_fn, key):
+        matrix = matrix_fn()
+        left, right = self.decompose(matrix)
+        left_right = left.mxm(right, op=graphblas.semiring.any_pair).new(dtype=BOOL)
+        assert left_right.dup(mask=~matrix.S).nvals == 0
+        remainder = matrix.dup(mask=~left_right.S).nvals
+        compression_factor = matrix.nvals / (left.nvals + right.nvals + remainder)
+        print(f"Compression factor for {key} is {compression_factor}")
+        assert compression_factor > MIN_COMPRESSION_FACTORS[key]
diff --git a/test/cfpq_decomposer/test_high_performance_decomposer.py b/test/cfpq_decomposer/test_high_performance_decomposer.py
@@ -1,7 +1,7 @@
 from cfpq_decomposer.decomposer import Decomposer
 from cfpq_decomposer.high_performance_decomposer import HighPerformanceDecomposer
-from test.cfpq_decomposer.test_abstract_decomposer import TestAbstractDecomposer
+from test.cfpq_decomposer.test_abstract_decomposer import AbstractDecomposerTest
 
-class TestHighPerformanceDecomposer(TestAbstractDecomposer):
+class TestHighPerformanceDecomposer(AbstractDecomposerTest):
     def create_decomposer(self) -> Decomposer:
         return HighPerformanceDecomposer()
diff --git a/test/cfpq_decomposer/test_prototype_decomposer.py b/test/cfpq_decomposer/test_prototype_decomposer.py
@@ -1,8 +1,8 @@
 from cfpq_decomposer.decomposer import Decomposer
 from cfpq_decomposer.prototype_decomposer import PrototypeDecomposer
-from test.cfpq_decomposer.test_abstract_decomposer import TestAbstractDecomposer
+from test.cfpq_decomposer.test_abstract_decomposer import AbstractDecomposerTest
 
 
-class TestPrototypeDecomposer(TestAbstractDecomposer):
+class TestPrototypeDecomposer(AbstractDecomposerTest):
     def create_decomposer(self) -> Decomposer:
         return PrototypeDecomposer()
diff --git a/test/conftest.py b/test/conftest.py
@@ -18,8 +18,12 @@ def pytest_configure(config):
 
 @pytest.fixture(params=os.listdir(POCR_FORMAT_DATA))
 def pocr_data_path(request):
-    base_path = os.path.abspath(POCR_FORMAT_DATA)
-    subfolder_path = os.path.join(base_path, request.param)
-    if os.path.isdir(subfolder_path):
-        return subfolder_path
-    pytest.skip(f"{request.param} is not a directory")
+    base_path: str = os.path.abspath(POCR_FORMAT_DATA)
+    subfolder_path: str = os.path.join(base_path, request.param)
+    if not os.path.isdir(subfolder_path):
+        pytest.skip(f"{request.param} is not a directory")
+        return None
+    if not os.path.exists(os.path.join(subfolder_path, "expected_all_pairs_reachability.txt")):
+        pytest.skip(f"{request.param} doesn't contain 'expected_all_pairs_reachability.txt'")
+        return None
+    return subfolder_path
