Steuerrechnungstool/pruef_wache.py

# -*- 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()
# 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

# 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

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()
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)