politics-tracking/backend/tests/test_parsers.py

from datetime import date
from decimal import Decimal

from politrack.enrich.embed import split_text
from politrack.enrich.linker import detect_context, tokens
from politrack.enrich.topics import keyword_pattern
from politrack.sources.parlament.common import parse_date, parse_vote_text
from politrack.sources.rechnungshof import (
    clean_name,
    decode,
    find_columns,
    parse_amount,
    parse_de_date,
    read_rows,
)
from politrack.sources.wikidata import classify_level
from politrack.text import fingerprint, normalize, strip_titles


def test_vote_text_simple():
    v = parse_vote_text("137. Sitzung: Gesetzesvorschlag in dritter Lesung angenommen Dafür: ÖVP, SPÖ, GRÜNE, "
                        "dagegen: FPÖ, NEOS")
    assert v["result"] == "accepted"
    assert v["for"] == ["ÖVP", "SPÖ", "GRÜNE"]
    assert v["against"] == ["FPÖ", "NEOS"]
    assert v["stage"] == "137. Sitzung: Gesetzesvorschlag in dritter Lesung"


def test_vote_text_rejected_motion():
    v = parse_vote_text("50. Sitzung: Entschließungsantrag abgelehnt Dafür: FPÖ, dagegen: ÖVP, SPÖ, NEOS, GRÜNE")
    assert v["result"] == "rejected"
    assert v["for"] == ["FPÖ"]


def test_vote_text_split_vote_is_not_attributed():
    v = parse_vote_text("12. Sitzung: angenommen Dafür: V, S, N (teilweise G), dagegen: F")
    assert v["split"] is True
    assert v["for"] == [] and v["against"] == []


def test_vote_text_empty_against():
    v = parse_vote_text("997. Sitzung: Antrag, keinen Einspruch zu erheben, angenommen Dafür: ÖVP, SPÖ, dagegen: -")
    assert v["against"] == []
    assert v["stage"].endswith("zu erheben")


def test_vote_text_without_result():
    assert parse_vote_text("Zuweisung an den Ausschuss") is None


def test_parse_date_formats():
    assert parse_date("16.12.2021") == date(2021, 12, 16)
    assert parse_date("2026-10-14T12:00:00") == date(2026, 10, 14)
    assert parse_date("") is None


def test_rechnungshof_header_after_title_row():
    rows = [["Zu veröffentlichen", "", ""],
            ["Partei", "Meldedatum_an_RH", "Spendeneingangsdatum", "Name_der_Spenderin_des_Spenders",
             "PLZ_der_Spenderin_des_Spenders", "Betrag", "Waehrung", "Empfaengerin_Empfaenger_der_Spende", "Anmerkungen"]]
    start, idx = find_columns(rows)
    assert start == 2
    assert idx["date"] == 2  # Spendeneingangsdatum wins over Meldedatum
    assert idx["donor"] == 3
    assert idx["recipient"] == 7
    assert 4 not in idx.values()  # postcode is never mapped


def test_rechnungshof_headerless_file_skips_address():
    text = "08.07.19;300.000,00;Dr. Hans Peter Haselsteiner;1220 Wien Donau-City-Straße 9;NEOS\n"
    idx, rows = read_rows(text)
    assert idx["date"] == 0 and idx["amount"] == 1 and idx["donor"] == 2 and idx["party"] == 4
    assert 3 not in idx.values()
    assert rows[0][2] == "Dr. Hans Peter Haselsteiner"


def test_rechnungshof_clean_name_drops_addresses():
    assert clean_name("2522 Oberwaltersdorf Hofeldstraße 4;Team Stronach für Österreich") == "Team Stronach für Österreich"
    assert clean_name("1220 Wien Donau-City-Straße 9") == ""
    assert clean_name("NEOS – Das Neue Österreich") == "NEOS – Das Neue Österreich"


def test_rechnungshof_decoding():
    assert decode("Österreich".encode("cp1252")) == "Österreich"
    assert decode("Österreich".encode("mac_roman")) == "Österreich"
    assert decode("Österreich".encode("utf-8")) == "Österreich"


def test_amount_and_date():
    assert parse_amount("1.971.000,00") == Decimal("1971000.00")
    assert parse_amount("790") == Decimal("790")
    assert parse_amount("") is None
    assert parse_de_date("26.08.24") == date(2024, 8, 26)


def test_text_helpers():
    assert normalize("  Österreichische   Volkspartei ") == "osterreichische volkspartei"
    assert strip_titles("Mag. Dr. Selma Yildirim, BA") == "Selma Yildirim"
    assert fingerprint("a", 1) == fingerprint("a", 1) != fingerprint("a", 2)


def test_linker_helpers():
    assert tokens("Karl Nehammer (ÖVP) im ZIB 2-Interview") == ["karl", "nehammer", "ovp", "im", "zib", "2-interview"]
    assert detect_context("Babler in der ZIB 2") == "tv"
    assert detect_context("Gastkommentar: Warum wir mehr brauchen") == "op_ed"
    assert detect_context("Budget beschlossen") is None


def test_keyword_pattern_short_words_are_whole_words():
    pattern = keyword_pattern(["EU", "Klima"])
    assert pattern.search("Die EU beschließt")
    assert not pattern.search("Euro-Rettung")
    assert pattern.search("Klimaschutzgesetz")


def test_classify_level():
    assert classify_level("Mitglied des Tiroler Landtages") == "state"
    assert classify_level("Bürgermeister von Graz") == "municipal"
    assert classify_level("Abgeordneter zum Nationalrat") == "federal"
    assert classify_level("Mitglied des Europäischen Parlaments") == "eu"


