Neue Rechnungen tauchen von selbst auf - das Journal hoert auf die gemeinsame SQLite
Stehen beide Programme offen und wird im Rechnungstool eine Rechnung geschrieben, erscheint sie hier ohne Zutun. Das Journal traegt sich dafuer beim Start mit seiner PID in der gemeinsamen SQLite ein (Tabelle "laeuft") und traegt sich beim Schliessen wieder aus; das Rechnungstool sieht daran, dass jemand zuhoert, und legt eine Zeile in die Tabelle "nachrichten". Nachgesehen wird nur dort - erst ein os.stat auf die eine Datei, und nur wenn sich etwas geruehrt hat, wird sie geoeffnet. Kein Absuchen des Rechnungsordners, kein Server, kein Port. Beim Zurueckklicken ins Fenster wird sofort nachgesehen statt zwei Sekunden zu warten. Die Wache faengt jeden Fehler ab und stirbt nie - ein Netzlaufwerk, das kurz weg ist, darf das Fenster nicht lahmlegen. ⚠ Ein Denkfehler daraus gleich behoben: der erste Blick nach dem Start merkt sich den Stand AUCH dann, wenn es die Datei noch gar nicht gibt. Sonst galt die erste Meldung danach als "schon gesehen" und wurde verschluckt. Pruefstand pruef_wache.py (faehrt BEIDE Programme): 23 Pruefungen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GYeAfLtccFrbU3MTj1MMTx
This commit is contained in:
TheMockTv
parent
59d2db75af
commit
31bb0c600b
3 changed files with 478 additions and 0 deletions
93
app.py
93
app.py
|
|
@ -104,6 +104,13 @@ class App(tk.Tk):
|
|||
if self.pdf_ordner and os.path.isdir(self.pdf_ordner):
|
||||
self.after(200, lambda: self.einlesen(still=True))
|
||||
self.after(400, self._frage_gemeinsam_falls_noetig)
|
||||
# Beide Programme laufen nebeneinander: schreibt das Rechnungstool eine
|
||||
# Rechnung, meldet es das hier herein - siehe _wache_jetzt().
|
||||
self._briefkasten_stand = "?" # "?" = noch nicht nachgesehen
|
||||
self._nachricht_stand = 0 # bis hierher sind Meldungen gelesen
|
||||
self.after(600, self._wache)
|
||||
# Beim Zurueckklicken sofort nachsehen statt zwei Sekunden zu warten.
|
||||
self.bind("<FocusIn>", lambda _e: self._wache_jetzt())
|
||||
|
||||
# ------------------------------------------------------------ Aufbau ----
|
||||
def _menue(self):
|
||||
|
|
@ -253,6 +260,62 @@ class App(tk.Tk):
|
|||
text=f"Rechnungen: {self.pdf_ordner}" if self.pdf_ordner
|
||||
else "kein Rechnungs-Ordner gewählt (Datei → Rechnungs-Ordner wählen…)")
|
||||
|
||||
# ------------------------------------------------------------- Wache
|
||||
# Das Rechnungstool MELDET, wenn es eine Rechnung geschrieben hat - es legt
|
||||
# eine Zeile in den Briefkasten im gemeinsamen Ordner. Hier wird nur diese
|
||||
# eine Datei angesehen (ein os.stat), nicht der ganze Rechnungsordner.
|
||||
# Damit das Rechnungstool weiss, dass jemand zuhoert, traegt sich dieses
|
||||
# Fenster beim Start mit seiner PID ein (siehe _anmelden).
|
||||
WACHE_SEKUNDEN = 2
|
||||
|
||||
def _wache_jetzt(self):
|
||||
"""Einmal nachsehen. True, wenn eine Meldung da war und eingelesen wurde.
|
||||
|
||||
Billig gehalten: erst ein os.stat auf die gemeinsame Datei. Nur wenn
|
||||
sich dort etwas geruehrt hat, wird sie ueberhaupt geoeffnet.
|
||||
"""
|
||||
if self.buch is None:
|
||||
return False
|
||||
stand = gemeinsam.datei_stand(self.buch.ordner)
|
||||
if self._briefkasten_stand == "?":
|
||||
# Erster Blick nach dem Start: nur merken, auch wenn es die Datei
|
||||
# noch gar nicht gibt - sonst gilt die erste Meldung danach als
|
||||
# "schon gesehen" und wird verschluckt.
|
||||
self._briefkasten_stand = stand
|
||||
self._nachricht_stand = self._letzte_nachricht()
|
||||
return False
|
||||
if stand is None or stand == self._briefkasten_stand:
|
||||
return False
|
||||
self._briefkasten_stand = stand
|
||||
try:
|
||||
texte, neu = self.buch.nachrichten_ab(self._nachricht_stand)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("Briefkasten nicht lesbar: %s", e)
|
||||
return False
|
||||
if not texte:
|
||||
return False # das Buch hat sich geaendert, aber ohne Meldung
|
||||
self._nachricht_stand = neu
|
||||
text = texte[-1] if len(texte) == 1 else f"{len(texte)} Meldungen, zuletzt: {texte[-1]}"
|
||||
log.info("Rechnungstool meldet: %s", text)
|
||||
self.einlesen(still=True)
|
||||
self.melde(f"Rechnungstool meldet: {text}")
|
||||
return True
|
||||
|
||||
def _letzte_nachricht(self):
|
||||
try:
|
||||
return self.buch.letzte_nachricht() if self.buch else 0
|
||||
except Exception: # noqa: BLE001
|
||||
return 0
|
||||
|
||||
def _wache(self):
|
||||
"""Der Wecker. Laeuft weiter, auch wenn ein Durchgang schiefgeht."""
|
||||
try:
|
||||
self._wache_jetzt()
|
||||
except Exception as e: # noqa: BLE001 - die Wache darf nie sterben
|
||||
log.warning("Wache: %s", e)
|
||||
finally:
|
||||
self.after(self.WACHE_SEKUNDEN * 1000, self._wache)
|
||||
|
||||
def suche_leeren(self):
|
||||
self.suche.set("")
|
||||
self.suchfeld.focus_set()
|
||||
|
|
@ -262,6 +325,32 @@ class App(tk.Tk):
|
|||
log.info(text)
|
||||
|
||||
# ---- Gemeinsames Nummernbuch -------------------------------------------
|
||||
def _anmelden(self):
|
||||
"""Mit der eigenen PID im gemeinsamen Ordner eintragen.
|
||||
|
||||
Daran sieht das Rechnungstool, dass dieses Fenster offen ist - es fragt
|
||||
Windows nach der Prozessnummer, nicht nach einem Fenster.
|
||||
"""
|
||||
ordner = self.journal.hole("gemeinsam_ordner", "")
|
||||
if not ordner:
|
||||
return
|
||||
if self.buch is None:
|
||||
self._buch_oeffnen()
|
||||
if self.buch is None:
|
||||
return
|
||||
try:
|
||||
self.buch.anmelden("journal", os.getpid(), os.path.basename(sys.executable))
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("Anmeldung fehlgeschlagen: %s", e)
|
||||
|
||||
def _abmelden(self):
|
||||
if self.buch is None:
|
||||
return
|
||||
try:
|
||||
self.buch.abmelden("journal")
|
||||
except Exception as e: # noqa: BLE001 - beim Schliessen nichts erzwingen
|
||||
log.debug("Abmelden fehlgeschlagen: %s", e)
|
||||
|
||||
def _buch_oeffnen(self):
|
||||
ordner = self.journal.hole("gemeinsam_ordner", "")
|
||||
if not ordner:
|
||||
|
|
@ -284,6 +373,7 @@ class App(tk.Tk):
|
|||
self._buch_oeffnen()
|
||||
if self.buch is None and not TESTLAUF:
|
||||
self.dlg_gemeinsam(erststart=True)
|
||||
self._anmelden()
|
||||
self._buch_fuellen()
|
||||
|
||||
def dlg_gemeinsam(self, erststart=False):
|
||||
|
|
@ -300,6 +390,7 @@ class App(tk.Tk):
|
|||
return
|
||||
self.journal.setze("gemeinsam_ordner", ordner)
|
||||
self._buch_oeffnen()
|
||||
self._anmelden()
|
||||
self._buch_fuellen(melden=True)
|
||||
|
||||
def _buch_fuellen(self, melden=False):
|
||||
|
|
@ -669,6 +760,8 @@ class App(tk.Tk):
|
|||
os.startfile(pfad) # noqa: S606
|
||||
|
||||
def _beenden(self):
|
||||
# Austragen, damit das Rechnungstool nicht weiter meldet.
|
||||
self._abmelden()
|
||||
self.journal.setze("jahr", self.jahr.get())
|
||||
self.journal.schliessen()
|
||||
self.destroy()
|
||||
|
|
|
|||
155
gemeinsam.py
155
gemeinsam.py
|
|
@ -27,6 +27,8 @@ hinueber - sonst schreiben zwei Fassungen in dieselbe Datei.
|
|||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import time
|
||||
|
||||
log = logging.getLogger("gemeinsam")
|
||||
|
||||
|
|
@ -56,6 +58,26 @@ CREATE TABLE IF NOT EXISTS nummern (
|
|||
geaendert TEXT NOT NULL DEFAULT '' -- ISO-Zeitstempel
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_nummern_jahr ON nummern (jahr);
|
||||
|
||||
-- Wer gerade laeuft. Eine Zeile je Programm, mit PID und EXE-Namen: ob es
|
||||
-- wirklich noch laeuft, beantwortet Windows (prozess_laeuft), nicht ein
|
||||
-- Zeitstempel. Ein abgestuerztes Programm hinterlaesst hier nur eine
|
||||
-- Karteileiche, die beim naechsten Nachsehen weggeraeumt wird.
|
||||
CREATE TABLE IF NOT EXISTS laeuft (
|
||||
programm TEXT PRIMARY KEY, -- 'journal' | 'rechnungstool'
|
||||
pid INTEGER NOT NULL DEFAULT 0,
|
||||
exe TEXT NOT NULL DEFAULT '',
|
||||
seit TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
|
||||
-- Der Briefkasten. Das Rechnungstool legt hier ab, dass es eine Rechnung
|
||||
-- geschrieben hat; das Journal liest die neuen Zeilen und liest die PDFs nach.
|
||||
CREATE TABLE IF NOT EXISTS nachrichten (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
zeit TEXT NOT NULL DEFAULT '',
|
||||
absender TEXT NOT NULL DEFAULT '',
|
||||
text TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
"""
|
||||
|
||||
FELDER = ("jahr", "datum", "nachname", "vorname", "naechte", "zwischensumme",
|
||||
|
|
@ -84,6 +106,78 @@ def ordner_taugt(ordner):
|
|||
return True, ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Briefkasten und Herzschlag: das Rechnungstool sagt dem Steuerjournal Bescheid
|
||||
#
|
||||
# Kein Programm durchsucht dauernd den Rechnungsordner. Stattdessen legt das
|
||||
# LAUFENDE Steuerjournal einen Herzschlag im gemeinsamen Ordner ab, und das
|
||||
# Rechnungstool schaut nach dem Schreiben einer Rechnung nur kurz hinein: laeuft
|
||||
# es, wirft es eine Nachricht in den Briefkasten. Das Journal sieht genau EINE
|
||||
# Datei nach (os.stat) und liest die Rechnung dann von selbst ein.
|
||||
#
|
||||
# Ist das Journal zu, passiert nichts - es liest den Ordner beim naechsten Start
|
||||
# ohnehin komplett. Die Nachricht ist eine Anklopfhilfe, keine Datenquelle.
|
||||
def prozess_laeuft(pid, name=""):
|
||||
"""Laeuft unter dieser Prozessnummer wirklich noch etwas?
|
||||
|
||||
Die Antwort kommt von Windows, nicht aus einer Uhrzeit und nicht aus einer
|
||||
Fenstersuche. PIDs werden wiederverwendet - deshalb wird zusaetzlich der
|
||||
Name der EXE verglichen, wenn einer eingetragen wurde. Sonst haette ein
|
||||
fremder Prozess mit derselben Nummer das andere Programm "am Leben"
|
||||
gehalten.
|
||||
"""
|
||||
if not pid:
|
||||
return False
|
||||
if not sys.platform.startswith("win"):
|
||||
try:
|
||||
os.kill(int(pid), 0)
|
||||
return True
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
PROCESS_QUERY_LIMITED = 0x1000
|
||||
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
||||
k32.OpenProcess.restype = wintypes.HANDLE
|
||||
griff = k32.OpenProcess(PROCESS_QUERY_LIMITED, False, int(pid))
|
||||
if not griff:
|
||||
return False
|
||||
try:
|
||||
if not name:
|
||||
return True
|
||||
puffer = ctypes.create_unicode_buffer(4096)
|
||||
laenge = wintypes.DWORD(len(puffer))
|
||||
if not k32.QueryFullProcessImageNameW(griff, 0, puffer, ctypes.byref(laenge)):
|
||||
return False
|
||||
return os.path.basename(puffer.value).lower() == str(name).lower()
|
||||
finally:
|
||||
k32.CloseHandle(griff)
|
||||
|
||||
|
||||
def datei_stand(ordner):
|
||||
"""(Aenderungszeit, Groesse) der gemeinsamen Datei - oder None.
|
||||
|
||||
Damit sieht ein Programm mit EINEM os.stat, ob sich im Buch etwas getan hat,
|
||||
ohne dafuer die Datenbank zu oeffnen.
|
||||
"""
|
||||
try:
|
||||
st = os.stat(pfad_zur_datei(ordner))
|
||||
except OSError:
|
||||
return None
|
||||
return st.st_mtime, st.st_size
|
||||
|
||||
|
||||
def journal_laeuft(ordner):
|
||||
"""Ist das Steuerjournal gerade offen?"""
|
||||
if not ordner or not os.path.exists(pfad_zur_datei(ordner)):
|
||||
return False
|
||||
try:
|
||||
return Nummernbuch(ordner).laeuft("journal")
|
||||
except Exception as e: # noqa: BLE001 - eine Auskunft darf nichts umwerfen
|
||||
log.debug("Anmeldung nicht lesbar: %s", e)
|
||||
return False
|
||||
|
||||
|
||||
class Nummernbuch:
|
||||
"""Das gemeinsame Register. Wird zum Schreiben kurz geoeffnet und wieder
|
||||
geschlossen - zwei Programme haben die Datei sonst dauerhaft in der Hand,
|
||||
|
|
@ -153,6 +247,67 @@ class Nummernbuch:
|
|||
return {z["storno_zu"] for z in
|
||||
con.execute("SELECT storno_zu FROM nummern WHERE storno_zu <> ''")}
|
||||
|
||||
# ---- Wer laeuft, und der Briefkasten -----------------------------------
|
||||
def anmelden(self, programm, pid, exe):
|
||||
"""Eintragen: dieses Programm ist offen."""
|
||||
import datetime
|
||||
with self._verbindung() as con:
|
||||
con.execute(
|
||||
"INSERT INTO laeuft (programm, pid, exe, seit) VALUES (?,?,?,?) "
|
||||
"ON CONFLICT(programm) DO UPDATE SET pid=excluded.pid, exe=excluded.exe, "
|
||||
"seit=excluded.seit",
|
||||
(str(programm), int(pid), str(exe),
|
||||
datetime.datetime.now().isoformat(timespec="seconds")))
|
||||
log.info("angemeldet: %s (PID %s)", programm, pid)
|
||||
|
||||
def abmelden(self, programm):
|
||||
with self._verbindung() as con:
|
||||
con.execute("DELETE FROM laeuft WHERE programm=?", (str(programm),))
|
||||
|
||||
def laeuft(self, programm):
|
||||
"""Laeuft dieses Programm noch? Karteileichen werden weggeraeumt."""
|
||||
with self._verbindung() as con:
|
||||
z = con.execute("SELECT pid, exe FROM laeuft WHERE programm=?",
|
||||
(str(programm),)).fetchone()
|
||||
if not z:
|
||||
return False
|
||||
if prozess_laeuft(z["pid"], z["exe"]):
|
||||
return True
|
||||
self.abmelden(programm)
|
||||
return False
|
||||
|
||||
def melden(self, absender, text):
|
||||
"""Eine Zeile in den Briefkasten legen."""
|
||||
import datetime
|
||||
with self._verbindung() as con:
|
||||
con.execute("INSERT INTO nachrichten (zeit, absender, text) VALUES (?,?,?)",
|
||||
(datetime.datetime.now().isoformat(timespec="seconds"),
|
||||
str(absender), str(text)))
|
||||
log.info("Nachricht von %s: %s", absender, text)
|
||||
return True
|
||||
|
||||
def letzte_nachricht(self):
|
||||
"""Hoechste Nachrichten-Nummer (0, wenn der Kasten leer ist)."""
|
||||
with self._verbindung() as con:
|
||||
z = con.execute("SELECT MAX(id) AS m FROM nachrichten").fetchone()
|
||||
return int(z["m"] or 0)
|
||||
|
||||
def nachrichten_ab(self, seit_id):
|
||||
"""Alle Nachrichten NACH dieser Nummer: (Liste von Texten, neue Nummer)."""
|
||||
with self._verbindung() as con:
|
||||
zeilen = list(con.execute(
|
||||
"SELECT id, absender, text FROM nachrichten WHERE id > ? ORDER BY id",
|
||||
(int(seit_id or 0),)))
|
||||
if not zeilen:
|
||||
return [], int(seit_id or 0)
|
||||
return [z["text"] for z in zeilen], int(zeilen[-1]["id"])
|
||||
|
||||
def aufraeumen(self, behalten=200):
|
||||
"""Alte Nachrichten wegwerfen - der Briefkasten ist kein Archiv."""
|
||||
with self._verbindung() as con:
|
||||
con.execute("DELETE FROM nachrichten WHERE id <= "
|
||||
"(SELECT MAX(id) FROM nachrichten) - ?", (int(behalten),))
|
||||
|
||||
def entfernen(self, nummer):
|
||||
"""Nur fuer den Rueckweg: ein verworfener Storno gibt seine Nummer zurueck."""
|
||||
with self._verbindung() as con:
|
||||
|
|
|
|||
230
pruef_wache.py
Normal file
230
pruef_wache.py
Normal file
|
|
@ -0,0 +1,230 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""Pruefstand: das Rechnungstool meldet, das Steuerjournal liest nach.
|
||||
|
||||
Sein Fall: beide Programme stehen offen. Er schreibt im Rechnungstool eine
|
||||
Rechnung - dann soll sie hier auftauchen, ohne dass jemand "Rechnungen
|
||||
einlesen" drueckt. Seine Ansage zum Weg:
|
||||
|
||||
"machs einfacher das rechnungstoll schaut ob der steuer ding da ist und
|
||||
meldet dem hier neue rechnung wurde erstellt"
|
||||
|
||||
Der Weg, den er vorgegeben hat: das laufende Journal traegt sich mit seiner PID
|
||||
in der GEMEINSAMEN SQLITE ein ("der briefkasten ist die sqlite"), das
|
||||
Rechnungstool fragt WINDOWS, ob dieser Prozess laeuft ("nicht auf fenster
|
||||
schauen"), und legt dann eine Zeile in den Briefkasten - dieselbe Datei. Das
|
||||
Journal sieht nur dort nach. Geprueft wird mit BEIDEN echten Programmen.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
|
||||
JOURNAL = r"C:\claude\beherbergungssteuer"
|
||||
TOOL = r"C:\claude\rechnungstool"
|
||||
|
||||
fehler = []
|
||||
|
||||
|
||||
def pruefe(bedingung, text):
|
||||
zeile = (" OK | " if bedingung else "FEHLER | ") + text
|
||||
kodierung = sys.stdout.encoding or "utf-8"
|
||||
print(zeile.encode(kodierung, "replace").decode(kodierung))
|
||||
if not bedingung:
|
||||
fehler.append(text)
|
||||
|
||||
|
||||
def _reissleine():
|
||||
print("ABBRUCH | Pruefstand haengt (Zeitgrenze)")
|
||||
os._exit(2)
|
||||
|
||||
|
||||
_wecker = threading.Timer(150, _reissleine)
|
||||
_wecker.daemon = True
|
||||
_wecker.start()
|
||||
|
||||
os.environ["CAMPINGHOF_TEST"] = "1"
|
||||
|
||||
tmp = tempfile.mkdtemp(prefix="bst_melden_")
|
||||
gemeinsam_ordner = os.path.join(tmp, "install")
|
||||
rechnungen = os.path.join(tmp, "rechnungen")
|
||||
daten = os.path.join(tmp, "daten")
|
||||
for d in (gemeinsam_ordner, rechnungen, daten):
|
||||
os.makedirs(d)
|
||||
|
||||
# ---- Das Steuerjournal ----------------------------------------------------
|
||||
sys.path.insert(0, JOURNAL)
|
||||
import ablage # noqa: E402
|
||||
|
||||
ablage.daten_ordner = lambda: daten
|
||||
import app as JA # noqa: E402
|
||||
import gemeinsam # noqa: E402
|
||||
|
||||
del sys.modules["app"]
|
||||
sys.path.remove(JOURNAL)
|
||||
|
||||
from tkinter import messagebox # noqa: E402
|
||||
|
||||
messagebox.showinfo = lambda *a, **k: None
|
||||
messagebox.showwarning = lambda *a, **k: None
|
||||
|
||||
journal = JA.App()
|
||||
journal.withdraw()
|
||||
journal.pdf_ordner = rechnungen
|
||||
journal.journal.setze("pdf_ordner", rechnungen)
|
||||
journal.journal.setze("gemeinsam_ordner", gemeinsam_ordner)
|
||||
journal._buch_oeffnen()
|
||||
journal.jahr.set(2026)
|
||||
journal.update()
|
||||
|
||||
# ---- 1. Anmeldung mit der PID --------------------------------------------
|
||||
pruefe(not gemeinsam.journal_laeuft(gemeinsam_ordner),
|
||||
"vor der Anmeldung gilt das Journal als geschlossen")
|
||||
journal._anmelden()
|
||||
pruefe(gemeinsam.journal_laeuft(gemeinsam_ordner),
|
||||
"nach der Anmeldung sieht man an der PID, dass es laeuft")
|
||||
pruefe(os.listdir(gemeinsam_ordner) == [gemeinsam.DATEINAME],
|
||||
f"im gemeinsamen Ordner liegt genau EINE Datei ({os.listdir(gemeinsam_ordner)})")
|
||||
pruefe(gemeinsam.prozess_laeuft(os.getpid(), os.path.basename(sys.executable)),
|
||||
"die PID-Pruefung erkennt den eigenen Prozess")
|
||||
pruefe(not gemeinsam.prozess_laeuft(999999, ""),
|
||||
"eine Nummer, die es nicht gibt, laeuft nicht")
|
||||
pruefe(not gemeinsam.prozess_laeuft(os.getpid(), "gibtsnicht.exe"),
|
||||
"und ein fremder EXE-Name zaehlt nicht (PIDs werden wiederverwendet)")
|
||||
|
||||
# ---- Das Rechnungstool ----------------------------------------------------
|
||||
sys.path.insert(0, TOOL)
|
||||
import config # noqa: E402
|
||||
|
||||
shutil.copy(os.path.join(TOOL, "config.json"), os.path.join(tmp, "config.json"))
|
||||
config.CONFIG_PATH = os.path.join(tmp, "config.json")
|
||||
roh = json.load(open(config.CONFIG_PATH, encoding="utf-8"))
|
||||
roh["rechnung_jahr"], roh["rechnung_zaehler"] = 2026, 0
|
||||
roh["output_dir"] = rechnungen
|
||||
roh["gemeinsam_ordner"] = gemeinsam_ordner
|
||||
json.dump(roh, open(config.CONFIG_PATH, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
|
||||
|
||||
import app as A # noqa: E402
|
||||
|
||||
antworten = {"Fertig": False, "endgültig erstellen": True, "Hausnummer": True,
|
||||
"Storno wirklich": True, "Storno verwerfen": True}
|
||||
|
||||
|
||||
def askyesno(titel, text, **kw):
|
||||
for schlue, wert in antworten.items():
|
||||
if schlue in titel:
|
||||
return wert
|
||||
raise AssertionError(f"unangemeldeter Dialog: {titel!r}")
|
||||
|
||||
|
||||
messagebox.askyesno = askyesno
|
||||
|
||||
tool = A.RechnungsApp()
|
||||
tool.cfg["output_dir"] = rechnungen
|
||||
tool.cfg["gemeinsam_ordner"] = gemeinsam_ordner
|
||||
tool._buch_oeffnen()
|
||||
|
||||
# ---- 2. Das Rechnungstool sieht: das Journal ist da -----------------------
|
||||
pruefe(gemeinsam.journal_laeuft(gemeinsam_ordner),
|
||||
"das Rechnungstool sieht an der PID, dass das Journal offen ist")
|
||||
|
||||
# ---- 3. Rechnung schreiben -> Nachricht in den Briefkasten ----------------
|
||||
journal._wache_jetzt() # Stand merken, Briefkasten ist noch leer
|
||||
tool.k_vorname.set("Anna")
|
||||
tool.k_name.set("Meldung")
|
||||
tool.k_strasse.set("Musterweg 1")
|
||||
tool.k_plzort.set("12345 Musterstadt")
|
||||
tool.r_anreise.set("01.08.2026")
|
||||
tool.r_abreise.set("04.08.2026")
|
||||
tool.anzahl_vars[sorted(tool.anzahl_vars)[0]].set("3")
|
||||
tool.update()
|
||||
tool.pdf_erstellen()
|
||||
|
||||
texte, _ = gemeinsam.Nummernbuch(gemeinsam_ordner).nachrichten_ab(0)
|
||||
pruefe(any("Rechnung 2026-001 erstellt" in t for t in texte),
|
||||
f"die Meldung steht in der SQLite ({texte})")
|
||||
|
||||
# ---- 4. Das Journal liest von selbst nach --------------------------------
|
||||
pruefe(journal._wache_jetzt() is True, "die Wache sieht die Meldung")
|
||||
journal.update()
|
||||
buchungen = journal.journal.buchungen(jahr=2026)
|
||||
pruefe(len(buchungen) == 1 and buchungen[0].nachname == "Meldung",
|
||||
f"die Rechnung steht im Journal ({[(b.voll_nummer, b.nachname) for b in buchungen]})")
|
||||
pruefe("Rechnungstool meldet" in journal.status.cget("text"),
|
||||
f"und es steht in der Statuszeile ({journal.status.cget('text')!r})")
|
||||
# Gebucht wird nach RECHNUNGSDATUM (heute), nicht nach Anreise.
|
||||
import datetime as _dt
|
||||
monat = _dt.date.today().month
|
||||
zeilen = journal.monate[monat].baum.get_children()
|
||||
pruefe(len(zeilen) == 1, f"das Monatsblatt zeigt sie ({len(zeilen)} im Monat {monat})")
|
||||
|
||||
# ---- 5. Ohne neue Meldung passiert nichts --------------------------------
|
||||
pruefe(journal._wache_jetzt() is False, "ohne neue Meldung wird nicht eingelesen")
|
||||
|
||||
# ---- 6. Auch der Storno meldet sich --------------------------------------
|
||||
import bestand as BE # noqa: E402
|
||||
|
||||
offen = [m for m in BE.rechnungen_im_ordner(rechnungen) if m.get("art") != "storno"]
|
||||
tool._storno_anlegen(offen[0], rechnungen)
|
||||
texte, _ = gemeinsam.Nummernbuch(gemeinsam_ordner).nachrichten_ab(0)
|
||||
pruefe(any("Storno" in t and "2026-001" in t for t in texte),
|
||||
f"der Storno meldet sich ebenfalls ({texte})")
|
||||
pruefe(journal._wache_jetzt() is True, "und das Journal liest ihn ein")
|
||||
storno = [b for b in journal.journal.buchungen(jahr=2026) if b.storno_zu]
|
||||
pruefe(len(storno) == 1 and storno[0].storno_zu == "2026-001",
|
||||
f"der Storno steht mit seinem Bezug im Journal ({[(b.voll_nummer, b.storno_zu) for b in storno]})")
|
||||
|
||||
# ---- 7. Ist das Journal zu, wird nicht gemeldet --------------------------
|
||||
journal._abmelden()
|
||||
pruefe(not gemeinsam.journal_laeuft(gemeinsam_ordner),
|
||||
"nach dem Abmelden gilt das Journal als geschlossen")
|
||||
vorher = gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht()
|
||||
tool.reset_formular()
|
||||
tool.k_vorname.set("Bernd")
|
||||
tool.k_name.set("Ohne")
|
||||
tool.k_strasse.set("Musterweg 2")
|
||||
tool.k_plzort.set("12345 Musterstadt")
|
||||
tool.r_anreise.set("10.08.2026")
|
||||
tool.r_abreise.set("12.08.2026")
|
||||
tool.anzahl_vars[sorted(tool.anzahl_vars)[0]].set("2")
|
||||
tool.update()
|
||||
tool.pdf_erstellen()
|
||||
pruefe(gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht() == vorher,
|
||||
"dann kommt keine Meldung dazu")
|
||||
pruefe(len([d for d in os.listdir(rechnungen) if "Ohne" in d]) == 1,
|
||||
"die Rechnung wird trotzdem geschrieben")
|
||||
|
||||
# ---- 8. Eine Karteileiche zaehlt nicht und wird weggeraeumt -------------
|
||||
buch = gemeinsam.Nummernbuch(gemeinsam_ordner)
|
||||
buch.anmelden("journal", 999999, "Beherbergungssteuer.exe")
|
||||
pruefe(not gemeinsam.journal_laeuft(gemeinsam_ordner),
|
||||
"eine tote PID (abgestuerzt) gilt nicht als laufend")
|
||||
pruefe(not buch.laeuft("journal"), "und die Karteileiche ist gleich weggeraeumt")
|
||||
|
||||
# ---- 9. Die Wache darf nie sterben --------------------------------------
|
||||
echt = gemeinsam.datei_stand
|
||||
gemeinsam.datei_stand = lambda o: (_ for _ in ()).throw(OSError("Laufwerk weg"))
|
||||
try:
|
||||
journal._wache()
|
||||
gestorben = False
|
||||
except Exception: # noqa: BLE001
|
||||
gestorben = True
|
||||
gemeinsam.datei_stand = echt
|
||||
pruefe(not gestorben, "ein Fehler im Durchgang wirft die Wache nicht um")
|
||||
|
||||
# ---- 10. Ohne gemeinsamen Ordner bleibt alles ruhig ---------------------
|
||||
journal.buch = None
|
||||
pruefe(journal._wache_jetzt() is False, "ohne gemeinsames Buch tut die Wache nichts")
|
||||
|
||||
tool.destroy()
|
||||
journal.journal.schliessen()
|
||||
journal.destroy()
|
||||
if not fehler:
|
||||
shutil.rmtree(tmp, ignore_errors=True)
|
||||
else:
|
||||
print(f" (Wegwerf-Ordner bleibt: {tmp})")
|
||||
print("---")
|
||||
print("alles gruen" if not fehler else f"{len(fehler)} Fehler: {fehler}")
|
||||
sys.exit(1 if fehler else 0)
|
||||
Loading…
Add table
Add a link
Reference in a new issue