Steuerrechnungstool/pruef_einzelinstanz.py

# -*- coding: utf-8 -*-
"""Pruefstand: geht das Steuerjournal wirklich nur EINMAL auf?

Zwei Fenster lesen denselben Rechnungsordner in dieselbe journal.sqlite3. Was
das eine loescht, steht im anderen noch auf dem Schirm - und wer dort auf
"Bericht fuers Amt" drueckt, meldet einen Stand, den es nicht mehr gibt.

Geprueft wird mit ECHTEN Prozessen, nicht mit einem nachgebauten Aufruf: nur so
zeigt sich, ob die Sperre wirklich vor dem Datenordner greift und ob sie sich
beim Beenden auch wieder loest. Das Programm laeuft dabei aus einer Kopie im
Temp-Ordner - die Sperre gilt trotzdem (sie haengt am Namen, nicht am Ordner),
das echte Journal wird nicht angefasst.

Waehrend des Laufs blitzt kurz ein Fenster auf. Das gehoert dazu.
"""
import os
import shutil
import subprocess
import sys
import tempfile
import threading
import time

HIER = r"C:\claude\beherbergungssteuer"
sys.path.insert(0, HIER)

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)


laeufer = []


def _reissleine():
    for p in laeufer:
        try:
            p.kill()
        except Exception:  # noqa: BLE001
            pass
    print("ABBRUCH | Pruefstand haengt (Zeitgrenze) - vermutlich ein Dialog, "
          "auf den niemand klickt")
    os._exit(2)


_wecker = threading.Timer(150, _reissleine)
_wecker.daemon = True
_wecker.start()

from einzelinstanz import pid_datei  # noqa: E402

if os.path.exists(pid_datei()):
    print(f"        (Merkzettel: {pid_datei()})")

# ---- Kopie des Programms, damit das echte Journal unberuehrt bleibt --------
tmp = tempfile.mkdtemp(prefix="bst_instanz_")
prog = os.path.join(tmp, "programm")
os.makedirs(prog)
for name in os.listdir(HIER):
    if name.endswith(".py"):
        shutil.copy(os.path.join(HIER, name), os.path.join(prog, name))


def starte():
    # --test: kein Erststart-Dialog nach dem gemeinsamen Ordner. Die Sperre
    # selbst greift unabhaengig davon - genau die wird hier geprueft.
    return subprocess.Popen([sys.executable, "app.py", "--test"], cwd=prog,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)


def warte_auf_ende(p, sekunden):
    """Wartet, bis der Prozess von selbst geht. None = er laeuft noch."""
    ende = time.time() + sekunden
    while time.time() < ende:
        if p.poll() is not None:
            return p.returncode
        time.sleep(0.2)
    return None


# ---- 1. Das erste Fenster ------------------------------------------------
erst = starte()
laeufer.append(erst)
time.sleep(6)          # Tk braucht einen Moment, bis das Fenster wirklich steht
pruefe(erst.poll() is None, f"das erste Fenster laeuft (Rueckgabe {erst.poll()})")

merkzettel = None
try:
    with open(pid_datei(), encoding="ascii") as f:
        merkzettel = int(f.read().strip())
except (OSError, ValueError):
    pass
pruefe(merkzettel == erst.pid,
       f"der Merkzettel nennt genau diesen Prozess ({merkzettel} / {erst.pid})")

# ---- 2. Der zweite Doppelklick -------------------------------------------
zweit = starte()
laeufer.append(zweit)
code = warte_auf_ende(zweit, 25)
if code is None:
    zweit.kill()
pruefe(code == 0, f"der zweite Start beendet sich von selbst (Rueckgabe {code})")
pruefe(erst.poll() is None, "und das erste Fenster laeuft dabei weiter")

ausgabe = ""
try:
    ausgabe = (zweit.stdout.read() or b"").decode("utf-8", "replace")
except Exception:  # noqa: BLE001
    pass
pruefe("Traceback" not in ausgabe, f"ohne Absturz im zweiten Prozess ({ausgabe[-200:]!r})")
# Die Sperre steht VOR dem Datenordner: der zweite Prozess darf gar nicht erst
# so weit kommen, dass er das Journal oeffnet.
pruefe("Journal" not in ausgabe,
       f"der zweite Prozess fasst das Journal nicht an ({ausgabe[-200:]!r})")

# ---- 3. Sperre loest sich beim Beenden ------------------------------------
erst.terminate()
warte_auf_ende(erst, 15)
pruefe(erst.poll() is not None, "das erste Fenster laesst sich beenden")
time.sleep(1.5)

dritt = starte()
laeufer.append(dritt)
code = warte_auf_ende(dritt, 8)
pruefe(code is None, f"danach geht das Programm wieder auf (Rueckgabe {code})")
dritt.terminate()
warte_auf_ende(dritt, 10)
if dritt.poll() is None:
    dritt.kill()

for p in laeufer:
    if p.poll() is None:
        p.kill()
    try:
        p.stdout.close()
    except Exception:  # noqa: BLE001
        pass

if not fehler:
    shutil.rmtree(tmp, ignore_errors=True)
else:
    print(f"        (Kopie bleibt zum Nachsehen: {tmp})")
print("---")
print("alles gruen" if not fehler else f"{len(fehler)} Fehler: {fehler}")
sys.exit(1 if fehler else 0)