dbe95e1ec6
Das Journal macht beim Start einen winzigen Server auf 127.0.0.1 auf und traegt
Port und PID in die gemeinsame SQLite ein (Tabelle "laeuft"). Das Rechnungstool
sieht daran, dass jemand zuhoert, und klopft nach jeder Rechnung an: "schau in
die sqlite". Dann wird sofort nachgelesen, statt bis zum naechsten Durchgang zu
warten. Faellt die API aus, findet die Wache die Meldung trotzdem - ein os.stat
auf die eine Datei, alle zwei Sekunden und sofort beim Zurueckklicken.
Die Proforma-Rechnung wird NICHT gebucht: keine_buchung() kennt jetzt
Berichtigung und Proforma. Der Scan fragt wieder keine_buchung() - seit dem
Berichtigungs-Umbau tat er das nicht mehr, und die Proforma waere als Buchung
im Journal gelandet (der Pruefstand hat es gefunden).
Pruefstand pruef_wache.py: 31 Pruefungen ueber BEIDE Programme - Anmeldung mit
PID, ping ("hallo, bin noch da"), genau eine Datei im gemeinsamen Ordner, Push
nach Rechnung und nach Storno, Einlesen samt Anzeige, Rueckfallebene ohne API,
tote PID wird weggeraeumt, kein Melden an ein geschlossenes Journal.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GYeAfLtccFrbU3MTj1MMTx
264 lines
10 KiB
Python
264 lines
10 KiB
Python
# -*- 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)})")
|
|
|
|
# ---- 1b. Die Mini-API: Port im Telefonbuch, und sie meldet sich -----------
|
|
import api # noqa: E402
|
|
|
|
eintrag = gemeinsam.Nummernbuch(gemeinsam_ordner).wer_laeuft("journal")
|
|
pruefe(bool(eintrag and eintrag.get("port")),
|
|
f"das Journal hat Port und PID im Telefonbuch ({eintrag})")
|
|
pruefe(journal.api.port > 0, f"sein kleiner Server laeuft ({journal.api.port})")
|
|
pruefe(api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "journal"),
|
|
"auf ein ping antwortet es: hallo, bin noch da")
|
|
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. Der PUSH kommt an: das Journal liest ohne Wache nach -------------
|
|
# Nicht _wache_jetzt() aufrufen - genau das soll der Push von selbst ausloesen.
|
|
for _ in range(40):
|
|
if not journal._push_kasten.empty():
|
|
break
|
|
time.sleep(0.05)
|
|
pruefe(not journal._push_kasten.empty(),
|
|
"der Push des Rechnungstools ist im Fenster angekommen")
|
|
journal._push_abholen()
|
|
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")
|
|
|
|
# ---- 5b. Gegenseitig: auch das Rechnungstool ist erreichbar --------------
|
|
pruefe(tool.api.port > 0, f"das Rechnungstool hat auch einen kleinen Server ({tool.api.port})")
|
|
pruefe(api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "rechnungstool"),
|
|
"und meldet sich ebenfalls mit: hallo, bin noch da")
|
|
pruefe(api.anklopfen(gemeinsam.Nummernbuch(gemeinsam_ordner), "rechnungstool"),
|
|
"beim Rechnungstool laesst sich anklopfen (die beiden sehen sich gegenseitig)")
|
|
|
|
# ---- 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]})")
|
|
|
|
# ---- 6b. Ist die API weg, findet die Meldung trotzdem ihren Weg ---------
|
|
journal.api.stoppen() # nur den Server zumachen, Eintrag bleibt stehen
|
|
pruefe(not api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "journal"),
|
|
"ohne Server antwortet niemand mehr auf das ping")
|
|
vorher_id = gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht()
|
|
tool._journal_benachrichtigen("Probe ohne API")
|
|
pruefe(gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht() > vorher_id,
|
|
"die Meldung steht trotzdem in der SQLite (Rueckfallebene)")
|
|
pruefe(journal._wache_jetzt() is True, "und die Wache findet sie beim naechsten Nachsehen")
|
|
|
|
# ---- 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)
|