From 83c85d11d09c6df256b7363a16e43a12a10a4f74 Mon Sep 17 00:00:00 2001 From: Stefano Braghin <527806+stefano81@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:47:38 +0100 Subject: [PATCH 1/3] feat: add Flash Signed-off-by: Stefano Braghin <527806+stefano81@users.noreply.github.com> --- src/risk_assessment/anonymization/flash.py | 465 +++++++++++++++++++ tests/anonymization/test_flash.py | 504 +++++++++++++++++++++ 2 files changed, 969 insertions(+) create mode 100644 src/risk_assessment/anonymization/flash.py create mode 100644 tests/anonymization/test_flash.py diff --git a/src/risk_assessment/anonymization/flash.py b/src/risk_assessment/anonymization/flash.py new file mode 100644 index 0000000..04a9760 --- /dev/null +++ b/src/risk_assessment/anonymization/flash.py @@ -0,0 +1,465 @@ +"""Flash: Efficient, Stable and Optimal K-Anonymity algorithm. + +Flash traverses the generalization lattice in a bottom-up, breadth-first +manner while continuously building and binary-searching vertical paths +("lightning-flash" paths) to find the globally optimal k-anonymous +generalization with minimal information loss. + +The algorithm is described in: + Kohlmayer, F., Prasser, F., Eckert, C., Kemper, A., & Kuhn, K. A. (2012). + Flash: Efficient, Stable and Optimal K-Anonymity. IEEE SocialCom-PASSAT. + +Key properties compared to OLA: +- Stable execution time regardless of column ordering in the input data. +- Uses a three-criterion ordering (c1/c2/c3) to induce a total order on all + lattice nodes, guaranteeing deterministic traversal. +- Combines binary-search over paths with a min-heap of non-anonymous + boundary nodes to continuously seed new paths. + +The key components are: + +- :class:`FlashOptions` — algorithm configuration. +- :class:`FlashLattice` — generalization lattice with Flash traversal logic. +- :class:`Flash` — top-level algorithm implementing + :class:`~risk_assessment.anonymization.AnonymizationAlgorithm`. + +Example:: + + from risk_assessment.anonymization import KAnonymity + from risk_assessment.anonymization.flash import Flash, FlashOptions + + options = FlashOptions(privacy_constraints=[KAnonymity(k=5)], suppression=5.0) + flash = Flash(options) + anonymized_df, report = flash.anonymize(df, column_information) +""" + +from __future__ import annotations + +import heapq +import math +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from pandas import DataFrame + +from risk_assessment.anonymization import AnonymizationAlgorithm, AnonymizationReport, PrivacyConstraint +from risk_assessment.anonymization.optimal_lattice_anonymization import ( + AnonymityChecker, + LatticeNode, + _calculate_product, + _generalized_dataset, + _partition_dataset, +) +from risk_assessment.metrics.informationloss import ColumnInformation, ColumnType, categorical_precision +from risk_assessment.utility.hierarchy import GeneralizationHierarchy + + +@dataclass +class FlashOptions: + """Configuration for the Flash anonymization algorithm. + + Attributes: + privacy_constraints: One or more constraints every equivalence class + must satisfy. + suppression: Maximum percentage of rows that may be suppressed + (0.0 = no suppression allowed). Defaults to 0.0. + information_loss: Callable used to measure information loss between the + original and generalized datasets. Defaults to + :func:`~risk_assessment.metrics.informationloss.categorical_precision`. + """ + + privacy_constraints: list[PrivacyConstraint] + suppression: float = 0.0 + information_loss: Callable[[DataFrame, DataFrame, list[ColumnInformation]], float] = categorical_precision + + +# --------------------------------------------------------------------------- +# Node priority (traversal order) +# --------------------------------------------------------------------------- + + +def _node_priority( + node: LatticeNode, + max_levels: list[int], + distinct_counts: list[list[int]], +) -> tuple[int, float, float]: + """Return the (c1, c2, c3) ordering vector for *node*. + + The three criteria induce a total order that prefers lower generalization + and is used consistently throughout the algorithm to guarantee stable, + input-independent traversal. + + c1 — sum of all per-column generalization levels (= lattice level). + c2 — average fractional generalization across quasi-identifiers. + c3 — 1 minus the average fraction of distinct values remaining (more + generalized → fewer distinct values → higher c3). + """ + j = len(node.values) + if j == 0: + return (0, 0.0, 0.0) + + # c1: lattice level + c1: int = node.sum() + + # c2: average normalized generalization level + c2: float = sum(node.values[i] / max_levels[i] for i in range(j) if max_levels[i] > 0) / j + + # c3: 1 – (average fraction of distinct values that remain) + distinct_fractions: list[float] = [] + for i in range(j): + level_counts = distinct_counts[i] + base = level_counts[0] if level_counts[0] > 0 else 1 + level = min(node.values[i], len(level_counts) - 1) + distinct_fractions.append(level_counts[level] / base) + c3: float = 1.0 - (sum(distinct_fractions) / j) + + return (c1, c2, c3) + + +# --------------------------------------------------------------------------- +# Flash lattice +# --------------------------------------------------------------------------- + + +class FlashLattice: + """Generalization lattice with Flash traversal. + + Builds the full lattice once and then applies the Flash algorithm to find + the globally optimal anonymous node (minimal information loss while still + satisfying all privacy constraints within the suppression budget). + + Args: + anonymity_checker: Evaluates suppression rate and information loss. + column_information: Per-column metadata. + suppression: Maximum allowed suppression percentage. + """ + + def __init__( + self, + anonymity_checker: AnonymityChecker, + column_information: list[ColumnInformation], + suppression: float, + ) -> None: + self._checker = anonymity_checker + self._column_information = column_information + self._suppression = suppression + + self._quasi_columns: list[int] = [ + idx for idx, c_i in enumerate(column_information) if c_i.column_type == ColumnType.QUASI + ] + + # Per quasi-identifier: maximum level index and hierarchy + self._max_levels: list[int] = [] + self._hierarchies: list[GeneralizationHierarchy] = [] + + for qi_idx in self._quasi_columns: + c_i = column_information[qi_idx] + hierarchy = c_i.hierarchy + if hierarchy is None: + raise ValueError(f"Missing hierarchy for column {qi_idx}") + self._max_levels.append(len(hierarchy) - 1) + self._hierarchies.append(hierarchy) + + n_qi = len(self._quasi_columns) + + # Precompute distinct value counts per (qi, level) for c3 criterion. + # distinct_counts[i][l] = number of distinct generalized values at level l + self._distinct_counts: list[list[int]] = [] + for i, hierarchy in enumerate(self._hierarchies): + max_lv = self._max_levels[i] + counts: list[int] = [] + for lv in range(max_lv + 1): + # Count distinct values at this level by iterating over level-0 values + generalized: set[Any] = set() + for raw_val in anonymity_checker.dataset[ + anonymity_checker.dataset.columns[self._quasi_columns[i]] + ].unique(): + generalized.add(hierarchy.encode(raw_val, lv)) + counts.append(len(generalized)) + self._distinct_counts.append(counts) + + # Build the full lattice: level → set[LatticeNode] + levels_range: list[list[int]] = [list(range(self._max_levels[i] + 1)) for i in range(n_qi)] + all_combos: list[list[int]] = _calculate_product(levels_range) + + self._lattice: dict[int, list[LatticeNode]] = {} + self._node_map: dict[int, LatticeNode] = {} # hash → node + self._lattice_top_level: int = 0 + + for values in all_combos: + node = LatticeNode(values) + lv = node.sum() + self._lattice.setdefault(lv, []).append(node) + self._node_map[hash(node)] = node + if lv > self._lattice_top_level: + self._lattice_top_level = lv + + # Sort each level according to the Flash traversal order (ascending c) + for lv in self._lattice: + self._lattice[lv].sort(key=lambda n: self._priority(n)) + + self._global_optimum: LatticeNode | None = None + + # ------------------------------------------------------------------ + # Public interface + # ------------------------------------------------------------------ + + def explore(self) -> None: + """Run the Flash algorithm over the lattice.""" + # Min-heap: (priority-tuple, node). Python heapq is a min-heap. + heap: list[tuple[tuple[int, float, float], LatticeNode]] = [] + + all_levels = sorted(self._lattice.keys()) + + for lv in all_levels: + for node in self._lattice[lv]: + if not node.tagged: + path = self._find_path(node) + self._check_path(path, heap) + + # Drain the heap: process successors of non-anonymous boundary nodes + while heap: + _, boundary_node = heapq.heappop(heap) + for successor in self._successors_up(boundary_node): + if not successor.tagged: + path = self._find_path(successor) + self._check_path(path, heap) + + def optimal_node(self) -> LatticeNode | None: + """Return the globally optimal anonymous node, or *None*.""" + return self._global_optimum + + # ------------------------------------------------------------------ + # Core Flash sub-procedures (Algorithms 1–3 in the paper) + # ------------------------------------------------------------------ + + def _find_path(self, start: LatticeNode) -> list[LatticeNode]: + """Build a path of untagged nodes from *start* towards the top. + + Corresponds to FINDPATH in Algorithm 2. At each step we greedily + follow the first untagged successor (according to the Flash order). + """ + path: list[LatticeNode] = [] + node = start + while True: + path.append(node) + # Find the first (lowest-priority) untagged successor + next_node: LatticeNode | None = None + for candidate in self._successors_up(node): + if not candidate.tagged: + next_node = candidate + break + if next_node is None or next_node is node: + break + node = next_node + return path + + def _check_path( + self, + path: list[LatticeNode], + heap: list[tuple[tuple[int, float, float], LatticeNode]], + ) -> None: + """Binary-search a path for the optimal anonymous boundary node. + + Corresponds to CHECKPATH in Algorithm 3. + """ + low, high = 0, len(path) - 1 + local_optimum: LatticeNode | None = None + + while low <= high: + mid = (low + high) // 2 + node = path[mid] + + if self._check_and_tag(node): + # Anonymous: candidate optimum, search lower half + local_optimum = node + high = mid - 1 + else: + # Non-anonymous: push to heap for future path-building, search upper half + heapq.heappush(heap, (self._priority(node), node)) + low = mid + 1 + + self._store(local_optimum) + + def _check_and_tag(self, node: LatticeNode) -> bool: + """Evaluate *node* and propagate predictive tags. Returns True if anonymous.""" + is_anonymous = self._check_anonymity(node) + node.is_anonymous = is_anonymous + node.tagged = True + + # Predictive tagging: propagate towards generalizations (if anonymous) + # or towards specializations (if not anonymous). + self._tag_upward(node) if is_anonymous else self._tag_downward(node) + return is_anonymous + + def _store(self, local_optimum: LatticeNode | None) -> None: + """Update the global optimum if *local_optimum* improves it.""" + if local_optimum is None: + return + if self._global_optimum is None: + self._global_optimum = local_optimum + return + # Prefer lower total generalization level first, then lower info loss + if local_optimum.sum() < self._global_optimum.sum(): + self._global_optimum = local_optimum + elif local_optimum.sum() == self._global_optimum.sum(): + lo_loss = local_optimum.information_loss or math.inf + go_loss = self._global_optimum.information_loss or math.inf + if lo_loss < go_loss: + self._global_optimum = local_optimum + + # ------------------------------------------------------------------ + # Lattice helpers + # ------------------------------------------------------------------ + + def _check_anonymity(self, node: LatticeNode) -> bool: + """Compute suppression rate via the checker and record information loss.""" + suppression_rate = self._checker.calculate_suppression_rate(node) + node.suppression_rate = suppression_rate + return suppression_rate <= self._suppression + + def _tag_upward(self, node: LatticeNode) -> None: + """Tag all generalizations (successors) of an anonymous node as anonymous.""" + for successor in self._successors_up(node): + if not successor.tagged: + successor.is_anonymous = True + successor.tagged = True + self._tag_upward(successor) + + def _tag_downward(self, node: LatticeNode) -> None: + """Tag all specializations (predecessors) of a non-anonymous node as non-anonymous.""" + for predecessor in self._successors_down(node): + if not predecessor.tagged: + predecessor.is_anonymous = False + predecessor.tagged = True + self._tag_downward(predecessor) + + def _successors_up(self, node: LatticeNode) -> list[LatticeNode]: + """Return direct generalizations of *node* (one level up), sorted by priority.""" + result: list[LatticeNode] = [] + for i in range(len(node.values)): + if node.values[i] < self._max_levels[i]: + new_values = list(node.values) + new_values[i] += 1 + candidate = LatticeNode(new_values) + existing = self._node_map.get(hash(candidate)) + if existing is not None: + result.append(existing) + result.sort(key=lambda n: self._priority(n)) + return result + + def _successors_down(self, node: LatticeNode) -> list[LatticeNode]: + """Return direct specializations of *node* (one level down), sorted by priority.""" + result: list[LatticeNode] = [] + for i in range(len(node.values)): + if node.values[i] > 0: + new_values = list(node.values) + new_values[i] -= 1 + candidate = LatticeNode(new_values) + existing = self._node_map.get(hash(candidate)) + if existing is not None: + result.append(existing) + result.sort(key=lambda n: self._priority(n)) + return result + + def _priority(self, node: LatticeNode) -> tuple[int, float, float]: + """Return the (c1, c2, c3) ordering tuple for *node*.""" + return _node_priority(node, self._max_levels, self._distinct_counts) + + +# --------------------------------------------------------------------------- +# Top-level algorithm +# --------------------------------------------------------------------------- + + +class Flash(AnonymizationAlgorithm): + """Flash: Efficient, Stable and Optimal K-Anonymity. + + Finds the generalization with minimal information loss that satisfies all + privacy constraints within the configured suppression budget, using the + Flash lattice-traversal strategy described in: + + Kohlmayer et al. (2012). Flash: Efficient, Stable and Optimal + K-Anonymity. IEEE SocialCom-PASSAT, pp. 708–714. + + Args: + options: Algorithm configuration. + """ + + def __init__(self, options: FlashOptions) -> None: + self._options = options + + def anonymize( + self, dataset: DataFrame, column_information: list[ColumnInformation] + ) -> tuple[DataFrame, AnonymizationReport]: + """Anonymize the dataset using Flash. + + Args: + dataset: The input DataFrame. Quasi-identifier columns will be + generalized according to the optimal lattice node found. + column_information: Per-column metadata. Length must equal the + number of columns in ``dataset``. + + Returns: + A tuple of ``(anonymized_dataset, report)``. + + Raises: + ValueError: If ``column_information`` length does not match the + number of dataset columns. + RuntimeError: If no suitable generalization can be found. + """ + if len(column_information) != len(dataset.columns): + raise ValueError( + f"Dataset and column information are inconsistent in shape " + f"{len(dataset.columns)} vs {len(column_information)}" + ) + + if dataset is None or len(dataset) == 0: + return (dataset, AnonymizationReport(False)) + + if not any(c_i.column_type == ColumnType.QUASI for c_i in column_information): + return (dataset, AnonymizationReport(True, 0.0, [])) + + checker = AnonymityChecker( + dataset, + column_information, + self._options.privacy_constraints, + self._options.information_loss, + ) + lattice = FlashLattice(checker, column_information, self._options.suppression) + lattice.explore() + + best_node = lattice.optimal_node() + + if best_node is None or not best_node.is_anonymous: + raise RuntimeError("Flash: unable to find a suitable generalization") + + if best_node.suppression_rate and best_node.suppression_rate > 0.0: + anonymized = self._anonymize_with_suppression(dataset, column_information, best_node, checker) + else: + anonymized = _generalized_dataset(dataset.copy(), column_information, best_node.values) + + return ( + anonymized, + AnonymizationReport( + True, + best_node.suppression_rate or 0.0, + best_node.values, + ), + ) + + def _anonymize_with_suppression( + self, + dataset: DataFrame, + column_information: list[ColumnInformation], + node: LatticeNode, + checker: AnonymityChecker, + ) -> DataFrame: + """Apply generalization then drop equivalence classes that still violate constraints.""" + result = _generalized_dataset(dataset.copy(), column_information, node.values) + partitions = _partition_dataset(result, column_information) + for _, partition in partitions: + if not checker._check_constraints(partition): + result = result.drop(index=list(partition.index)) + return result diff --git a/tests/anonymization/test_flash.py b/tests/anonymization/test_flash.py new file mode 100644 index 0000000..8838eb6 --- /dev/null +++ b/tests/anonymization/test_flash.py @@ -0,0 +1,504 @@ +"""Tests for the Flash anonymization algorithm. + +Covers: +- FlashOptions dataclass defaults +- _node_priority ordering criteria (c1, c2, c3) +- FlashLattice construction, successor helpers, tagging propagation +- FlashLattice.explore() end-to-end +- Flash.anonymize() happy paths (no suppression, with suppression, l-diversity) +- Flash.anonymize() edge cases (empty dataset, no quasi columns, shape mismatch, + no suitable generalization) +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from risk_assessment.anonymization import DistinctLDiversity, KAnonymity +from risk_assessment.anonymization.flash import ( + Flash, + FlashLattice, + FlashOptions, + _node_priority, +) +from risk_assessment.anonymization.optimal_lattice_anonymization import ( + AnonymityChecker, + LatticeNode, +) +from risk_assessment.metrics.informationloss import ( + ColumnClass, + ColumnInformation, + ColumnType, + categorical_precision, +) +from risk_assessment.utility.hierarchy import MaterializedHierarchy +from risk_assessment.utility.hierarchy.datatypes import DummyHierarchy + +# --------------------------------------------------------------------------- +# Shared helpers / fixtures +# --------------------------------------------------------------------------- + + +def _date_hierarchy(): + return MaterializedHierarchy( + [ + ["01/01/2008", "Jan_2008", "2008"], + ["02/01/2008", "Jan_2008", "2008"], + ["03/01/2008", "Jan_2008", "2008"], + ] + ) + + +def _gender_hierarchy(): + return MaterializedHierarchy( + [ + ["M", "Person"], + ["F", "Person"], + ] + ) + + +def _age_hierarchy(): + return MaterializedHierarchy( + [ + ["13", "10-14", "10-19", "0-49", "0-99"], + ["18", "15-19", "10-19", "0-49", "0-99"], + ["19", "15-19", "10-19", "0-49", "0-99"], + ["21", "20-24", "20-29", "0-49", "0-99"], + ["22", "20-24", "20-29", "0-49", "0-99"], + ["23", "20-24", "20-29", "0-49", "0-99"], + ] + ) + + +def _three_col_info(): + return [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_date_hierarchy()), + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_age_hierarchy()), + ] + + +def _sample_dataset(): + rows = [ + "01/01/2008,M,18", + "01/01/2008,M,18", + "01/01/2008,M,18", + "01/01/2008,M,13", + "01/01/2008,M,19", + "02/01/2008,F,18", + "02/01/2008,F,22", + "02/01/2008,F,23", + "02/01/2008,F,21", + "01/01/2008,M,22", + ] + df = pd.DataFrame([r.split(",") for r in rows]) + df.rename(columns={i: f"col_{i}" for i in range(3)}, inplace=True) + return df + + +def _make_single_qi_lattice(k=2): + """FlashLattice with one categorical quasi-identifier (gender, 2 levels).""" + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ] + dataset = pd.DataFrame({"sex": ["M", "F", "M", "F"]}) + checker = AnonymityChecker(dataset, col_info, [KAnonymity(k)], categorical_precision) + return FlashLattice(checker, col_info, 0.0) + + +def _make_three_qi_lattice(): + """FlashLattice with three categorical quasi-identifiers.""" + col_info = _three_col_info() + dataset = pd.DataFrame( + [["01/01/2008", "M", "18"], ["02/01/2008", "F", "22"]], + columns=["date", "sex", "age"], + ) + checker = AnonymityChecker(dataset, col_info, [KAnonymity(2)], categorical_precision) + return FlashLattice(checker, col_info, 0.0) + + +# --------------------------------------------------------------------------- +# FlashOptions +# --------------------------------------------------------------------------- + + +def test_flash_options_default_suppression(): + opts = FlashOptions(privacy_constraints=[KAnonymity(2)]) + assert opts.suppression == 0.0 + + +def test_flash_options_default_information_loss(): + opts = FlashOptions(privacy_constraints=[KAnonymity(2)]) + assert opts.information_loss is categorical_precision + + +def test_flash_options_custom_suppression(): + opts = FlashOptions(privacy_constraints=[KAnonymity(2)], suppression=10.0) + assert opts.suppression == 10.0 + + +# --------------------------------------------------------------------------- +# _node_priority +# --------------------------------------------------------------------------- + +# Two quasi-identifiers: max_levels=[1, 3], distinct_counts=[[2,1],[4,2,1,1]] +_MAX_LEVELS = [1, 3] +_DISTINCT_COUNTS = [[2, 1], [4, 2, 1, 1]] + + +def test_node_priority_c1_equals_sum_of_levels(): + node = LatticeNode([1, 2]) + c1, _, _ = _node_priority(node, _MAX_LEVELS, _DISTINCT_COUNTS) + assert c1 == 3 + + +def test_node_priority_bottom_node_all_zeros(): + node = LatticeNode([0, 0]) + c1, c2, c3 = _node_priority(node, _MAX_LEVELS, _DISTINCT_COUNTS) + assert c1 == 0 + assert c2 == 0.0 + assert c3 == 0.0 + + +def test_node_priority_more_general_has_higher_c1(): + low = LatticeNode([0, 1]) + high = LatticeNode([1, 2]) + assert _node_priority(low, _MAX_LEVELS, _DISTINCT_COUNTS) < _node_priority(high, _MAX_LEVELS, _DISTINCT_COUNTS) + + +def test_node_priority_empty_node_returns_zeros(): + assert _node_priority(LatticeNode([]), [], []) == (0, 0.0, 0.0) + + +def test_node_priority_same_c1_differentiates_by_c2(): + # [1,0]: c2 = (1/1 + 0/3)/2 = 0.5 + # [0,1]: c2 = (0/1 + 1/3)/2 ≈ 0.167 + n1 = LatticeNode([1, 0]) + n2 = LatticeNode([0, 1]) + p1 = _node_priority(n1, _MAX_LEVELS, _DISTINCT_COUNTS) + p2 = _node_priority(n2, _MAX_LEVELS, _DISTINCT_COUNTS) + assert p1[0] == p2[0] # same c1 + assert p1[1] > p2[1] # n1 has higher c2, comes later in traversal + + +# --------------------------------------------------------------------------- +# FlashLattice — successor helpers +# --------------------------------------------------------------------------- + + +def test_successors_up_from_bottom_yields_one_per_qi(): + lattice = _make_three_qi_lattice() + bottom = lattice._node_map[hash(LatticeNode([0, 0, 0]))] + ups = lattice._successors_up(bottom) + assert len(ups) == 3 + for n in ups: + assert n.sum() == 1 + + +def test_successors_up_from_top_is_empty(): + lattice = _make_three_qi_lattice() + top_values = list(lattice._max_levels) + top = lattice._node_map[hash(LatticeNode(top_values))] + assert lattice._successors_up(top) == [] + + +def test_successors_down_from_above_bottom(): + lattice = _make_three_qi_lattice() + node = lattice._node_map[hash(LatticeNode([1, 0, 0]))] + downs = lattice._successors_down(node) + assert len(downs) == 1 + assert downs[0].values == [0, 0, 0] + + +def test_successors_down_from_bottom_is_empty(): + lattice = _make_three_qi_lattice() + bottom = lattice._node_map[hash(LatticeNode([0, 0, 0]))] + assert lattice._successors_down(bottom) == [] + + +def test_successors_are_sorted_by_priority(): + lattice = _make_three_qi_lattice() + bottom = lattice._node_map[hash(LatticeNode([0, 0, 0]))] + ups = lattice._successors_up(bottom) + priorities = [lattice._priority(n) for n in ups] + assert priorities == sorted(priorities) + + +# --------------------------------------------------------------------------- +# FlashLattice — tagging propagation +# --------------------------------------------------------------------------- + + +def test_tag_upward_marks_all_generalizations_anonymous(): + lattice = _make_single_qi_lattice() + bottom = lattice._node_map[hash(LatticeNode([0]))] + bottom.is_anonymous = True + bottom.tagged = True + lattice._tag_upward(bottom) + top = lattice._node_map[hash(LatticeNode([1]))] + assert top.tagged + assert top.is_anonymous is True + + +def test_tag_downward_marks_all_specializations_not_anonymous(): + lattice = _make_single_qi_lattice() + top = lattice._node_map[hash(LatticeNode([1]))] + top.is_anonymous = False + top.tagged = True + lattice._tag_downward(top) + bottom = lattice._node_map[hash(LatticeNode([0]))] + assert bottom.tagged + assert bottom.is_anonymous is False + + +def test_tag_upward_does_not_retag_already_tagged_nodes(): + lattice = _make_single_qi_lattice() + top = lattice._node_map[hash(LatticeNode([1]))] + top.tagged = True + top.is_anonymous = False # pre-tagged as non-anonymous + + bottom = lattice._node_map[hash(LatticeNode([0]))] + bottom.is_anonymous = True + bottom.tagged = True + lattice._tag_upward(bottom) + + # top was already tagged — its is_anonymous must not be overwritten + assert top.is_anonymous is False + + +# --------------------------------------------------------------------------- +# FlashLattice.explore() +# --------------------------------------------------------------------------- + + +def test_explore_finds_optimal_node(): + lattice = _make_single_qi_lattice(k=2) + lattice.explore() + optimal = lattice.optimal_node() + assert optimal is not None + assert optimal.is_anonymous is True + + +def test_explore_tags_all_nodes(): + lattice = _make_single_qi_lattice(k=2) + lattice.explore() + for nodes in lattice._lattice.values(): + for node in nodes: + assert node.tagged, f"Node {node} was not tagged after explore()" + + +def test_explore_impossible_k_yields_no_optimal_node(): + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ] + dataset = pd.DataFrame({"sex": ["M", "F"]}) + checker = AnonymityChecker(dataset, col_info, [KAnonymity(100)], categorical_precision) + lattice = FlashLattice(checker, col_info, 0.0) + lattice.explore() + assert lattice.optimal_node() is None + + +# --------------------------------------------------------------------------- +# Flash.anonymize() — happy paths +# --------------------------------------------------------------------------- + + +def test_flash_k2_no_suppression(): + dataset = _sample_dataset() + col_info = _three_col_info() + flash = Flash(FlashOptions([KAnonymity(2)], suppression=0.0)) + anonymized, report = flash.anonymize(dataset, col_info) + + assert report.anonymized + assert report.suppression_rate == 0.0 + assert report.generalization_levels is not None + + quasi_cols = [dataset.columns[i] for i, c in enumerate(col_info) if c.column_type == ColumnType.QUASI] + for size in anonymized.groupby(quasi_cols).size(): + assert size >= 2 + + +def test_flash_k3_with_suppression(): + dataset = _sample_dataset() + col_info = _three_col_info() + flash = Flash(FlashOptions([KAnonymity(3)], suppression=20.0)) + anonymized, report = flash.anonymize(dataset, col_info) + + assert report.anonymized + assert report.suppression_rate <= 20.0 + assert len(anonymized) >= 0.8 * len(dataset) + + quasi_cols = [dataset.columns[i] for i, c in enumerate(col_info) if c.column_type == ColumnType.QUASI] + for size in anonymized.groupby(quasi_cols).size(): + assert size >= 3 + + +def test_flash_named_columns(): + col_info = _three_col_info() + dataset = pd.DataFrame( + { + "date": ["01/01/2008"] * 5 + ["02/01/2008"] * 5, + "sex": ["M"] * 5 + ["F"] * 5, + "age": ["18", "18", "18", "13", "19", "18", "22", "23", "21", "22"], + } + ) + flash = Flash(FlashOptions([KAnonymity(3)], suppression=20.0)) + anonymized, report = flash.anonymize(dataset, col_info) + assert report.anonymized + assert len(anonymized) >= 0.8 * len(dataset) + + +def test_flash_k_anonymity_and_l_diversity(): + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_date_hierarchy()), + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_age_hierarchy()), + ColumnInformation(ColumnType.SENSITIVE), + ] + rows = [ + "01/01/2008,M,18,Cancer", + "01/01/2008,M,18,Cancer", + "01/01/2008,M,18,HIV", + "01/01/2008,M,13,HIV", + "01/01/2008,M,19,HIV", + "02/01/2008,F,18,Pneumonia", + "02/01/2008,F,22,Pneumonia", + "02/01/2008,F,23,Pneumonia", + "02/01/2008,F,21,Flu", + "01/01/2008,M,22,Flu", + ] + df = pd.DataFrame([r.split(",") for r in rows]) + df.rename(columns={i: f"col_{i}" for i in range(4)}, inplace=True) + + flash = Flash(FlashOptions([KAnonymity(3), DistinctLDiversity(2)], suppression=20.0)) + anonymized, report = flash.anonymize(df, col_info) + + assert report.anonymized + assert report.suppression_rate <= 20.0 + assert len(anonymized) >= 0.8 * len(df) + + +def test_flash_single_quasi_column(): + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ColumnInformation(ColumnType.SENSITIVE), + ] + dataset = pd.DataFrame({"sex": ["M", "F", "M", "F"], "disease": ["A", "B", "A", "B"]}) + flash = Flash(FlashOptions([KAnonymity(2)], suppression=0.0)) + anonymized, report = flash.anonymize(dataset, col_info) + assert report.anonymized + assert len(anonymized) == len(dataset) + + +def test_flash_output_preserves_all_columns(): + dataset = _sample_dataset() + col_info = _three_col_info() + flash = Flash(FlashOptions([KAnonymity(2)])) + anonymized, _ = flash.anonymize(dataset, col_info) + assert list(anonymized.columns) == list(dataset.columns) + + +# --------------------------------------------------------------------------- +# Flash.anonymize() — edge cases +# --------------------------------------------------------------------------- + + +def test_flash_empty_dataset_returns_false_report(): + col_info = _three_col_info() + empty_df = pd.DataFrame(columns=["date", "sex", "age"]) + flash = Flash(FlashOptions([KAnonymity(2)])) + _, report = flash.anonymize(empty_df, col_info) + assert not report.anonymized + + +def test_flash_column_count_mismatch_raises_value_error(): + dataset = _sample_dataset() + col_info = [ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=DummyHierarchy())] + flash = Flash(FlashOptions([KAnonymity(2)])) + with pytest.raises(ValueError): + flash.anonymize(dataset, col_info) + + +def test_flash_no_quasi_columns_returns_original_unchanged(): + dataset = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}) + col_info = [ColumnInformation(ColumnType.SENSITIVE), ColumnInformation(ColumnType.SENSITIVE)] + flash = Flash(FlashOptions([KAnonymity(2)])) + anonymized, report = flash.anonymize(dataset, col_info) + assert report.anonymized + assert report.generalization_levels == [] + assert len(anonymized) == len(dataset) + + +def test_flash_raises_when_no_suitable_generalization(): + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ] + dataset = pd.DataFrame({"sex": ["M", "F", "M"]}) + flash = Flash(FlashOptions([KAnonymity(100)], suppression=0.0)) + with pytest.raises(RuntimeError): + flash.anonymize(dataset, col_info) + + +# --------------------------------------------------------------------------- +# FlashLattice — missing hierarchy raises ValueError +# --------------------------------------------------------------------------- + + +def test_flash_lattice_raises_on_missing_hierarchy(): + col_info = [ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL)] # no hierarchy + dataset = pd.DataFrame({"sex": ["M", "F"]}) + checker = AnonymityChecker(dataset, col_info, [KAnonymity(2)], categorical_precision) + with pytest.raises(ValueError, match="Missing hierarchy"): + FlashLattice(checker, col_info, 0.0) + + +# --------------------------------------------------------------------------- +# _store — global optimum replacement branches +# --------------------------------------------------------------------------- + + +def test_flash_lattice_store_replaces_with_lower_level(): + lattice = _make_single_qi_lattice(k=2) + # Set a high-level global optimum manually + lattice._global_optimum = LatticeNode([1], information_loss=0.5, is_anonymous=True) + # A lower-level candidate should replace it + better = LatticeNode([0], information_loss=0.5, is_anonymous=True) + lattice._store(better) + assert lattice._global_optimum is better + + +def test_flash_lattice_store_replaces_on_same_level_lower_loss(): + lattice = _make_single_qi_lattice(k=2) + lattice._global_optimum = LatticeNode([1], information_loss=0.8, is_anonymous=True) + same_level_better = LatticeNode([1], information_loss=0.2, is_anonymous=True) + lattice._store(same_level_better) + assert lattice._global_optimum is same_level_better + + +def test_flash_lattice_store_does_not_replace_on_same_level_higher_loss(): + lattice = _make_single_qi_lattice(k=2) + original = LatticeNode([1], information_loss=0.2, is_anonymous=True) + lattice._global_optimum = original + worse = LatticeNode([1], information_loss=0.9, is_anonymous=True) + lattice._store(worse) + assert lattice._global_optimum is original + + +# --------------------------------------------------------------------------- +# Flash.anonymize() with active suppression (drops violating partitions) +# --------------------------------------------------------------------------- + + +def test_flash_anonymize_with_suppression_drops_small_partitions(): + """Force the suppression branch in _anonymize_with_suppression.""" + col_info = [ + ColumnInformation(ColumnType.QUASI, ColumnClass.CATEGORICAL, hierarchy=_gender_hierarchy()), + ] + # 5 M, 1 F → k=3 forces suppression of the lone-F partition + dataset = pd.DataFrame({"sex": ["M", "M", "M", "M", "M", "F"]}) + flash = Flash(FlashOptions([KAnonymity(3)], suppression=30.0)) + anonymized, report = flash.anonymize(dataset, col_info) + assert report.anonymized + assert len(anonymized) < len(dataset) # the F row(s) were suppressed From 72e5c67cc4d6a7a81482fb7fa7b3c5f4246d2d3a Mon Sep 17 00:00:00 2001 From: Stefano Braghin <527806+stefano81@users.noreply.github.com> Date: Mon, 31 Aug 2026 11:11:08 +0100 Subject: [PATCH 2/3] fix: patch test and fix identifier Signed-off-by: Stefano Braghin <527806+stefano81@users.noreply.github.com> --- .../classification/identifiers/__init__.py | 2 ++ .../identifiers/test_language_based_dictionary.py | 11 +++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/risk_assessment/classification/identifiers/__init__.py b/src/risk_assessment/classification/identifiers/__init__.py index 00c6917..d67bf15 100644 --- a/src/risk_assessment/classification/identifiers/__init__.py +++ b/src/risk_assessment/classification/identifiers/__init__.py @@ -470,6 +470,8 @@ def _enrich_with_language(identifier: LanguageBasedDictionaryIdentifier, languag for new_term in language_terms: if not identifier.case_sensitive: new_terms.add(new_term.casefold()) + else: + new_terms.add(new_term) except Exception as e: logger.info(f"error querying sparql for {term}") logger.debug(str(e)) diff --git a/tests/classification/identifiers/test_language_based_dictionary.py b/tests/classification/identifiers/test_language_based_dictionary.py index 69f3f8c..725f254 100644 --- a/tests/classification/identifiers/test_language_based_dictionary.py +++ b/tests/classification/identifiers/test_language_based_dictionary.py @@ -1,9 +1,16 @@ +from unittest.mock import patch + +import risk_assessment.classification.identifiers as identifiers_module from risk_assessment.classification.identifiers import LanguageBasedDictionaryIdentifier def test_expansion(): identifier = LanguageBasedDictionaryIdentifier("FOOBAR", {"en": ["Gear"]}, False) + def fake_enrich(ident, language): + ident.add_language(language, ["Engrenage"]) + assert identifier.is_of_this_type("gear"), "gear" - assert identifier.is_of_this_type_with_language("gear", "en"), "gear in french" - assert identifier.is_of_this_type_with_language("Engrenage", "fr"), "gear in french" + assert identifier.is_of_this_type_with_language("gear", "en"), "gear in english" + with patch.object(identifiers_module, "_enrich_with_language", side_effect=fake_enrich): + assert identifier.is_of_this_type_with_language("Engrenage", "fr"), "gear in french" From 12f9c2b4e3d2a7e13f89f5dc87f9e57581f37a76 Mon Sep 17 00:00:00 2001 From: Stefano Braghin <527806+stefano81@users.noreply.github.com> Date: Mon, 31 Aug 2026 13:46:36 +0100 Subject: [PATCH 3/3] fix: make github-advanced-security happy Signed-off-by: Stefano Braghin <527806+stefano81@users.noreply.github.com> --- src/risk_assessment/anonymization/flash.py | 6 +++--- .../identifiers/test_language_based_dictionary.py | 3 +-- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/risk_assessment/anonymization/flash.py b/src/risk_assessment/anonymization/flash.py index 04a9760..fac5a18 100644 --- a/src/risk_assessment/anonymization/flash.py +++ b/src/risk_assessment/anonymization/flash.py @@ -197,7 +197,7 @@ def __init__( # Sort each level according to the Flash traversal order (ascending c) for lv in self._lattice: - self._lattice[lv].sort(key=lambda n: self._priority(n)) + self._lattice[lv].sort(key=self._priority) self._global_optimum: LatticeNode | None = None @@ -346,7 +346,7 @@ def _successors_up(self, node: LatticeNode) -> list[LatticeNode]: existing = self._node_map.get(hash(candidate)) if existing is not None: result.append(existing) - result.sort(key=lambda n: self._priority(n)) + result.sort(key=self._priority) return result def _successors_down(self, node: LatticeNode) -> list[LatticeNode]: @@ -360,7 +360,7 @@ def _successors_down(self, node: LatticeNode) -> list[LatticeNode]: existing = self._node_map.get(hash(candidate)) if existing is not None: result.append(existing) - result.sort(key=lambda n: self._priority(n)) + result.sort(key=self._priority) return result def _priority(self, node: LatticeNode) -> tuple[int, float, float]: diff --git a/tests/classification/identifiers/test_language_based_dictionary.py b/tests/classification/identifiers/test_language_based_dictionary.py index 725f254..5b93f82 100644 --- a/tests/classification/identifiers/test_language_based_dictionary.py +++ b/tests/classification/identifiers/test_language_based_dictionary.py @@ -1,11 +1,10 @@ from unittest.mock import patch import risk_assessment.classification.identifiers as identifiers_module -from risk_assessment.classification.identifiers import LanguageBasedDictionaryIdentifier def test_expansion(): - identifier = LanguageBasedDictionaryIdentifier("FOOBAR", {"en": ["Gear"]}, False) + identifier = identifiers_module.LanguageBasedDictionaryIdentifier("FOOBAR", {"en": ["Gear"]}, False) def fake_enrich(ident, language): ident.add_language(language, ["Engrenage"])