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)


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
from pruefhelfer import fehler, pruefe

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


# Auch die Beilagen mitnehmen - ohne assets/ faende das Programm sein Logo
# nicht. Das stuerzt zwar nicht ab, aber ein Pruefstand soll das Programm
# fahren, wie es ausgeliefert wird, nicht eine abgespeckte Fassung.
for name in ("assets",):
    quelle = os.path.join(HIER, name)
    if os.path.isdir(quelle):
        shutil.copytree(quelle, os.path.join(prog, name))


# Der Erprobungshinweis wuerde in JEDEM dieser drei Starts ein echtes Fenster
# oeffnen. Hier laeuft das Programm als eigener Prozess - da laesst sich kein
# Dialog abfangen, es klickt niemand, und der Pruefstand liefe in die
# Reissleine. Geprueft wird an dieser Stelle die SPERRE; der Hinweis hat seinen
# eigenen Pruefstand (pruef_hinweis.py).
#
# Also wird die Kenntnisnahme vorher eingetragen - mit genau dem Code, den auch
# der Klick auf "Ja" ausfuehrt (hinweis.vermerken auf die Einstellungstabelle).
# Nichts nachgebaut, nichts von Hand ins SQL geschrieben.
#
# Das trifft NUR die Kopie im Temp-Ordner: aus der Quelle heraus liegt der
# Datenordner neben app.py (ablage.daten_ordner()). Im echten Journal des
# Betriebs entsteht dadurch keine Zustimmung, die niemand gegeben hat.
def hinweis_vorab_eintragen():
    import db
    import hinweis
    journal = db.Journal(os.path.join(prog, "journal.sqlite3"))

    class Ablage:
        def get(self, schluessel, standard=""):
            return journal.hole(schluessel, standard)

        def __setitem__(self, schluessel, wert):
            journal.setze(schluessel, str(wert))

    hinweis.vermerken(Ablage())
    print(f"        (Kenntnisnahme in der Kopie vermerkt, Fassung {hinweis.FASSUNG})")


hinweis_vorab_eintragen()


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)