Steuerrechnungstool/pruef_wache.py
TheMockTv 9d7aa1eec7 Darstellung: das Programm hatte ueberhaupt kein Theme
Sein Befund
  "und beim steuertool fehlt das theam in genze oder wird nicht durch die
   automatik gesetzt"
  "nur die einstellunden hilfe leiste die ist noch hell und der icon und
   titel leiste auch"

  Er hatte recht, und zwar wortwoertlich: dieses Programm lief im
  Tk-Standard, mit dreimal fest eingetragenem foreground="black". Es gab hier
  gar keine theme.py.

Gebaut
  theme.py aus dem Rechnungstool uebernommen und dabei von config.py geloest
  - eine Datei, die in BEIDEN Programmen liegt, darf nichts aus einem der
  beiden importieren. Beim ersten Uebernehmen flog sie genau daran beim
  Import auseinander.

  Die Einstellungen liegen hier in der Journal-Tabelle, nicht in einer
  config.json. _einstellungen() ist die Bruecke: etwas mit .get() und []=,
  wie hinweis.py und theme.py es erwarten.

  Menueleiste aus menueleiste.py statt tk.Menu - ein natives Menue laesst
  sich unter Windows nicht einfaerben, das war der letzte helle Rest.
  Treeview-Stil in theme.py ergaenzt (ttk zeichnet Listen nicht ueber den
  Grundstil), Reiter schmaler, damit alle zwoelf Monatsnamen lesbar bleiben.
  Die Summenleisten in gui_monat.py holen ihre Farben jetzt vom Fenster statt
  sie fest zu tragen - sie waren ein weisser Block unten im dunklen Bild.

  Logo und Fenstersymbol werden fuer die dunkle Darstellung aufgehellt.

⚠️ Altfehler behoben: die Meldung des Rechnungstools stand nie
  pruef_wache lief mal gruen und mal rot - je nachdem, wann er hinsah. Das
  war kein Wackler im Pruefstand, sondern sein Symptom: meldet das
  Rechnungstool eine neue Rechnung, laeuft sofort das Einlesen an, und dessen
  Ergebnis landete in derselben Zeile. Der Benutzer sah NIE, dass etwas
  hereingekommen war. melde() kennt jetzt einen Vorrang in Sekunden; die
  Meldung des Rechnungstools bekommt sechs. Dreimal hintereinander stabil
  gruen.

  Ein wackelnder Pruefstand ist keine Nebensache, die man wegdrueckt.

AUFGABE_STEUERSAETZE_DYNAMISCH.md
  Seine naechste Ansage notiert: die festen Saetze sollen raus, dynamisch per
  Tabelle wie im Rechnungstool. Mit dem, worauf dabei zu achten ist - ein
  Satz gilt ab einem DATUM, und die Saetze bereits erfasster Buchungen
  bleiben stehen (§ 147 AO).

Fassung 1.9.0.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V4t48uxDok5rJXhC9bX1Ax
2026-09-07 00:34:38 +02:00

294 lines
12 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
# Der Abfang MUSS vor dem ersten Fenster stehen: sonst geht ein echter Dialog
# auf und der Lauf wartet auf einen Klick, den niemand macht. Genau das ist am
# 06.09.2026 passiert - der Hinweis kam, bevor der Abfang gesetzt war.
# Erprobungshinweis und Freigabe vor dem Beleg werden hier bejaht - geprueft
# werden sie eigens in pruef_hinweis.py und pruef_freigabe.py des Rechnungstools.
antworten = {"Fertig": False, "endgültig erstellen": True, "Hausnummer": True,
"Storno wirklich": True, "Storno verwerfen": True}
antworten["Erprobungsfassung"] = True
antworten["jetzt erstellen?"] = 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
journal = JA.App()
# Der Erprobungshinweis haengt an after(0) - ohne Ereignisschleife kaeme er nie
# dran, und damit auch der verzoegerte Start nicht. update() treibt die Schleife
# einmal an, genau wie es mainloop() taete. Abgekuerzt wird nichts.
journal.update()
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")
# Die Absicht dieser Pruefung: im gemeinsamen Ordner liegt NICHTS Fremdes -
# keine Herzschlagdateien, keine Zwischenstaende. Erlaubt sind genau zwei:
# das Nummernbuch und der Schluessel der Mini-API, der seit dem Umbau vom
# 05.09.2026 dorthin gehoert (beide Programme muessen ihn lesen koennen).
# Alles darueber hinaus faellt weiterhin auf.
ERLAUBT = {gemeinsam.DATEINAME, "api_schluessel.txt"}
_liegt = set(os.listdir(gemeinsam_ordner))
pruefe(_liegt <= ERLAUBT,
f"im gemeinsamen Ordner liegt nichts Fremdes ({sorted(_liegt)})")
# ---- 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
tool = A.RechnungsApp()
tool.update() # dasselbe fuer das Rechnungstool
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()
# Auf die Meldung WARTEN statt einmal hinzusehen. Das Einlesen laeuft
# nebenher und schreibt danach sein eigenes Ergebnis in dieselbe Zeile - wer
# genau dazwischen misst, sieht "Nummernbuch: 1 Nummern eingetragen" und
# meldet einen Fehler, den es nicht gibt. Genau das ist am 07.09.2026 einmal
# in einer Prueflauf-Runde passiert und beim naechsten Lauf nicht mehr.
for _ in range(60):
if "Rechnungstool meldet" in journal.status.cget("text"):
break
journal.update()
time.sleep(0.05)
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)