268 lines
12 KiB
Python
268 lines
12 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")
|
||
assert [d.name for d in dateien] == ["Händler A - freenet - Oktober 2026.xlsx",
|
||
"Händler A - Vodafone - Oktober 2026.xlsx"]
|
||
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()
|