diff --git a/src/freshdata/enterprise/join_assist.py b/src/freshdata/enterprise/join_assist.py index cded83f..e16c755 100644 --- a/src/freshdata/enterprise/join_assist.py +++ b/src/freshdata/enterprise/join_assist.py @@ -14,10 +14,12 @@ from __future__ import annotations +import math from dataclasses import dataclass, field from itertools import product from typing import Any +import numpy as np import pandas as pd from ..render import html as H @@ -32,6 +34,39 @@ def _similarity(a: str, b: str) -> float: return round(0.5 * jaro_winkler(a, b) + 0.5 * levenshtein_similarity(a, b), 4) +def _is_missing(value: Any) -> bool: + """True for None / NaN / NaT / ``pd.NA`` / ``""`` — a key carrying no evidence.""" + if value is None: + return True + if isinstance(value, str): + return value == "" + try: + return bool(pd.isna(value)) + except (TypeError, ValueError): # array-like cell: not a missing scalar + return False + + +def _key_text(value: Any) -> str: + """Comparable text for a key value; integral floats render without ``.0``. + + A numeric key column holding a missing value is promoted to float64, so + ``101`` and ``101.0`` must compare equal (#273). Only finite, integral + floats below 2**53 (exactly representable) are rewritten. + """ + if isinstance(value, (float, np.floating)): + f = float(value) + if math.isfinite(f) and f.is_integer() and abs(f) < 2**53: + return str(int(f)) + return str(value) + + +def _field_similarity(a: Any, b: Any) -> float: + """Similarity of two key cells; a missing value on either side scores 0 (#272).""" + if _is_missing(a) or _is_missing(b): + return 0.0 + return _similarity(_key_text(a), _key_text(b)) + + @dataclass(frozen=True) class JoinCandidate: """One suggested (left, right) match with per-field evidence.""" @@ -161,9 +196,13 @@ def _html_sections(self) -> list[str]: return [cards, exact, tbl, dl] +def _key_values(series: pd.Series) -> set[str]: + return {_key_text(v) for v in series.tolist() if not _is_missing(v)} + + def _exact_key_overlap(left: pd.DataFrame, right: pd.DataFrame, col: str) -> float: - lv = set(left[col].dropna().astype(str)) - rv = set(right[col].dropna().astype(str)) + lv = _key_values(left[col]) + rv = _key_values(right[col]) if not lv or not rv: return 0.0 return len(lv & rv) / min(len(lv), len(rv)) @@ -203,44 +242,54 @@ def suggest_join_keys( raise ValueError("none of the `on` columns are present in both frames") blocking = [c for c in (exact_within or []) if c in left.columns and c in right.columns] - exact_keys = [ - {"column": c, "overlap": round(_exact_key_overlap(left, right, c), 4), - "recommended": _exact_key_overlap(left, right, c) >= 0.95} - for c in on - ] - - # Build blocks (exact match on blocking columns); one block if none given. - def block_key(row: pd.Series) -> str: - return "|".join(str(row[c]) for c in blocking) if blocking else "*" - - left_blocks: dict[str, list[Any]] = {} - for idx, row in left.iterrows(): - left_blocks.setdefault(block_key(row), []).append(idx) - right_blocks: dict[str, list[Any]] = {} - for idx, row in right.iterrows(): - right_blocks.setdefault(block_key(row), []).append(idx) - - candidates: list[JoinCandidate] = [] + exact_keys = [] + for c in on: + overlap = _exact_key_overlap(left, right, c) + exact_keys.append( + {"column": c, "overlap": round(overlap, 4), "recommended": overlap >= 0.95}) + + # Rows are addressed by *position* throughout: row labels need not be unique + # (e.g. after pd.concat), and a label lookup would return a Series (#231). + # The original labels are only used in the reported candidates. + def block_keys(frame: pd.DataFrame) -> list[str]: + """Blocking key per row (exact match on blocking columns); "*" if none.""" + if not blocking: + return ["*"] * len(frame) + cols = [frame[c].tolist() for c in blocking] + return ["|".join(_key_text(v) for v in vals) for vals in zip(*cols)] + + left_blocks: dict[str, list[int]] = {} + for pos, key in enumerate(block_keys(left)): + left_blocks.setdefault(key, []).append(pos) + right_blocks: dict[str, list[int]] = {} + for pos, key in enumerate(block_keys(right)): + right_blocks.setdefault(key, []).append(pos) + + left_vals = {c: left[c].tolist() for c in on} + right_vals = {c: right[c].tolist() for c in on} + + candidates: list[tuple[int, JoinCandidate]] = [] pairs = 0 truncated = False - # best-per-left tracking for ambiguity detection - per_left: dict[Any, list[JoinCandidate]] = {} + # best-per-left tracking for ambiguity detection (keyed by left position) + per_left: dict[int, list[JoinCandidate]] = {} - for bkey, l_idxs in left_blocks.items(): - r_idxs = right_blocks.get(bkey, []) - for li, ri in product(l_idxs, r_idxs): + for bkey, l_pos in left_blocks.items(): + r_pos = right_blocks.get(bkey, []) + for lp, rp in product(l_pos, r_pos): if pairs >= max_pairs: truncated = True break pairs += 1 fscores = { - c: _similarity(str(left.at[li, c]), str(right.at[ri, c])) for c in on + c: _field_similarity(left_vals[c][lp], right_vals[c][rp]) for c in on } score = round(sum(fscores.values()) / len(on), 4) if score >= review_threshold: - cand = JoinCandidate(li, ri, score, "review", fscores, bkey) - candidates.append(cand) - per_left.setdefault(li, []).append(cand) + cand = JoinCandidate( + left.index[lp], right.index[rp], score, "review", fscores, bkey) + candidates.append((lp, cand)) + per_left.setdefault(lp, []).append(cand) if truncated: break @@ -248,8 +297,8 @@ def block_key(row: pd.Series) -> str: resolved: list[JoinCandidate] = [] from dataclasses import replace - for cand in candidates: - siblings = [c for c in per_left[cand.left_index] if c.score >= threshold] + for lp, cand in candidates: + siblings = [c for c in per_left[lp] if c.score >= threshold] if cand.score >= threshold: if len([c for c in siblings if abs(c.score - max(s.score for s in siblings)) <= 0.05]) > 1: diff --git a/src/freshdata/fieldcheck.py b/src/freshdata/fieldcheck.py index e296d06..d859214 100644 --- a/src/freshdata/fieldcheck.py +++ b/src/freshdata/fieldcheck.py @@ -24,7 +24,7 @@ class to an action. The default policy is non-destructive: nothing is deleted, import re from collections.abc import Callable, Collection, Mapping, Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Any import pandas as pd @@ -350,6 +350,15 @@ def _date_bound(value: float | str | None) -> pd.Timestamp | None: return None if pd.isna(ts) else ts +def _as_utc(ts: pd.Timestamp) -> pd.Timestamp: + """``ts`` in UTC; a naive timestamp is taken to already be UTC. + + Lets tz-aware values be compared with naive bounds (and vice versa) instead + of raising ``TypeError: Cannot compare tz-naive and tz-aware`` (#233). + """ + return ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC") + + def _parse_numeric(s: str) -> float | None: if is_plain_number(s): return float(str(s).strip().replace(",", "")) @@ -442,12 +451,12 @@ def issue(classification: str, reason: str, rule: str, *, "date_parse", ) lo, hi = _date_bound(spec.min_value), _date_bound(spec.max_value) - if lo is not None and ts < lo: + if lo is not None and _as_utc(ts) < _as_utc(lo): return issue( "domain_mismatch", f"{col}={ts.date()} is before the configured minimum {lo.date()}", "min_value") - if hi is not None and ts > hi: + if hi is not None and _as_utc(ts) > _as_utc(hi): return issue( "domain_mismatch", f"{col}={ts.date()} is after the configured maximum {hi.date()}", @@ -594,13 +603,15 @@ def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index: with warnings.catch_warnings(): warnings.simplefilter("ignore") - parsed_dt = pd.to_datetime(strs, errors="coerce") + # utc=True: same naive-as-UTC semantics as the per-cell check, + # and mixed offsets parse instead of raising + parsed_dt = pd.to_datetime(strs, errors="coerce", utc=True) fine = parsed_dt.notna() lo_d, hi_d = _date_bound(spec.min_value), _date_bound(spec.max_value) if lo_d is not None: - fine &= parsed_dt >= lo_d + fine &= parsed_dt >= _as_utc(lo_d) if hi_d is not None: - fine &= parsed_dt <= hi_d + fine &= parsed_dt <= _as_utc(hi_d) except (ValueError, TypeError): # pragma: no cover - exotic payloads fine = pd.Series(False, index=series.index) return series.index[must_flag | (checkable & ~fine.fillna(False))] @@ -700,7 +711,43 @@ def _validate_column( clean_config: TextCleanConfig | None, sensitive: bool = False, ) -> None: - """Run every per-column check for one column, appending to ``report``.""" + """Run every per-column check for one column, appending to ``report``. + + The checks run on a positional (``reset_index(drop=True)``) copy, so a + duplicated row label (common after ``pd.concat``) can never turn a cell + lookup into a Series (#231). Positions are mapped back to the original + row labels in the issues and the normalization audit. + """ + labels = series.index + positional = FieldValidationReport(n_rows=report.n_rows) + _validate_column_by_position( + col, series.reset_index(drop=True), spec, policy, positional, + rare_threshold=rare_threshold, outlier_fence=outlier_fence, + clean_config=clean_config, sensitive=sensitive, + ) + report.issues.extend( + issue if issue.row is None else replace(issue, row=labels[issue.row]) + for issue in positional.issues + ) + report.normalized_cells.extend( + {**cell, "row": labels[cell["row"]]} for cell in positional.normalized_cells + ) + report.inferred_types.update(positional.inferred_types) + + +def _validate_column_by_position( + col: str, + series: pd.Series, + spec: FieldSpec | None, + policy: RemediationPolicy, + report: FieldValidationReport, + *, + rare_threshold: float, + outlier_fence: float, + clean_config: TextCleanConfig | None, + sensitive: bool = False, +) -> None: + """Body of :func:`_validate_column`; ``series`` must have a unique index.""" # --- text normalization (audited, never in-place) ------------------------- cleaned_col: dict = {} transforms_col: dict = {} diff --git a/src/freshdata/textclean.py b/src/freshdata/textclean.py index dccd600..1da46eb 100644 --- a/src/freshdata/textclean.py +++ b/src/freshdata/textclean.py @@ -331,19 +331,25 @@ def clean_text( series = df[col] report.values_seen += int(series.notna().sum()) # ponytail: per-cell python loop; vectorize per-op if profiling demands - cleaned_values = {} - for idx, val in series.items(): + # Changes are collected by *position* and written back with iloc: row + # labels need not be unique (e.g. after pd.concat), and a label-keyed + # write would copy one row's cleaned value over its namesakes (#231). + positions: list[int] = [] + cleaned_values: list[str] = [] + for pos, (idx, val) in enumerate(series.items()): if not isinstance(val, str): continue result = clean_text_value(val, cfg) if result.changed: - cleaned_values[idx] = result.cleaned + positions.append(pos) + cleaned_values.append(result.cleaned) report.changes.append({ "row": idx, "column": str(col), "original": val, "cleaned": result.cleaned, "transforms": list(result.transforms), }) - if cleaned_values: - out[col] = series.copy() - out.loc[list(cleaned_values), col] = pd.Series(cleaned_values) + if positions: + new_col = series.copy() + new_col.iloc[positions] = cleaned_values + out[col] = new_col return out, report diff --git a/tests/test_index_alignment_and_join_keys.py b/tests/test_index_alignment_and_join_keys.py new file mode 100644 index 0000000..33e7f44 --- /dev/null +++ b/tests/test_index_alignment_and_join_keys.py @@ -0,0 +1,331 @@ +"""Duplicate row labels, tz-aware date bounds and honest join-key scoring. + +Covers #231 (parts 2-4: clean_text, validate_fields, suggest_join_keys on a +non-unique row index), #233 (part 3: tz-aware vs naive date bounds in +validate_fields), #272 (missing join keys score 0) and #273 (int vs +NaN-promoted float key columns overlap). +""" + +from __future__ import annotations + +import warnings + +import numpy as np +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.enterprise.join_assist import _key_text + +# --------------------------------------------------------------------------- +# helpers + + +def _dup(df: pd.DataFrame) -> pd.DataFrame: + """``df`` with every row label duplicated, as pd.concat would leave it.""" + out = df.copy() + out.index = [i // 2 for i in range(len(df))] + assert not out.index.is_unique + return out + + +def _issue_key(issue) -> tuple: + return ( + issue.column, + issue.classification, + issue.rule, + repr(issue.original), + repr(issue.cleaned), + ) + + +def _assert_same_field_report(frame: pd.DataFrame, control, report) -> None: + """``report`` (on ``frame``) matches ``control`` (on a RangeIndex copy).""" + assert [_issue_key(i) for i in report.issues] == [_issue_key(i) for i in control.issues] + assert [i.row for i in report.issues] == [ + None if i.row is None else frame.index[i.row] for i in control.issues + ] + assert [{k: v for k, v in c.items() if k != "row"} for c in report.normalized_cells] == [ + {k: v for k, v in c.items() if k != "row"} for c in control.normalized_cells + ] + assert [c["row"] for c in report.normalized_cells] == [ + frame.index[c["row"]] for c in control.normalized_cells + ] + assert report.inferred_types == control.inferred_types + + +# --------------------------------------------------------------------------- +# #231 part 2 — clean_text + + +def test_clean_text_duplicate_labels_keep_each_rows_value() -> None: + df = pd.DataFrame({"t": [" alice", "bob "]}, index=[7, 7]) + out, rep = fd.clean_text(df) + assert out["t"].tolist() == ["alice", "bob"] + assert list(out.index) == [7, 7] + assert [c["row"] for c in rep.changes] == [7, 7] + assert [(c["original"], c["cleaned"]) for c in rep.changes] == [ + (" alice", "alice"), + ("bob ", "bob"), + ] + assert df["t"].tolist() == [" alice", "bob "] # input untouched + + +def test_clean_text_after_concat_matches_unique_index_control() -> None: + part = pd.DataFrame({"name": [" Ann", "Bo b", None, "ok"], "n": [1, 2, 3, 4]}) + df = pd.concat([part, part.assign(name=["x ", " y", "z", None])]) + out, rep = fd.clean_text(df) + ctrl_out, ctrl_rep = fd.clean_text(df.reset_index(drop=True)) + assert out["name"].tolist() == ctrl_out["name"].tolist() + assert list(out.index) == list(df.index) + assert len(rep.changes) == len(ctrl_rep.changes) > 0 + assert [c["row"] for c in rep.changes] == [df.index[c["row"]] for c in ctrl_rep.changes] + + +# --------------------------------------------------------------------------- +# #231 part 3 — validate_fields + + +def test_validate_fields_duplicate_labels_issue_repro() -> None: + df = pd.DataFrame({"x": ["ok", "bad id!", "ok2"]}, index=[0, 0, 1]) + rep = fd.validate_fields(df, {"x": "identifier"}) + flagged = [(i.row, i.original) for i in rep.issues] + assert flagged == [(0, "bad id!")] + _assert_same_field_report( + df, fd.validate_fields(df.reset_index(drop=True), {"x": "identifier"}), rep + ) + + +@pytest.mark.parametrize( + ("frame", "schema", "kwargs"), + [ + # suspect cells + text normalization audit + ( + pd.DataFrame({"code": [" A1", "bad id!", "B2 ", "C3", "no way", "D4"]}), + {"code": "identifier"}, + {}, + ), + # numeric outliers + ( + pd.DataFrame({"amt": [10, 11, 12, 10, 11, 12, 10, 11, 5000, "abc"]}), + {"amt": "numeric"}, + {}, + ), + # rare allowed categories + ( + pd.DataFrame({"c": ["a"] * 30 + ["b", "zzz"]}), + {"c": fd.FieldSpec(allowed_values=["a", "b"])}, + {"rare_threshold": 0.05}, + ), + # no spec: column-consensus contamination + (pd.DataFrame({"n": ["1", "2", "3", "4", "five", "6", "7", "8"]}), None, {}), + ], +) +def test_validate_fields_duplicate_labels_match_unique_control(frame, schema, kwargs) -> None: + df = _dup(pd.concat([frame, frame], ignore_index=True)) + control = fd.validate_fields(df.reset_index(drop=True), schema, **kwargs) + report = fd.validate_fields(df, schema, **kwargs) + assert control.issues, "fixture must produce issues" + _assert_same_field_report(df, control, report) + + +def test_validate_fields_policy_split_keeps_labels_on_string_index() -> None: + df = pd.DataFrame({"x": ["ok", "bad id!", "ok2"]}, index=["r1", "r2", "r2"]) + rep = fd.validate_fields(df, {"x": "identifier"}) + assert [i.row for i in rep.issues] == ["r2"] + + +# --------------------------------------------------------------------------- +# #233 part 3 — tz-aware vs naive date bounds + + +def test_validate_fields_offset_aware_value_with_naive_bound_issue_repro() -> None: + df = pd.DataFrame({"ts": ["2024-01-01T10:00:00+05:30", "2023-05-05"]}) + rep = fd.validate_fields( + df, {"ts": fd.FieldSpec(semantic_type="date", min_value="1900-01-01")} + ) + assert isinstance(rep, fd.FieldValidationReport) + assert rep.issues == [] + + +def test_date_bounds_compare_in_utc() -> None: + df = pd.DataFrame( + { + "ts": [ + "2024-01-01T02:00:00+05:30", # 2023-12-31T20:30Z -> below the minimum + "2024-01-01T10:00:00+05:30", # 2024-01-01T04:30Z -> fine + "2024-01-05", # naive, taken as UTC -> fine + "2024-12-31T23:00:00-05:00", # 2025-01-01T04:00Z -> above the maximum + ] + } + ) + spec = fd.FieldSpec( + semantic_type="date", min_value="2024-01-01", max_value="2024-12-31T23:59:59" + ) + rep = fd.validate_fields(df, {"ts": spec}) + assert sorted((i.row, i.rule) for i in rep.issues) == [(0, "min_value"), (3, "max_value")] + + +def test_aware_bound_with_naive_values_and_aware_column() -> None: + spec = fd.FieldSpec( + semantic_type="date", + min_value="2024-01-01T00:00:00+00:00", + max_value=pd.Timestamp("2024-06-30", tz="UTC"), + ) + naive = pd.DataFrame({"ts": ["2023-12-31", "2024-03-01", "2024-07-01"]}) + rep = fd.validate_fields(naive, {"ts": spec}) + assert sorted((i.row, i.rule) for i in rep.issues) == [(0, "min_value"), (2, "max_value")] + + aware = pd.DataFrame( + { + "ts": pd.to_datetime(["2023-12-31", "2024-03-01", "2024-07-01"]).tz_localize( + "Asia/Tokyo" + ) + } + ) + naive_spec = fd.FieldSpec(semantic_type="date", min_value="2024-01-01", max_value="2024-06-30") + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + rep = fd.validate_fields(aware, {"ts": naive_spec}) + assert sorted((i.row, i.rule) for i in rep.issues) == [(0, "min_value"), (2, "max_value")] + + +def test_mixed_offsets_with_bounds() -> None: + df = pd.DataFrame( + { + "ts": [ + "2024-03-10T01:30:00-05:00", + "2024-03-10T03:30:00-04:00", + "1850-01-01T00:00:00+01:00", + ] + } + ) + rep = fd.validate_fields( + df, {"ts": fd.FieldSpec(semantic_type="date", min_value="1900-01-01")} + ) + assert [(i.row, i.rule) for i in rep.issues] == [(2, "min_value")] + + +# --------------------------------------------------------------------------- +# #231 part 4 — suggest_join_keys on duplicate labels + + +def test_join_keys_duplicate_labels_issue_repro() -> None: + left = pd.DataFrame({"company": ["Acme Corp", "Zenith Ltd"]}, index=[0, 0]) + right = pd.DataFrame({"company": ["Acme Corp"]}) + rep = fd.suggest_join_keys(left, right, on=["company"]) + assert [(c.left_index, c.right_index, c.score, c.status) for c in rep.candidates] == [ + (0, 0, 1.0, "match") + ] + + +def test_join_keys_duplicate_labels_match_unique_control() -> None: + base_left = pd.DataFrame( + { + "company": ["Acme Corp", "Acme Corp", "Zenith Ltd", "Globex Inc"], + "country": ["US", "US", "UK", "US"], + } + ) + base_right = pd.DataFrame( + { + "company": ["Acme Corp", "Zenith Limited", "Globex Inc."], + "country": ["US", "UK", "US"], + } + ) + left = _dup(base_left) + right = base_right.copy() + right.index = [5, 5, 6] + kwargs = {"on": ["company"], "exact_within": ["country"], "review_threshold": 0.5} + control = fd.suggest_join_keys(base_left, base_right, **kwargs) + rep = fd.suggest_join_keys(left, right, **kwargs) + assert control.candidates + assert [ + ( + left.index[c.left_index], + right.index[c.right_index], + c.score, + c.status, + c.field_scores, + c.block, + ) + for c in control.candidates + ] == [ + (c.left_index, c.right_index, c.score, c.status, c.field_scores, c.block) + for c in rep.candidates + ] + assert rep.exact_keys == control.exact_keys + assert rep.pairs_compared == control.pairs_compared + + +# --------------------------------------------------------------------------- +# #272 — missing keys carry no evidence + + +def test_nan_keys_on_both_sides_issue_repro() -> None: + left = pd.DataFrame({"company": [np.nan, "Acme Corp"]}) + right = pd.DataFrame({"company": [np.nan, "Zenith Ltd"]}) + rep = fd.suggest_join_keys(left, right, on=["company"]) + assert all((c.left_index, c.right_index) != (0, 0) for c in rep.candidates) + assert rep.matches == [] + + +@pytest.mark.parametrize("missing", [None, np.nan, pd.NA, pd.NaT, ""]) +def test_missing_field_scores_zero(missing) -> None: + left = pd.DataFrame({"company": ["Acme Corp"], "city": [missing]}, dtype=object) + right = pd.DataFrame({"company": ["Acme Corp"], "city": [missing]}, dtype=object) + rep = fd.suggest_join_keys(left, right, on=["company", "city"], review_threshold=0.1) + [cand] = rep.candidates + assert cand.field_scores == {"company": 1.0, "city": 0.0} + assert cand.score == 0.5 + city = next(k for k in rep.exact_keys if k["column"] == "city") + assert city["overlap"] == 0.0 + + +def test_missing_on_one_side_scores_zero() -> None: + left = pd.DataFrame({"company": [None]}, dtype=object) + right = pd.DataFrame({"company": ["None"]}) + rep = fd.suggest_join_keys(left, right, on=["company"], review_threshold=0.0) + assert [c.field_scores for c in rep.candidates] == [{"company": 0.0}] + + +# --------------------------------------------------------------------------- +# #273 — int vs NaN-promoted float keys + + +def test_int_vs_float_key_columns_issue_repro() -> None: + left = pd.DataFrame({"customer_id": [101, 102, 103]}) + right = pd.DataFrame({"customer_id": [101, 102, None]}) + assert right["customer_id"].dtype == np.float64 + rep = fd.suggest_join_keys(left, right, on=["customer_id"]) + assert rep.exact_keys == [{"column": "customer_id", "overlap": 1.0, "recommended": True}] + assert sorted((c.left_index, c.right_index) for c in rep.matches) == [(0, 0), (1, 1)] + assert all(c.field_scores == {"customer_id": 1.0} for c in rep.matches) + + +def test_int_vs_float_blocking_columns_line_up() -> None: + left = pd.DataFrame({"region": [1, 2], "company": ["Acme Corp", "Zenith Ltd"]}) + right = pd.DataFrame({"region": [1.0, np.nan], "company": ["Acme Corp", "Zenith Ltd"]}) + rep = fd.suggest_join_keys(left, right, on=["company"], exact_within=["region"]) + assert [(c.left_index, c.right_index, c.block) for c in rep.matches] == [(0, 0, "1")] + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (101.0, "101"), + (np.float64(-7.0), "-7"), + (np.float32(3.0), "3"), + (0.0, "0"), + (1.5, "1.5"), + (float("inf"), "inf"), + (float("-inf"), "-inf"), + (float("nan"), "nan"), + (float(2**53), str(float(2**53))), + (1e300, str(1e300)), + (101, "101"), + (True, "True"), + ("101.0", "101.0"), + ], +) +def test_key_text(value, expected) -> None: + assert _key_text(value) == expected