govoplan-core/tests/test_review_mechanics.py

from __future__ import annotations

import random
import unittest
from decimal import Decimal

from govoplan_core.core.modules import DocumentationTopic, localize_documentation_topics
from govoplan_core.core.tabular_sources import (
    TabularColumn,
    TabularCsvSource,
    TabularSourceUnavailableError,
    TabularSourceValidationError,
    csv_source_payload,
    csv_source_summary,
    verified_csv_source_text,
    csv_projection_matches,
    infer_tabular_schema,
    parse_tabular_csv,
    tabular_type_name,
)


class SharedReviewMechanicsTests(unittest.TestCase):
    def test_invalid_csv_unicode_fails_explicitly_without_echoing_content(self) -> None:
        source = TabularCsvSource(text="value\nprivate-\ud800\n")
        for action in (
            lambda: csv_source_payload(source),
            lambda: parse_tabular_csv(source.text),
        ):
            with (
                self.subTest(action=action),
                self.assertRaises(TabularSourceValidationError) as raised,
            ):
                action()
            self.assertNotIn("private", str(raised.exception))
        with self.assertRaises(TabularSourceUnavailableError):
            verified_csv_source_text({"text": source.text})

    def test_csv_huge_integer_has_explicit_validation_and_lossless_text_option(
        self,
    ) -> None:
        import sys

        maximum = sys.get_int_max_str_digits()
        if not maximum:
            self.skipTest("Interpreter integer conversion limit is disabled")
        text = "9" * (maximum + 1)
        with self.assertRaisesRegex(TabularSourceValidationError, "use text mode"):
            parse_tabular_csv("value\n" + text + "\n")
        self.assertEqual(
            ({"value": text},),
            parse_tabular_csv("value\n" + text + "\n", value_mode="text"),
        )

    def test_csv_projection_binding_distinguishes_boolean_integer_and_float(
        self,
    ) -> None:
        for expected in (True, 1, 1.0):
            for actual in (True, 1, 1.0):
                with self.subTest(
                    expected_type=type(expected), actual_type=type(actual)
                ):
                    self.assertEqual(
                        type(expected) is type(actual),
                        csv_projection_matches(
                            ({"value": expected},), ({"value": actual},)
                        ),
                    )
        self.assertFalse(csv_projection_matches(({"value": 1},), ({"other": 1},)))

    def test_translation_merge_preserves_owner_data_and_untranslated_identity(
        self,
    ) -> None:
        translations = {"de": {"summary": "vorhanden"}, "fr": {"title": "Français"}}
        first = DocumentationTopic(
            id="one", title="One", summary="First", translations=translations
        )
        unchanged = DocumentationTopic(id="two", title="Two", summary="Second")
        localized = localize_documentation_topics(
            iter((first, unchanged)),
            locale="de",
            translations={"one": {"title": "Eins"}},
        )
        self.assertEqual(
            {"title": "Eins", "summary": "vorhanden"}, localized[0].translations["de"]
        )
        self.assertEqual({"title": "Français"}, localized[0].translations["fr"])
        self.assertIs(unchanged, localized[1])
        self.assertEqual({"summary": "vorhanden"}, first.translations["de"])
        self.assertIsNot(translations["fr"], localized[0].translations["fr"])

    def test_schema_inference_matches_legacy_projection_for_sparse_mixed_rows(
        self,
    ) -> None:
        generator = random.Random(298)
        values = [None, True, False, 1, 1.5, Decimal("1.00"), "001", [], {}]
        rows = [
            {
                f"field_{column}": generator.choice(values)
                for column in generator.sample(range(80), 20)
            }
            for _ in range(100)
        ]
        names = list(dict.fromkeys(name for row in rows for name in row))
        expected = []
        for name in names:
            concrete = [row[name] for row in rows if row.get(name) is not None]
            types = {tabular_type_name(value) for value in concrete}
            kind = (
                "unknown"
                if not types
                else next(iter(types))
                if len(types) == 1
                else "mixed"
            )
            expected.append(
                TabularColumn(
                    name=name, data_type=kind, nullable=len(concrete) != len(rows)
                )
            )
        self.assertEqual(tuple(expected), infer_tabular_schema(rows))
        self.assertEqual(
            (TabularColumn("only_null", "unknown", True),),
            infer_tabular_schema([{"only_null": None}]),
        )
        self.assertEqual((), infer_tabular_schema([]))
        unknown = type("ẞ", (), {})()
        self.assertEqual("ß", tabular_type_name(unknown))
        self.assertEqual("ss", tabular_type_name(unknown, casefold_unknown=True))

    def test_csv_evidence_keeps_exact_text_but_summary_never_contains_content(
        self,
    ) -> None:
        source = TabularCsvSource(
            text='\ufeffvalue\r\n"  text  "\r\n', value_mode="text"
        )
        payload = csv_source_payload(source)
        self.assertEqual(source.text, verified_csv_source_text(payload))
        self.assertNotIn("text", csv_source_summary(payload))
        self.assertEqual(len(source.text.encode("utf-8")), payload["byte_count"])
        with self.assertRaises(TabularSourceUnavailableError):
            verified_csv_source_text(
                {**payload, "text": source.text.replace("text", "edited")}
            )