def test_split_text():
    chunks = split_text("\n".join(["x" * 500] * 6), size=1200)
    assert all(len(c) <= 1800 for c in chunks)
    assert sum(len(c.replace("\n", "")) for c in chunks) == 3000


def test_employer_without_address():
    from politrack.sources.parlament.persons import employer

    assert employer("Hypo Tirol Bank AG, Meraner Straße 8, 6020 Innsbruck") == ("Hypo Tirol Bank AG", None)
    # parliament's JSON glues name, address and register number together
    assert employer("25 PersonaldienstleistungsgesmbHFranz Josef Kai 95020 Salzburg FN306527a") == (
        "25 PersonaldienstleistungsgesmbH", "306527a")
    assert employer("Salzburger Parkgaragen Gesellschaft m.b.H") == ("Salzburger Parkgaragen Gesellschaft m.b.H", None)
    assert employer("Verein Musterhilfe Hauptstraße 12 1010 Wien")[0] == "Verein Musterhilfe"


def test_firmenbuch_number():
    from politrack.text import firmenbuch_number

    assert firmenbuch_number("FN 306527 a") == firmenbuch_number("FN306527A") == firmenbuch_number("306527a") == "306527a"
    assert firmenbuch_number("ATU12345678") is None


def test_parlament_calendar_titles_are_html_decoded():
    import json

    from politrack.core.runner import make_response, parse_offline
    from politrack.sources.parlament.news import ParlamentCalendar

    body = {"header": [{"label": "Datum"}, {"label": "Terminart"}, {"label": "Bezeichnung"}, {"label": "Gremium"},
                       {"label": "Link"}, {"label": "WENTRY_ID"}],
            "rows": [["2026-06-10T09:00:00", "Plenarsitzung<BR>NR", "84.&nbsp;Sitzung des Nationalrates", "Nationalrat",
                      "/termin/1", "1"]]}
    events = parse_offline(ParlamentCalendar(), make_response("https://www.parlament.gv.at/x", json.dumps(body).encode(), kind="list"),
                           params={"since_year": 2026})
    assert [e.title for e in events] == ["84. Sitzung des Nationalrates"]
    assert events[0].description == "Plenarsitzung NR"


def test_procurement_rejects_implausible_award_dates():
    from datetime import date

    from politrack.sources.procurement import plausible_date

    today = date(2026, 11, 13)
    assert plausible_date("2025-02-17", today) == date(2025, 2, 17)
    assert plausible_date("1900-12-30", today) is None      # spreadsheet placeholder
    assert plausible_date("2028-09-30", today) is None      # the contract's end, not its award
    assert plausible_date(None, today) is None


def test_procurement_placeholder_contractor_and_framework_buyer():
    from politrack.sources.procurement import buyer_name, org_ref

    assert org_ref({"name": "-"}, "company") is None
    assert org_ref({"name": " n/a "}, "company") is None
    assert org_ref({"name": "Swietelsky AG"}, "company").name == "Swietelsky AG"
    sentence = ("Auftraggeber sind die Republik Österreich (Bund), die Bundesbeschaffung GmbH sowie alle weiteren "
                "Auftraggeber gemäß der den Ausschreibungsunterlagen beiliegenden Liste")
    assert buyer_name(sentence) == ("Bundesbeschaffung GmbH", sentence)
    assert buyer_name("ASFINAG Alpenstraßen GmbH") == ("ASFINAG Alpenstraßen GmbH", None)


def test_source_freshness_alerts_after_two_missed_slots():
    from datetime import datetime, timezone

    from politrack.queries.admin import source_freshness

    class Conn:
        def __init__(self, rows):
            self.rows = rows

        def execute(self, *_):
            return self.rows

    now = datetime(2026, 11, 13, 12, 0, tzinfo=timezone.utc)
    daily = "15 3 * * *"
    rows = [{"adapter": "fresh", "at": datetime(2026, 11, 13, 3, 40, tzinfo=timezone.utc)},
            {"adapter": "one_missed", "at": datetime(2026, 11, 12, 3, 40, tzinfo=timezone.utc)},
            {"adapter": "two_missed", "at": datetime(2026, 11, 11, 3, 40, tzinfo=timezone.utc)}]
    result = {r["adapter"]: r["stale"] for r in source_freshness(
        Conn(rows), {"fresh": daily, "one_missed": daily, "two_missed": daily, "never": daily}, now)}
    assert result == {"fresh": False, "one_missed": False, "two_missed": True, "never": True}


def test_election_preferences_eu_layouts():
    from politrack.sources.elections import parse_preferences, split_name

    assert split_name("Babler, MSc, Andreas") == ("Babler", "Andreas")
    assert split_name("Brunner, Dr. Magnus") == ("Brunner", "Dr. Magnus")
    eu2019 = [("ÖVP", [(None, "Kandidat(in)", "BGLD", "Gesamt"), (1, "Karas, Dr. Othmar ", 3012, 103035)])]
    eu2024 = [("ÖVP", [("LP", "Bewerber(in)", "BGLD", "Gesamt", None, "Darstellung", None, None),
                       (None, None, None, None, None, "(LP)", "Bewerber(in)", "Gesamt"),
                       (1, "Lopatka, Dr. Reinhold", 2268, 58013, None, 3, "Bernhuber, Alexander", 44641)])]
    assert [(c["name"], c["votes"]) for c in parse_preferences("eu", eu2019)] == [("Dr. Othmar Karas", 103035)]
    assert [(c["name"], c["votes"]) for c in parse_preferences("eu", eu2024)] == [("Dr. Reinhold Lopatka", 58013)]