simonandClaude Opus 5.5
583ccba31c
Initial import of Politrack, a self-hosted service that tracks Austrian politicians from federal to municipal level - roles, votes, money flows, statements, meetings, social media and legal proceedings - with every claim linked to its sources. - backend: FastAPI + Postgres/pgvector; adapter framework (fetch -> parse -> load -> enrich) with robots.txt/TDM-aware fetching, a raw-response archive for reparsing, and a claim/evidence model with corroboration - sources: Parliament, Wikidata, EU Parliament/Commission, Council of Ministers, Rechnungshof party reports and donations, procurement, RTR media transparency, political ad libraries, WKStA, news feeds, Wikipedia edits, Bluesky, Mastodon, public Telegram channels, YouTube and X (official API) - web: Next.js frontend in German and English with typed API client - mcp: MCP server for AI agents and RAG - docs: architecture, data model, sources, legal notes, Unraid deployment and contributor guides; Docker Compose setup for Unraid Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
130 lines
7.3 KiB
Python
130 lines
7.3 KiB
Python
"""Framework invariants (no DB): requests, redaction, registry, serialisation, helpers shared by adapters."""
|
||
|
||
from datetime import date
|
||
|
||
from politrack.core import registry
|
||
from politrack.core.adapter import Request, redact_url
|
||
from politrack.core.records import Claim, Evidence, Money, OrgRef, to_jsonable
|
||
from politrack.openapi_ts import to_typescript, ts_type
|
||
from politrack.sources.bundesregierung import ITEM, german_date, offices, period_start
|
||
from politrack.text import format_eur_de, normalize_org_name
|
||
|
||
|
||
def test_request_fingerprint_ignores_meta_and_headers():
|
||
a = Request(url="https://example.org/x", params={"a": 1}, meta={"person": 1})
|
||
b = Request(url="https://example.org/x", params={"a": 1}, meta={"person": 2}, headers={"Authorization": "x"})
|
||
assert a.fingerprint == b.fingerprint
|
||
assert a.fingerprint != Request(url="https://example.org/x", params={"a": 2}).fingerprint
|
||
|
||
|
||
def test_secrets_never_reach_the_archive():
|
||
req = Request(url="https://api.example.org/v3", params={"key": "SECRET", "q": "x"},
|
||
headers={"Authorization": "Bearer SECRET"})
|
||
assert "SECRET" not in str(req.describe())
|
||
assert "SECRET" not in redact_url("https://api.example.org/v3?q=x&access_token=SECRET")
|
||
|
||
|
||
def test_registry_names_are_unique_and_specs_consistent():
|
||
registry.load_builtin()
|
||
assert "parlament.persons" in registry.ADAPTERS and "enrich.linker" in registry.ENRICHERS
|
||
assert not set(registry.ADAPTERS) & set(registry.ENRICHERS)
|
||
for name, cls in registry.ADAPTERS.items():
|
||
assert cls.spec.slug in registry.SPECS, name
|
||
assert cls.queue in ("scrape", "social", "bulk"), name
|
||
assert name.split(".")[0], name
|
||
for name, cls in registry.ENRICHERS.items():
|
||
assert cls.queue == "enrich", name
|
||
|
||
|
||
def test_to_jsonable_is_stable():
|
||
claim = Claim(claim_type="donation", summary_de="x", org=OrgRef(slug="oevp"), occurred_at=date(2024, 1, 2),
|
||
evidence=[Evidence(url="https://example.org")], facet=Money(kind="donation", amount=1000))
|
||
data = to_jsonable(claim)
|
||
assert data["__type__"] == "Claim" and data["facet"]["__type__"] == "Money"
|
||
assert data["occurred_at"] == "2024-01-02"
|
||
|
||
|
||
def test_normalize_org_name():
|
||
assert normalize_org_name("HYPO TIROL BANK AG") == normalize_org_name("Hypo Tirol Bank Aktiengesellschaft")
|
||
assert normalize_org_name("Österreichische Post AG") == "osterreichische post"
|
||
assert normalize_org_name("Muster Bau Gesellschaft m.b.H.") == "muster bau"
|
||
|
||
|
||
def test_format_eur_de():
|
||
assert format_eur_de(1234567.8) == "1.234.567,80"
|
||
assert format_eur_de("n/a") == "n/a"
|
||
|
||
|
||
MINISTERRAT = (
|
||
"Beschlussprotokoll des 57. Ministerrates vom 17. Juni 2026 | "
|
||
"4.1. Gemeinsamer Bericht des Bundesministers für Inneres und der Bundesministerin für Justiz, "
|
||
"Zahl 2026-0.256.545, betreffend Tagung des Rates für Justiz und Inneres. | "
|
||
"Der Ministerrat beschließt im Sinne des Antrages. | "
|
||
"Gemeinsamer Bericht des Bundeskanzlers, des Bundesministers für Wohnen, Kunst, Kultur, Medien und Sport "
|
||
"und der Bundesministerin für europäische und internationale Angelegenheiten, Zahl 2026-0.513.56, betreffend "
|
||
"Ausbau der Verwaltung. | Der Ministerrat beschließt im Sinne des Antrages. | "
|
||
"Bericht des Bundeskanzlers, Zahl 2026-0.367.613, betreffend § 4 | Abs. | 3 des Gesetzes. | "
|
||
"Der Ministerrat beschließt im Sinne des Antrages. | "
|
||
"Bericht des Bundesministers für Innovation, Mobilität und Infrastruktur, Zahl 2026‑0.481.080, "
|
||
"betreffend ÖBB-Rahmenplan 2027 bis 2032. | Der Ministerrat beschließt im Sinne des Antrages."
|
||
)
|
||
|
||
|
||
def test_ministerrat_items_and_offices():
|
||
items = list(ITEM.finditer(MINISTERRAT))
|
||
assert [m.group("zahl")[:6] for m in items] == ["2026-0", "2026-0", "2026-0", "2026‑0"]
|
||
assert offices(items[0].group("offices")) == ["Bundesminister für Inneres", "Bundesministerin für Justiz"]
|
||
assert offices(items[1].group("offices")) == [
|
||
"Bundeskanzler", "Bundesminister für Wohnen, Kunst, Kultur, Medien und Sport",
|
||
"Bundesministerin für europäische und internationale Angelegenheiten"]
|
||
assert offices(items[3].group("offices")) == ["Bundesminister für Innovation, Mobilität und Infrastruktur"]
|
||
assert "Abs." in items[2].group("subject")
|
||
assert german_date("Beschlussprotokoll des 57. Ministerrates vom 17. Juni 2026") == date(2026, 6, 17)
|
||
assert german_date("am 3. Jänner 2025") == date(2025, 1, 3)
|
||
assert period_start("/x/ministerratsprotokolle-regierungsperiode-xxviii-2025.html") == 2025
|
||
assert period_start("/x/ministerratsprotokolle-der-xxvi-regierungsperiode-2017-2018.html") == 2018
|
||
assert period_start("/x/ministerratsprotokolle-xxv-regierungsperiode.html") == 0
|
||
|
||
|
||
def test_openapi_to_typescript():
|
||
assert ts_type({"anyOf": [{"type": "string"}, {"type": "null"}]}) == "string | null"
|
||
assert ts_type({"type": "array", "items": {"$ref": "#/components/schemas/Claim"}}) == "Claim[]"
|
||
assert ts_type({"enum": ["a", "b"]}) == "'a' | 'b'"
|
||
schema = {"components": {"schemas": {"X": {"type": "object", "required": ["a"],
|
||
"properties": {"a": {"type": "integer"}, "b": {"type": "string"}}}}},
|
||
"paths": {"/x": {"get": {"responses": {"200": {"content": {"application/json": {
|
||
"schema": {"$ref": "#/components/schemas/X"}}}}}}}}}
|
||
ts = to_typescript(schema)
|
||
assert "export type X = {\n a: number;\n b?: string;\n};" in ts
|
||
assert "'/x': X;" in ts
|
||
|
||
|
||
def test_office_key_is_gender_neutral_and_strict():
|
||
from politrack.core.resolve import office_key
|
||
|
||
assert office_key("Bundesministerin für Justiz") == office_key("Bundesminister für Justiz")
|
||
assert office_key("Bundeskanzler der Republik Österreich") == "bundeskanzler"
|
||
assert office_key("Bundesministerin für Europa, Integration und Familie im Bundeskanzleramt") == \
|
||
office_key("Bundesministerin für Europa, Integration und Familie")
|
||
assert office_key("Staatssekretärin im Bundesministerium für Finanzen") == \
|
||
"staatssekretar im bundesministerium fur finanzen"
|
||
assert office_key("Bundesminister") != office_key("Bundesminister für Finanzen")
|
||
|
||
|
||
def test_office_portfolios_for_colloquial_labels():
|
||
from politrack.core.resolve import office_key, office_kind, portfolios
|
||
|
||
official = office_key("Bundesministerin für europäische und internationale Angelegenheiten")
|
||
assert office_kind(official) == office_kind(office_key("Außenminister Österreichs")) == "minister"
|
||
assert portfolios(official) & portfolios(office_key("Außenminister Österreichs"))
|
||
culture = office_key("Federal Minister for Arts, Culture, the Civil Service and Sport")
|
||
assert portfolios(office_key("Bundesminister für Wohnen, Kunst, Kultur, Medien und Sport")) & portfolios(culture)
|
||
assert office_kind(office_key("Vizekanzler")) == "vizekanzler"
|
||
agriculture = office_key("Bundesminister für Land- und Forstwirtschaft, Klima- und Umweltschutz, Regionen und Wasserwirtschaft")
|
||
assert "wirtschaft" not in portfolios(agriculture)
|
||
from politrack.core.resolve import lead_portfolio
|
||
|
||
economy = office_key("Bundesminister für Wirtschaft, Energie und Tourismus")
|
||
climate = office_key("Federal Minister for Climate Action, Environment, Energy, Mobility, Innovation and Technology")
|
||
assert lead_portfolio(economy) == "wirtschaft" and lead_portfolio(economy) not in portfolios(climate)
|