"""Tests für Import, Datenbank und Export – ausschließlich mit erfundenen Testdaten.""" from datetime import datetime import openpyxl import pytest from provtool import export from provtool.db import UNZUGEORDNET, Datenbank, Haendler from provtool.importer import ImportFehler, lies_datei from provtool.money import apply_percent, fmt, parse_cent FREENET_KOPF = ("Zeitraum;Verguetungsart;Haendlernummer;Unterhaendler;Vertragsnummer;Aktiv;Deaktiv;Rufnummer;" "Kartennummer;IMEI_SN;Geraetename;Kundenvorname;Kundename;Tarif-Kurz;Tarif;Prozent;Betrag;Grund;" "externe Vertragsnummer;Mitarbeiternummer;Abrechnungsnummer;Netz-Kurz;Netz;Dienst-Kurz;Dienst;" "Aktionscode;Auftragserfassungsdatum;Detailausweisung;Teilzahlung Nr.;Event-Datum;FAQ;Bonusschreiben;" "Leistungsdatum;Couponcode") def freenet_zeile(art, unterh, aktiv, rufnr, name, tarif, betrag, netz="VF", aktion="", zeitraum="2026.10.01-2026.10.07"): f = [""] * 34 f[0], f[1], f[2], f[3], f[4], f[5] = zeitraum, art, "28000000", unterh, "123", aktiv f[7], f[12], f[14], f[16], f[22], f[24], f[25] = rufnr, name, tarif, betrag, netz, "Teilnehmerverhaeltnis", aktion return ";".join(f) @pytest.fixture def freenet_csv(tmp_path): zeilen = [ FREENET_KOPF, freenet_zeile("Grundvergütung Mobilfunk Credit freenet", "111", "05.10.2026", "1601111111", "Muster GmbH", "Allnet Flat 5GB", "50"), freenet_zeile("SIM-Karten", "111", "05.10.2026", "1601111111", "Muster GmbH", "Allnet Flat 5GB", "8"), freenet_zeile("Sonderbonus Aktion", "111", "07.10.2026", "1602222222", "Beispiel KG", "Business Classic", "170", "D1", "A123"), freenet_zeile("Grundvergütung Mobilfunk Credit freenet", "111", "06.10.2026", "1602222222", "Beispiel KG", "Business Classic", "12,50", "D1"), freenet_zeile("SIM-Karten", "999", "01.10.2026", "1609999999", "Eigener Kunde", "Flex", "-8"), ] p = tmp_path / "freenet.csv" p.write_bytes(("\r\n".join(zeilen) + "\r\n").encode("cp1252")) return p @pytest.fixture def vodafone_xlsx(tmp_path): wb = openpyxl.Workbook() ws = wb.active ws.title = "All" ws["A1"] = "Provisionsclearing Euronics e.G." ws["B3"], ws["C3"] = "Kundennummer:", 1 ws.append([]) ws.append([]) ws["A6"] = None kopf = ["Anschaltdatum", "Kartennummer", "IMEI", "Rufnummer", "Kunde", "Artikelbezeichnung", "Tarif", "TarifKey02", "Prämie in EUR", "Abrechnungslauf", "Vo-Nr", "Lieferantennr", "Text"] for c, h in enumerate(kopf, 1): ws.cell(6, c, h) daten = [ (datetime(2026, 9, 8), "3225601 ", "", "491726400000 ", "Testkunde ", "Freischaltungsvergütung Vodafone", "Aktivierung Smart L ", "Smart ", 345, "VF2609 ", "80211721 ", "705136", "SMART3L"), (datetime(2026, 9, 8), "3225601 ", "", "491726400000 ", "Testkunde ", "Freischaltungsvergütung Vodafone", "Basispreisrabatt", "Promotion", -40, "VF2609 ", "80211721 ", "705136", "SMART3L"), (datetime(2026, 9, 3), "", "", "182935058", "Kabelkunde", "Freischaltungsvergütung Vodafone Cable", "VF Cable/GigaZuhause 50", "VF Internet Cable", 120, "VF2609", "EU133301", "705136", "5BV:FPT"), ] for r, row in enumerate(daten, 7): for c, v in enumerate(row, 1): ws.cell(r, c, v) leer = wb.create_sheet("VF2609 Fil 0 Ref 1") leer["A5"], leer["B5"] = "Anschaltdatum", "Kartennummer" p = tmp_path / "vodafone.xlsx" wb.save(p) return p @pytest.fixture def db(tmp_path): d = Datenbank(tmp_path / "test.db") yield d d.close() # --------------------------------------------------------------------------- Geld @pytest.mark.parametrize("ein,cent", [("30", 3000), ("12,50", 1250), ("1.234,56", 123456), ("-8", -800), (12.5, 1250), (8, 800), ("30,00-", -3000), ("", 0), (None, 0), ("1,234.56", 123456), ("0,1", 10)]) def test_parse_cent(ein, cent): assert parse_cent(ein) == cent def test_prozent_und_format(): assert apply_percent(17000, 80) == 13600 assert apply_percent(1250, 33.33) == 417 assert apply_percent(-800, 50) == -400 assert fmt(123456) == "1.234,56 €" assert fmt(-5) == "-0,05 €" # --------------------------------------------------------------------------- Import def test_freenet_import(freenet_csv): erg = lies_datei(freenet_csv) assert erg.format == "freenet" assert len(erg.positionen) == 5 p = erg.positionen[0] assert p.provisionsart == "Grundvergütung Mobilfunk Credit freenet" assert p.kennung == "111" assert p.datum.isoformat() == "2026-10-05" assert p.betrag_cent == 5000 assert erg.positionen[3].betrag_cent == 1250 assert erg.positionen[2].aktionscode == "A123" def test_vodafone_import(vodafone_xlsx): erg = lies_datei(vodafone_xlsx) assert erg.format == "vodafone" assert len(erg.positionen) == 3 p = erg.positionen[0] assert p.kennung == "80211721" # Leerzeichen entfernt assert p.rufnummer == "491726400000" assert p.betrag_cent == 34500 assert p.zeitraum == "VF2609" assert erg.positionen[2].netz == "Cable" def test_unbekanntes_format(tmp_path): p = tmp_path / "x.csv" p.write_text("a;b;c\n1;2;3\n") with pytest.raises(ImportFehler): lies_datei(p) # --------------------------------------------------------------------------- Datenbank def test_zuordnung_regeln_und_export(db, freenet_csv, vodafone_xlsx, tmp_path): hid = db.haendler_speichern(Haendler(0, "Händler A", standard_prozent=100), [("freenet", "111"), ("vodafone", "80211721")]) aid = db.abrechnung_neu("Oktober 2026") db.importieren(aid, lies_datei(freenet_csv)) db.importieren(aid, lies_datei(vodafone_xlsx)) pos = db.positionen(aid) eigene = [p for p in pos if p["haendler_id"] == hid] assert len(eigene) == 6 assert sum(1 for p in pos if p["haendler_id"] == UNZUGEORDNET) == 2 # Regel: Sonderbonus nur 80 % db.regel_setzen(hid, "Sonderbonus Aktion", 80) db.neu_berechnen(aid) sb = next(p for p in db.positionen(aid) if p["provisionsart"] == "Sonderbonus Aktion") assert sb["betrag"] == 13600 and sb["prozent"] == 80 # manuelle Änderung bleibt bei Neuberechnung erhalten sim = next(p for p in db.positionen(aid) if p["provisionsart"] == "SIM-Karten" and p["haendler_id"] == hid) db.betrag_setzen(sim["id"], 500) db.neu_berechnen(aid) sim = next(p for p in db.positionen(aid) if p["id"] == sim["id"]) assert sim["betrag"] == 500 and sim["manuell"] == 1 db.neu_berechnen(aid, auch_manuelle=True) sim = next(p for p in db.positionen(aid) if p["id"] == sim["id"]) assert sim["betrag"] == 800 and sim["manuell"] == 0 # Export eigene = [p for p in db.positionen(aid) if p["haendler_id"] == hid] dateien = export.exportiere_haendler(tmp_path / "out", "Händler A", eigene, "Oktober 2026", pro_anbieter=True) assert [d.name for d in dateien] == ["Händler A - freenet - Oktober 2026.xlsx", "Händler A - Vodafone - Oktober 2026.xlsx"] # Standard: freenet UND Vodafone in einer Datei (gemeldet: Vodafone fehlte im Export) eine = export.exportiere_haendler(tmp_path / "out1", "Händler A", eigene, "Oktober 2026") assert [d.name for d in eine] == ["Händler A - Oktober 2026.xlsx"] arten = [r[1] for r in openpyxl.load_workbook(eine[0]).active.iter_rows(values_only=True) if r[1]] assert "Freischaltungsvergütung Vodafone" in arten and "SIM-Karten" in arten ws = openpyxl.load_workbook(dateien[0]).active zeilen = [[c.value for c in r] for r in ws.iter_rows(min_row=1, max_row=4)] # Sortierung Zeitraum → Provisionsart → Datum assert [z[1] for z in zeilen] == ["Grundvergütung Mobilfunk Credit freenet"] * 2 + ["SIM-Karten", "Sonderbonus Aktion"] assert zeilen[0][2] == 111 and zeilen[0][4] == 1601111111 assert zeilen[3][10] == 136.0 and zeilen[3][13] == "A123" assert ws["K6"].value == "=SUM(K1:K4)" def test_doppelte_kennung_verboten(db): db.haendler_speichern(Haendler(0, "A"), [("freenet", "1")]) with pytest.raises(ValueError): db.haendler_speichern(Haendler(0, "B"), [("freenet", "1")]) def test_datei_doppelt_erkannt_und_entfernbar(db, freenet_csv): aid = db.abrechnung_neu("X") erg = lies_datei(freenet_csv) db.importieren(aid, erg) assert db.datei_schon_importiert(erg.sha256) == ["X"] did = db.dateien(aid)[0]["id"] db.datei_entfernen(did) assert db.positionen(aid) == [] # --------------------------------------------------------------------------- Vodafone: GROSSBUCHSTABEN-Regel def test_grossbuchstaben_erkennung(): from provtool.db import ist_grossgeschrieben assert ist_grossgeschrieben("HUBERMEIER") assert ist_grossgeschrieben("ÖSTERLEHNER ") assert ist_grossgeschrieben("GROß") # ß gibt es nicht als Großbuchstabe assert not ist_grossgeschrieben("Musterfrau") assert not ist_grossgeschrieben("Max & Moritz Test") assert not ist_grossgeschrieben("") assert not ist_grossgeschrieben("X") @pytest.fixture def vodafone_caps_xlsx(tmp_path): wb = openpyxl.Workbook() ws = wb.active ws.title = "All" kopf = ["Anschaltdatum", "Kartennummer", "IMEI", "Rufnummer", "Kunde", "Artikelbezeichnung", "Tarif", "TarifKey02", "Prämie in EUR", "Abrechnungslauf", "Vo-Nr", "Lieferantennr", "Text"] for c, h in enumerate(kopf, 1): ws.cell(6, c, h) for r, (kunde, betrag) in enumerate([("HUBERMEIER ", 295), ("Musterfrau", 370), ("BEISPIELER", 235)], 7): for c, v in enumerate((datetime(2026, 9, 1), "1", "", "4917", kunde, "Vergütung aus VVL otelo", "Tarif", "otelo Allnet Flat", betrag, "VF2609", "80210760", "705136", "10000"), 1): ws.cell(r, c, v) # zweite Seite mit Daten – darf NICHT eingelesen werden ws2 = wb.create_sheet("VF2609") for c, h in enumerate(kopf, 1): ws2.cell(5, c, h) for c, v in enumerate((datetime(2026, 9, 1), "1", "", "4917", "DOPPELT", "Vergütung aus VVL otelo", "T", "x", 999, "VF2609", "80210760", "705136", ""), 1): ws2.cell(6, c, v) p = tmp_path / "vf_caps.xlsx" wb.save(p) return p def test_vodafone_nur_erste_seite(vodafone_caps_xlsx): erg = lies_datei(vodafone_caps_xlsx) assert [p.name for p in erg.positionen] == ["HUBERMEIER", "Musterfrau", "BEISPIELER"] assert erg.warnungen def test_caps_regel_und_feste_zuweisung(db, vodafone_caps_xlsx): from provtool.db import CAPS_ANBIETER, CAPS_WERT laden = db.haendler_speichern(Haendler(0, "Eigener Laden"), [("vodafone", "80210760")]) uh = db.haendler_speichern(Haendler(0, "Unterhändler Caps", standard_prozent=90), [(CAPS_ANBIETER, CAPS_WERT)]) with pytest.raises(ValueError): db.haendler_speichern(Haendler(0, "Noch einer"), [(CAPS_ANBIETER, CAPS_WERT)]) aid = db.abrechnung_neu("September") db.importieren(aid, lies_datei(vodafone_caps_xlsx)) pos = {p["name"]: p for p in db.positionen(aid)} assert pos["HUBERMEIER"]["haendler_id"] == uh and pos["BEISPIELER"]["haendler_id"] == uh assert pos["Musterfrau"]["haendler_id"] == laden assert pos["HUBERMEIER"]["betrag"] == 26550 # 90 % von 295 € # Fehltreffer: BEISPIELER gehört eigentlich zum Laden → fest zuweisen db.haendler_zuweisen([pos["BEISPIELER"]["id"]], laden) mayr = next(p for p in db.positionen(aid) if p["name"] == "BEISPIELER") assert mayr["haendler_id"] == laden and mayr["betrag"] == 23500 # zurück auf automatisch db.haendler_zuweisen([mayr["id"]], None) mayr = next(p for p in db.positionen(aid) if p["name"] == "BEISPIELER") assert mayr["haendler_id"] == uh and mayr["betrag"] == 21150 def test_migration_alter_datenbank(tmp_path): import sqlite3 from provtool.db import SCHEMA pfad = tmp_path / "alt.db" con = sqlite3.connect(pfad) con.executescript(SCHEMA) # Schema v1 ohne Spalte haendler_fest con.close() d = Datenbank(pfad) spalten = {r[1] for r in d.con.execute("PRAGMA table_info(position)")} assert "haendler_fest" in spalten d.close() def test_eigene_kennungen_ausblenden(db, freenet_csv): from provtool.db import EIGEN hid = db.haendler_speichern(Haendler(0, "Händler A"), [("freenet", "111")]) aid = db.abrechnung_neu("X") db.importieren(aid, lies_datei(freenet_csv)) assert {p["haendler_id"] for p in db.positionen(aid)} == {hid, UNZUGEORDNET} db.eigene_setzen("freenet", "999") ids = {p["kennung"]: p["haendler_id"] for p in db.positionen(aid)} assert ids["999"] == EIGEN and ids["111"] == hid # Händler-Kennung hat Vorrang vor „eigen“ db.eigene_setzen("freenet", "111") assert {p["kennung"]: p["haendler_id"] for p in db.positionen(aid)}["111"] == hid db.eigene_setzen("freenet", "999", False) assert {p["kennung"]: p["haendler_id"] for p in db.positionen(aid)}["999"] == UNZUGEORDNET # --------------------------------------------------------------------------- Airtime / Jahresbonus def _pos(art, betrag, unterh="111", rufnr="1601111111", name="Kunde", dienst="Teilnehmerverhaeltnis"): z = freenet_zeile(art, unterh, "05.10.2026", rufnr, name, "Tarif", betrag).split(";") z[24] = dienst return ";".join(z) @pytest.fixture def bonus_csv(tmp_path): zeilen = [FREENET_KOPF, _pos("Grundvergütung Mobilfunk Credit freenet", "50"), _pos("Grundvergütung Mobilfunk Credit freenet", "0"), _pos("Prepaidvergütung", "20", dienst="AirTime Provision"), _pos("Airtime Umsatzbeteiligung", "10,50")] zeilen += [_pos("Jahresbonus Neuaktivierung", "40", rufnr=f"16000000{i:02d}") for i in range(12)] zeilen += [_pos("Jahresbonus Neuaktivierung", "3", rufnr="1600000099")] # weniger als die Stufe p = tmp_path / "bonus.csv" p.write_bytes(("\r\n".join(zeilen) + "\r\n").encode("cp1252")) return p def test_typ_erkennung(): from provtool import regeln assert regeln.erkennen("Prepaidvergütung Vodafone", "AirTime Provision") == regeln.AIRTIME assert regeln.erkennen("Airtime freenet") == regeln.AIRTIME assert regeln.erkennen("Jahresbonus Neuaktivierung") == regeln.JAHRESBONUS assert regeln.erkennen("SIM-Karten", "Teilnehmerverhaeltnis") == regeln.NORMAL def test_staffel_stufen(): from provtool import regeln st = regeln.STANDARD_STAFFEL assert regeln.bonus_betrag(st, 0, 4000) == 0 assert regeln.bonus_betrag(st, 1, 4000) == 500 and regeln.bonus_betrag(st, 9, 4000) == 500 assert regeln.bonus_betrag(st, 10, 4000) == 1000 and regeln.bonus_betrag(st, 14, 4000) == 1000 assert regeln.bonus_betrag(st, 39, 4000) == 3500 assert regeln.bonus_betrag(st, 40, 4000) == 4000 and regeln.bonus_betrag(st, 80, 6000) == 6000 # voll assert regeln.bonus_betrag(st, 25, 1000) == 1000 # nie mehr als laut Abrechnung assert regeln.bonus_betrag(st, 12, -4000) == -1000 # Storno mit derselben Stufe with pytest.raises(ValueError): regeln.staffel_pruefen([(1, 500), (1, 600)]) assert "1-9: 5,00 €" in regeln.staffel_beschreibung(st) and "ab 40: voll" in regeln.staffel_beschreibung(st) def test_airtime_und_jahresbonus(db, bonus_csv, tmp_path): from provtool import regeln hid = db.haendler_speichern(Haendler(0, "Sub", standard_prozent=100), [("freenet", "111")]) aid = db.abrechnung_neu("Okt") db.importieren(aid, lies_datei(bonus_csv)) pos = db.positionen(aid) air = [p for p in pos if p["typ"] == regeln.AIRTIME] bonus = [p for p in pos if p["typ"] == regeln.JAHRESBONUS] assert len(air) == 2 and len(bonus) == 13 assert all(p["betrag"] == 0 for p in air), "Standard: 0 % Airtime = keine Airtime" # 13 Karten -> Stufe 10-14 = 10 € je Karte (die 3-€-Position bleibt bei 3 €) assert sorted(p["betrag"] for p in bonus) == [300] + [1000] * 12 assert all(p["prozent"] is None for p in bonus) # Airtime 50 %, eigene Staffel h = db.haendler(hid) h.airtime_prozent = 50 h.staffel = [(1, 200), (13, None)] db.haendler_speichern(h, db.kennungen(hid)) db.neu_berechnen(aid) pos = db.positionen(aid) assert sorted(p["betrag"] for p in pos if p["typ"] == regeln.AIRTIME) == [525, 1000] assert sorted(p["betrag"] for p in pos if p["typ"] == regeln.JAHRESBONUS) == [300] + [4000] * 12 # Zuweisung ändert die Anzahl Karten -> Staffel der übrigen wird neu berechnet weg = [p["id"] for p in pos if p["typ"] == regeln.JAHRESBONUS][:2] db.haendler_zuweisen(weg, UNZUGEORDNET) rest = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and p["haendler_id"] == hid] assert len(rest) == 11 and all(p["betrag"] == min(200, p["betrag_orig"]) for p in rest) # Typ fest einstellen db.art_typ_setzen("Airtime Umsatzbeteiligung", regeln.NORMAL) assert any(a == "Airtime Umsatzbeteiligung" for a, _, _ in db.provisionsarten()) db.neu_berechnen(aid) p = next(p for p in db.positionen(aid) if p["provisionsart"] == "Airtime Umsatzbeteiligung") assert p["typ"] == regeln.NORMAL and p["betrag"] == 1050 db.art_typ_setzen("Airtime Umsatzbeteiligung", None) # Export: Abrechnung mit Airtime-Summenzeile + eigene Airtime-Liste, 0-€-Positionen weg db.neu_berechnen(aid) eigene = [p for p in db.positionen(aid) if p["haendler_id"] == hid] dateien = export.exportiere_haendler(tmp_path / "out", "Sub", eigene, "Okt") assert [d.name for d in dateien] == ["Sub - Okt.xlsx", "Sub - Airtime - Okt.xlsx"] zeilen = [r for r in openpyxl.load_workbook(dateien[0]).active.iter_rows(values_only=True)] daten = [r for r in zeilen if r[1]] assert all(r[10] != 0 for r in daten), "0-€-Positionen dürfen nicht im Export stehen" assert daten[-1][1] == "Airtime" and daten[-1][10] == 15.25 assert not any(r[1] == "Prepaidvergütung" for r in daten), "Airtime-Einzelposten gehören in die eigene Liste" air_zeilen = [r for r in openpyxl.load_workbook(dateien[1]).active.iter_rows(values_only=True) if r[1]] assert sorted(r[10] for r in air_zeilen) == [5.25, 10.0] # 0 % Airtime: keine Airtime-Zeile, keine Airtime-Datei h = db.haendler(hid) h.airtime_prozent = 0 db.haendler_speichern(h, db.kennungen(hid)) db.neu_berechnen(aid) eigene = [p for p in db.positionen(aid) if p["haendler_id"] == hid] dateien = export.exportiere_haendler(tmp_path / "out0", "Sub", eigene, "Okt") assert [d.name for d in dateien] == ["Sub - Okt.xlsx"] assert not any(r[1] == "Airtime" for r in openpyxl.load_workbook(dateien[0]).active.iter_rows(values_only=True)) def test_migration_v2_haendler_ohne_airtime(tmp_path): import sqlite3 pfad = tmp_path / "v2.db" con = sqlite3.connect(pfad) con.executescript("""CREATE TABLE haendler (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE COLLATE NOCASE, ansprechpartner TEXT DEFAULT '', email TEXT DEFAULT '', notiz TEXT DEFAULT '', standard_prozent REAL NOT NULL DEFAULT 100, aktiv INTEGER NOT NULL DEFAULT 1); INSERT INTO haendler (name, standard_prozent) VALUES ('Alt', 90);""") con.close() d = Datenbank(pfad) h = d.haendler_liste()[0] assert h.name == "Alt" and h.airtime_prozent == 0 and h.staffel is None assert d.staffel_fuer(h.id) == d.standard_staffel() d.standard_staffel_setzen([(1, 100), (5, None)]) assert d.staffel_fuer(h.id) == [(1, 100), (5, None)] d.close()