699 lines
34 KiB
Python
699 lines
34 KiB
Python
"""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 ", "80000002 ", "700000", "SMART3L"),
|
||
(datetime(2026, 9, 8), "3225601 ", "", "491726400000 ", "Testkunde ", "Freischaltungsvergütung Vodafone",
|
||
"Basispreisrabatt", "Promotion", -40, "VF2609 ", "80000002 ", "700000", "SMART3L"),
|
||
(datetime(2026, 9, 3), "", "", "160000001", "Kabelkunde", "Freischaltungsvergütung Vodafone Cable",
|
||
"VF Cable/GigaZuhause 50", "VF Internet Cable", 120, "VF2609", "EU000001", "700000", "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 == "80000002" # 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", "80000002")])
|
||
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"]
|
||
zeilen1 = [r for r in openpyxl.load_workbook(eine[0]).active.iter_rows(values_only=True) if r[1]]
|
||
arten = [r[1] for r in zeilen1]
|
||
assert "Freischaltungsvergütung Vodafone" in arten and "SIM-Karten" in arten
|
||
# Spalte A: Vodafone mit Aktivierungsdatum, freenet mit Zeitraum
|
||
vf = next(r for r in zeilen1 if r[1] == "Freischaltungsvergütung Vodafone")
|
||
assert vf[0] == datetime(2026, 9, 8)
|
||
assert next(r for r in zeilen1 if r[1] == "SIM-Karten")[0] == "2026.10.01-2026.10.07"
|
||
ws1 = openpyxl.load_workbook(eine[0]).active
|
||
zeile_vf = next(c.row for c in ws1["B"] if c.value == "Freischaltungsvergütung Vodafone")
|
||
assert ws1.cell(zeile_vf, 1).number_format == "DD.MM.YYYY"
|
||
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", "80000001", "700000", "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", "80000001", "700000", ""), 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", "80000001")])
|
||
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()
|
||
|
||
|
||
# --------------------------------------------------------------------------- Zeitraum / abgerechnet / erledigt
|
||
|
||
def test_zeitraum_lesen_und_bezugsdatum():
|
||
from datetime import date
|
||
from provtool import zeitraum as zr
|
||
assert zr.datum_lesen("1.10.2026") == date(2026, 10, 1) and zr.datum_lesen("01.10.26") == date(2026, 10, 1)
|
||
assert zr.datum_lesen("") is None
|
||
with pytest.raises(ValueError):
|
||
zr.datum_lesen("32.13.2026")
|
||
assert zr.abrechnungsdatum("2026.10.08-2026.10.15") == date(2026, 10, 8)
|
||
assert zr.abrechnungsdatum("VF2609") == date(2026, 9, 1) == zr.abrechnungsdatum("CY202609UM")
|
||
assert zr.abrechnungsdatum("VF2609P6 Fil 0") == date(2026, 9, 1)
|
||
p = {"zeitraum": "VF2609", "datum": "2026-08-30"}
|
||
assert zr.bezugsdatum(p, zr.AKTIVIERUNG) == date(2026, 8, 30)
|
||
assert zr.bezugsdatum(p, zr.ABRECHNUNG) == date(2026, 9, 1)
|
||
okt = zr.Zeitraum(date(2026, 10, 1), date(2026, 10, 31))
|
||
assert okt.enthaelt({"datum": "2026-10-31"}) and not okt.enthaelt({"datum": "2026-11-01"})
|
||
assert not okt.enthaelt({"datum": None})
|
||
assert zr.vormonat(date(2026, 3, 15)) == (date(2026, 2, 1), date(2026, 2, 28))
|
||
assert zr.dieser_monat(date(2026, 12, 5)) == (date(2026, 12, 1), date(2026, 12, 31))
|
||
|
||
|
||
def test_abgerechnet_einfrieren_und_doppelt_erkennen(db, freenet_csv):
|
||
from provtool.db import AbrechnungErledigt
|
||
hid = db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")])
|
||
a1 = db.abrechnung_neu("Okt")
|
||
db.importieren(a1, lies_datei(freenet_csv))
|
||
pos = [p for p in db.positionen(a1) if p["haendler_id"] == hid]
|
||
assert len(pos) == 4 and all(p["fp"] for p in pos)
|
||
erste = pos[:2]
|
||
assert db.abrechnen([p["id"] for p in erste]) == 2
|
||
# abgerechnet: Betrag und Händler eingefroren
|
||
h = db.haendler(hid)
|
||
h.standard_prozent = 50
|
||
db.haendler_speichern(h, []) # Kennung weg, 50 %
|
||
db.neu_berechnen(a1)
|
||
nach = {p["id"]: p for p in db.positionen(a1)}
|
||
for p in erste:
|
||
assert nach[p["id"]]["haendler_id"] == hid and nach[p["id"]]["betrag"] == p["betrag"]
|
||
with pytest.raises(AbrechnungErledigt):
|
||
db.betrag_setzen(erste[0]["id"], 1)
|
||
assert db.prozent_anwenden([erste[0]["id"]], 10) == 0
|
||
|
||
# dieselbe Datei in einer neuen Abrechnung: die beiden abgerechneten Positionen sind „doppelt“
|
||
db.haendler_speichern(db.haendler(hid), [("freenet", "111")])
|
||
a2 = db.abrechnung_neu("Nov")
|
||
db.importieren(a2, lies_datei(freenet_csv))
|
||
# (2 davon schon abgerechnet, die 3 anderen sind in „Okt“ noch offen – auch die zählen nur einmal)
|
||
doppelt = [p for p in db.positionen(a2) if p["doppelt"]]
|
||
assert len(doppelt) == 5 and all(p["doppelt"] == "Okt" for p in doppelt)
|
||
assert db.abrechnen([p["id"] for p in doppelt]) == 0, "doppelte Positionen dürfen nicht erneut abgerechnet werden"
|
||
|
||
# Kennzeichnung aufheben -> die Positionen in „Okt“ sind wieder offen und bleiben die zuerst importierten;
|
||
# „Nov“ bleibt doppelt. Erst wenn „Okt“ weg ist, ist „Nov“ frei.
|
||
assert db.abrechnung_aufheben([p["id"] for p in erste]) == 2
|
||
assert all(p["doppelt"] == "Okt" for p in db.positionen(a2))
|
||
db.abrechnung_loeschen(a1)
|
||
assert not any(p["doppelt"] for p in db.positionen(a2))
|
||
|
||
|
||
def test_abrechnung_erledigt_sperrt(db, freenet_csv):
|
||
from provtool.db import AbrechnungErledigt
|
||
hid = db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")])
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
assert db.erledigt_setzen(aid, True) == 4 # nur die 4 Händler-Positionen, nicht die unzugeordnete
|
||
assert db.erledigt(aid)
|
||
assert all(p["abgerechnet"] for p in db.positionen(aid) if p["haendler_id"] == hid)
|
||
with pytest.raises(AbrechnungErledigt):
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
with pytest.raises(AbrechnungErledigt):
|
||
db.datei_entfernen(db.dateien(aid)[0]["id"])
|
||
unz = next(p for p in db.positionen(aid) if p["haendler_id"] <= 0)
|
||
db.haendler_zuweisen([unz["id"]], hid)
|
||
assert next(p for p in db.positionen(aid) if p["id"] == unz["id"])["haendler_id"] <= 0, "erledigt: keine Änderung"
|
||
assert db.abrechnungen()[0]["abgerechnet"] == 4
|
||
# wieder öffnen, Positionen bleiben abgerechnet bzw. werden auf Wunsch freigegeben
|
||
db.erledigt_setzen(aid, False)
|
||
assert not db.erledigt(aid) and all(p["abgerechnet"] for p in db.positionen(aid) if p["haendler_id"] == hid)
|
||
db.erledigt_setzen(aid, True)
|
||
db.erledigt_setzen(aid, False, positionen_aufheben=True)
|
||
assert not any(p["abgerechnet"] for p in db.positionen(aid))
|
||
|
||
|
||
def test_migration_v3_fingerabdruck(tmp_path, freenet_csv):
|
||
import sqlite3
|
||
d = Datenbank(tmp_path / "x.db")
|
||
aid = d.abrechnung_neu("Alt")
|
||
d.importieren(aid, lies_datei(freenet_csv))
|
||
d.close()
|
||
con = sqlite3.connect(tmp_path / "x.db") # wie eine v3-Datenbank: ohne Fingerabdrücke
|
||
con.execute("UPDATE position SET fp=NULL")
|
||
con.commit()
|
||
con.close()
|
||
d = Datenbank(tmp_path / "x.db")
|
||
assert all(p["fp"] for p in d.positionen(aid))
|
||
assert len({p["fp"] for p in d.positionen(aid)}) == 5
|
||
d.close()
|
||
|
||
|
||
# --------------------------------------------------------------------------- Audit-Befunde (Nachtests)
|
||
|
||
def test_offene_doppelte_nur_einmal_exportieren_und_abrechnen(db, freenet_csv, tmp_path):
|
||
"""Wochen- und Monatsdatei mit denselben Zeilen in EINER Abrechnung: die zweite ist „doppelt“ und
|
||
wird weder exportiert noch abgerechnet. Zwei gleiche Zeilen in EINER Datei bleiben zwei Positionen."""
|
||
hid = db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")])
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
monat = tmp_path / "monat.csv"
|
||
monat.write_bytes(freenet_csv.read_bytes() + freenet_csv.read_bytes().split(b"\r\n", 1)[1]) # Zeilen 2x
|
||
db.importieren(aid, lies_datei(monat))
|
||
pos = db.positionen(aid)
|
||
assert len(pos) == 15
|
||
erste_datei = [p for p in pos if p["datei_id"] == db.dateien(aid)[0]["id"]]
|
||
assert not any(p["doppelt"] for p in erste_datei), "die zuerst importierte Datei zählt"
|
||
zweite = [p for p in pos if p["datei_id"] != db.dateien(aid)[0]["id"]]
|
||
assert sum(1 for p in zweite if p["doppelt"] == "Okt") == 5, "je Zeile ist nur die erste Wiederholung doppelt"
|
||
assert sum(1 for p in zweite if not p["doppelt"]) == 5, "zweite Vorkommen innerhalb der Datei (#2) sind eigene Positionen"
|
||
# abrechnen aller Händler-Positionen: Doppelte bleiben offen
|
||
n = db.abrechnen([p["id"] for p in pos])
|
||
offen_haendler = [p for p in db.positionen(aid) if p["haendler_id"] == hid and not p["abgerechnet"]]
|
||
assert n == 8 and all(p["doppelt"] for p in offen_haendler)
|
||
# und noch einmal: derselbe Fingerabdruck wird in der Transaktion erneut geprüft
|
||
assert db.abrechnen([p["id"] for p in db.positionen(aid)]) == 0
|
||
|
||
|
||
def test_staffel_ohne_doppelte_und_datei_entfernen_rechnet_neu(db, bonus_csv, tmp_path):
|
||
from provtool import regeln
|
||
hid = db.haendler_speichern(Haendler(0, "Sub"), [("freenet", "111")])
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(bonus_csv))
|
||
kopie = tmp_path / "bonus2.csv"
|
||
kopie.write_bytes(bonus_csv.read_bytes())
|
||
db.importieren(aid, lies_datei(kopie))
|
||
bonus = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and p["haendler_id"] == hid]
|
||
assert len(bonus) == 26 and sum(1 for p in bonus if p["doppelt"]) == 13
|
||
echte = [p for p in bonus if not p["doppelt"]]
|
||
assert sorted(p["betrag"] for p in echte) == [300] + [1000] * 12, "Doppelte zählen nicht als Karten (13, nicht 26)"
|
||
# zweite Datei (3 Zeilen) zusätzlich mit anderen Rufnummern -> 16 Karten -> Stufe 15: 15 € je Karte
|
||
extra = tmp_path / "extra.csv"
|
||
extra.write_bytes(("\r\n".join([FREENET_KOPF] + [_pos("Jahresbonus Neuaktivierung", "40", rufnr=f"17000000{i:02d}")
|
||
for i in range(3)]) + "\r\n").encode("cp1252"))
|
||
db.importieren(aid, lies_datei(extra))
|
||
echte = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and not p["doppelt"]]
|
||
assert sorted(p["betrag"] for p in echte) == [300] + [1500] * 15
|
||
# Datei entfernen -> Staffel der übrigen wieder 10 € je Karte
|
||
db.datei_entfernen(db.dateien(aid)[-1]["id"])
|
||
echte = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and not p["doppelt"]]
|
||
assert sorted(p["betrag"] for p in echte) == [300] + [1000] * 12
|
||
|
||
|
||
def test_abgerechnete_positionen_nicht_loeschbar(db, freenet_csv):
|
||
"""Datei/Abrechnung mit abgerechneten Positionen: Löschen verweigert (sonst Fingerabdruck-Leichen
|
||
oder doppelte Auszahlung). Nach Aufheben der Kennzeichnung geht es – und hinterlässt keine Fingerabdrücke."""
|
||
from provtool.db import AbrechnungErledigt
|
||
db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")])
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
pos = db.positionen(aid)
|
||
assert db.abrechnen([pos[0]["id"]]) == 1
|
||
with pytest.raises(AbrechnungErledigt):
|
||
db.datei_entfernen(db.dateien(aid)[0]["id"])
|
||
with pytest.raises(AbrechnungErledigt):
|
||
db.abrechnung_loeschen(aid)
|
||
assert len(db.positionen(aid)) == 5
|
||
db.abrechnung_aufheben([pos[0]["id"]])
|
||
db.abrechnung_loeschen(aid)
|
||
assert db.con.execute("SELECT count(*) FROM abgerechnet_fp").fetchone()[0] == 0
|
||
a2 = db.abrechnung_neu("Nov")
|
||
db.importieren(a2, lies_datei(freenet_csv))
|
||
assert not any(p["doppelt"] for p in db.positionen(a2))
|
||
|
||
|
||
def test_erledigt_offene_positionen_bekommen_keinen_haendler_mehr(db, freenet_csv):
|
||
"""Erledigt: eine damals nicht zugeordnete Position darf einem später angelegten Händler nicht mit
|
||
100 % zur Auszahlung erscheinen; abrechnen() fasst erledigte Abrechnungen nicht an."""
|
||
db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")])
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
db.erledigt_setzen(aid, True)
|
||
hid_b = db.haendler_speichern(Haendler(0, "B", standard_prozent=50), [("freenet", "999")])
|
||
p = next(p for p in db.positionen(aid) if p["kennung"] == "999")
|
||
assert p["haendler_id"] == UNZUGEORDNET and not p["abgerechnet"]
|
||
assert db.abrechnen([p["id"]]) == 0
|
||
# in einer offenen Abrechnung gehört dieselbe Kennung sehr wohl zu B (mit 50 %)
|
||
a2 = db.abrechnung_neu("Nov")
|
||
db.importieren(a2, lies_datei(freenet_csv))
|
||
q = next(p for p in db.positionen(a2) if p["kennung"] == "999")
|
||
assert q["haendler_id"] == hid_b and q["betrag"] == -400
|
||
|
||
|
||
def test_export_text_mit_gleichheitszeichen_bleibt_text(tmp_path):
|
||
pos = [{"anbieter": "freenet", "zeitraum": "2026.10.01-2026.10.07", "provisionsart": "=1+1", "kennung": "111",
|
||
"datum": "2026-10-05", "rufnummer": "1601", "imei": "", "geraet": "", "vorname": "",
|
||
"name": '=HYPERLINK("http://böse.example","Klick")', "tarif": "T", "betrag": 800, "betrag_orig": 800,
|
||
"netz": "", "dienst": "", "aktionscode": "=cmd|' /C calc'!A0", "typ": "normal", "reihenfolge": 1}]
|
||
datei = export.exportiere_haendler(tmp_path, "A", pos, "Okt")[0]
|
||
ws = openpyxl.load_workbook(datei).active
|
||
zeile = [c for c in ws[1]]
|
||
assert zeile[1].value == "=1+1" and zeile[1].data_type == "s"
|
||
assert zeile[8].value.startswith("=HYPERLINK") and zeile[8].data_type == "s"
|
||
assert zeile[13].data_type == "s"
|
||
summe = next(c for row in ws.iter_rows() for c in row if isinstance(c.value, str) and c.value.startswith("=SUM"))
|
||
assert summe.data_type == "f", "unsere Summenformel bleibt eine Formel"
|
||
|
||
|
||
def test_export_gleiche_dateinamen_ueberschreiben_sich_nicht(tmp_path):
|
||
p = {"anbieter": "freenet", "zeitraum": "", "provisionsart": "X", "kennung": "1", "datum": None, "rufnummer": "",
|
||
"imei": "", "geraet": "", "vorname": "", "name": "K", "tarif": "", "betrag": 100, "betrag_orig": 100,
|
||
"netz": "", "dienst": "", "aktionscode": "", "typ": "normal", "reihenfolge": 1}
|
||
vergeben: set[str] = set()
|
||
a = export.exportiere_haendler(tmp_path, "A/B", [p], "Okt", vergeben=vergeben)
|
||
b = export.exportiere_haendler(tmp_path, "A:B", [dict(p, betrag=200)], "Okt", vergeben=vergeben)
|
||
assert a[0].name == "A_B - Okt.xlsx" and b[0].name == "A_B - Okt (2).xlsx"
|
||
assert openpyxl.load_workbook(a[0]).active["K1"].value == 1.0
|
||
assert openpyxl.load_workbook(b[0]).active["K1"].value == 2.0
|
||
|
||
|
||
def test_restore_prueft_sicherung(db, freenet_csv, tmp_path):
|
||
import sqlite3
|
||
aid = db.abrechnung_neu("Okt")
|
||
db.importieren(aid, lies_datei(freenet_csv))
|
||
db.close()
|
||
ziel = db.path
|
||
vorher = ziel.read_bytes()
|
||
# keine SQLite-Datei
|
||
muell = tmp_path / "muell.db"
|
||
muell.write_bytes(b"kein sqlite")
|
||
with pytest.raises(Exception):
|
||
Datenbank.restore(muell, ziel)
|
||
# SQLite, aber nicht unsere Tabellen
|
||
fremd = tmp_path / "fremd.db"
|
||
con = sqlite3.connect(fremd)
|
||
con.execute("CREATE TABLE haendler (id INTEGER)")
|
||
con.commit()
|
||
con.close()
|
||
with pytest.raises(ValueError):
|
||
Datenbank.restore(fremd, ziel)
|
||
assert ziel.read_bytes() == vorher and not ziel.with_suffix(".pruefung.db").exists()
|
||
# echte Sicherung
|
||
d = Datenbank(ziel)
|
||
sicherung = tmp_path / "sicherung.db"
|
||
d.backup(sicherung)
|
||
d.abrechnung_loeschen(aid)
|
||
d.close()
|
||
Datenbank.restore(sicherung, ziel)
|
||
d = Datenbank(ziel)
|
||
assert len(d.positionen(aid)) == 5
|
||
d.close()
|
||
|
||
|
||
def test_import_grenzen(tmp_path, freenet_csv, vodafone_xlsx, monkeypatch):
|
||
from provtool import importer
|
||
monkeypatch.setattr(importer, "MAX_ZEILEN", 3)
|
||
with pytest.raises(ImportFehler, match="Zeilen"):
|
||
lies_datei(freenet_csv)
|
||
monkeypatch.setattr(importer, "MAX_ZEILEN", 200_000)
|
||
monkeypatch.setattr(importer, "MAX_BYTES", 10)
|
||
with pytest.raises(ImportFehler, match="MB"):
|
||
lies_datei(freenet_csv)
|
||
monkeypatch.setattr(importer, "MAX_BYTES", vodafone_xlsx.stat().st_size + 1) # gepackt klein, entpackt zu groß
|
||
with pytest.raises(ImportFehler, match="entpackt"):
|
||
lies_datei(vodafone_xlsx)
|
||
|
||
|
||
def test_einstellungen_atomar_und_defekt(tmp_path, monkeypatch):
|
||
from provtool import config
|
||
monkeypatch.setenv("PROVTOOL_HOME", str(tmp_path))
|
||
config.save({"db_pfad": "Z:/x.db", "defekt": "weg"})
|
||
assert not (tmp_path / "einstellungen.json.tmp").exists()
|
||
assert config.load()["db_pfad"] == "Z:/x.db" and "defekt" not in config.load()
|
||
(tmp_path / "einstellungen.json").write_text("{ kaputt", encoding="utf-8")
|
||
daten = config.load()
|
||
assert daten["db_pfad"] is None and daten["defekt"].endswith("einstellungen.defekt.json")
|
||
assert (tmp_path / "einstellungen.defekt.json").read_text(encoding="utf-8") == "{ kaputt"
|