Automatischer Push: 8 Datei(en) geändert
- .vault-tools/trading-plattform-pruefung/LIESMICH.md - .vault-tools/trading-plattform-pruefung/abbau_alpaca_lesen.py - .vault-tools/trading-plattform-pruefung/abbau_glattstellen.sh - .vault-tools/trading-plattform-pruefung/abbau_ibkr_lesen.sh - .vault-tools/trading-plattform-pruefung/abbau_restbestand.py - .vault-tools/trading-plattform-pruefung/abbau_restbestand_angriffe.py - .vault-tools/trading-plattform-pruefung/abbau_restbestand_einrichten.ps1 - AI/Regelwerk/windows-shell.md Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5 committed
1 parent
30ed04a643
commit
3784322a76
8 files changed
+1687
-3
No files matched your search
@@ -0,0 +1,817 @@
|
||||
"""Angriffsfaelle fuer den Abbau des Restbestands (Paket AB): abbau_restbestand.py und seine WSL-Teile.
|
||||
|
||||
Ersteller: Pascal Michel
|
||||
Erstellt: 2026-10-07
|
||||
Doku: LIESMICH.md in diesem Ordner, Abschnitt "Paket AB"
|
||||
|
||||
Koordinator-Werkzeug, nicht fuers Repo. Jeder Fall ist eine Funktion test_<was er angreift>; so laufen
|
||||
dieselben Faelle als Skript und unter pytest (mutanten_lauf.py mit mutanten.toml im Planordner).
|
||||
Abdichtung, doppelt: Alle Pfade nach draussen zeigen unten ins Leere (wsl.exe, die WSL-Skripte, die
|
||||
Instanzordner), und jeder Fall ersetzt wsl() durch die Attrappe Welt, die jeden Aufruf zaehlt und nichts
|
||||
startet; der Fall zu wsl() selbst ersetzt starte (subprocess.run). Kein Fall erreicht wsl.exe, das Netz
|
||||
oder ntfy. Die Attrappe lehnt wie CPython jedes Argument mit NUL ab.
|
||||
Die Faelle test_wsl_* fahren bash und die Venvs in WSL gegen Wegwerf-Instanzordner mit einem falschen
|
||||
scripts/killswitch.py, dem erfundenen Wort TESTWORT-NICHT-ECHT, der erfundenen Kontonummer DUQ123456 und
|
||||
erfundenen Betraegen, nie gegen einen echten Instanzordner. Unter Windows sind sie uebersprungen.
|
||||
Aufruf unter Windows: python -I -X utf8 abbau_restbestand_angriffe.py
|
||||
Aufruf in WSL: /home/claude/venvs/trading/bin/python -I abbau_restbestand_angriffe.py
|
||||
Exit 0, wenn jeder Fall wie erwartet ausgeht.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from collections import namedtuple
|
||||
from datetime import datetime
|
||||
from datetime import timedelta
|
||||
from datetime import timezone
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
HIER = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HIER))
|
||||
import abbau_restbestand as ab # noqa: E402
|
||||
|
||||
# Abdichtung (Vorfall 2026-09-26: ein Mutant schickte ueber den absoluten WSL_SKRIPT Meldungen ans
|
||||
# Telefon). Jeder Pfad nach draussen zeigt ins Leere, bevor ein Fall laeuft.
|
||||
ab.WSL = str(Path(tempfile.gettempdir()) / "abbau-angriffe-kein-wsl" / "wsl.exe")
|
||||
ab.WSL_SKRIPT = "/nicht-vorhanden/abendkontrolle_wsl.py"
|
||||
ab.ALPACA_LESEN = "/nicht-vorhanden/abbau_alpaca_lesen.py"
|
||||
ab.IBKR_LESEN = "/nicht-vorhanden/abbau_ibkr_lesen.sh"
|
||||
ab.GLATTSTELLEN = "/nicht-vorhanden/abbau_glattstellen.sh"
|
||||
ab.ORDNER_WSL = {"alpaca": "/nicht-vorhanden/alpaca", "ibkr": "/nicht-vorhanden/ibkr"}
|
||||
|
||||
WRAPPER = HIER / "abbau_glattstellen.sh"
|
||||
IBKR_SKRIPT = HIER / "abbau_ibkr_lesen.sh"
|
||||
NUL = chr(0)
|
||||
TESTWORT = "TESTWORT-NICHT-ECHT"
|
||||
KONTO = "DUQ123456"
|
||||
BETRAG = "12345.67"
|
||||
BETRAG_OHNE_GELDWORT = "98.76"
|
||||
JETZT = "2026-10-08T15:45:00"
|
||||
GRUND = (
|
||||
"Abbau Restbestand nach Netzschalter-Abriss 2026-10-07; Auftrag Pascal 2026-10-07, "
|
||||
"ausgefuehrt von der Aufgabe Abbau Restbestand"
|
||||
)
|
||||
LOG = "projekt/data/paper/abbau/abbau.log"
|
||||
ERGEBNIS = "projekt/data/paper/abbau/abbau.json"
|
||||
|
||||
OFFEN = {"offen": True, "sekunden_bis_schluss": 6 * 3600}
|
||||
ZU = {"offen": False, "sekunden_bis_schluss": 18 * 3600}
|
||||
FLACH = {"positionen": [], "orders": 0}
|
||||
AAPL10 = {"positionen": [("AAPL", 10)], "orders": 0}
|
||||
IBKR7 = {
|
||||
"positionen": [
|
||||
("AAPL.XNAS", 10), ("MSFT.XNAS", 10), ("NVDA.XNAS", 20), ("JPM.XNYS", 10),
|
||||
("CAT.XNYS", 10), ("KO.XNYS", 20), ("NEE.XNYS", 20),
|
||||
],
|
||||
"orders": 0,
|
||||
}
|
||||
WRAPPER_AUSGABE = (
|
||||
"Zustand vorher: angehalten\nBroker: alpaca\nErgebnis: glattstellen, erfolgreich=True\n"
|
||||
"Zustand nachher: gesperrt\n"
|
||||
)
|
||||
# Ein falsches scripts/killswitch.py fuer die Faelle test_wsl_*: schreibt seine Argumente (mit dem Wort)
|
||||
# in den Wegwerf-Ordner, druckt das Wort auf stderr und eine erfundene Kontonummer und erfundene Betraege
|
||||
# auf stdout. Nichts davon darf aus dem Wrapper herauskommen.
|
||||
FALSCHER_KILLSWITCH = """import sys
|
||||
from pathlib import Path
|
||||
Path("aufgerufen").write_text(" ".join(sys.argv[1:]), encoding="utf-8")
|
||||
print("Zustand vorher: aktiv")
|
||||
print("Interpreter: " + sys.executable)
|
||||
print("Argumente: " + " ".join(sys.argv[1:]))
|
||||
print("Ergebnis: glattstellen, erfolgreich=False")
|
||||
print(" Fehler: Kontowert 12345.67 USD nicht gedeckt")
|
||||
print(" Fehler: Order an DUQ123456 abgelehnt, limit 98.76")
|
||||
print("Zustand nachher: gesperrt", flush=True)
|
||||
print("Abgelehnt, es wurde nichts geaendert: " + sys.argv[-1], file=sys.stderr)
|
||||
sys.exit(int(Path("exit.txt").read_text(encoding="utf-8")))
|
||||
"""
|
||||
|
||||
Roh = namedtuple("Roh", "exit text")
|
||||
|
||||
|
||||
class Uebersprungen(Exception):
|
||||
"""Ein Fall, der bash und die Venvs in WSL braucht, unter Windows."""
|
||||
|
||||
|
||||
def nur_in_wsl() -> None:
|
||||
if sys.platform == "win32":
|
||||
raise Uebersprungen("nur in WSL")
|
||||
|
||||
|
||||
def alpaca_json(uhr, bestand, fehler="BrokerFehler: Test") -> str:
|
||||
"""Die JSON-Zeile, die abbau_alpaca_lesen.py drucken wuerde; fehler ist der bestand_fehler."""
|
||||
daten = {"uhr": uhr, "uhr_fehler": None, "positionen": None, "orders": None, "bestand_fehler": None}
|
||||
if uhr is None:
|
||||
daten["uhr_fehler"] = "BrokerFehler: Marktuhr nicht lesbar"
|
||||
if bestand is None:
|
||||
daten["bestand_fehler"] = fehler
|
||||
else:
|
||||
daten["positionen"] = [{"symbol": symbol, "menge": menge} for symbol, menge in bestand["positionen"]]
|
||||
daten["orders"] = bestand["orders"]
|
||||
return json.dumps(daten) + "\n"
|
||||
|
||||
|
||||
def ibkr_text(bestand, schluss="ERGEBNIS: gelesen") -> str:
|
||||
"""stdout von scripts/ibkr_pruefen.py lesen in der gemessenen Form, ohne Kontonummer."""
|
||||
zeilen = [
|
||||
"verbunden: Gateway 127.0.0.1:4004, Client-ID 12, Handelsmodus paper, Konto XX (8 Zeichen), bestaetigt",
|
||||
"Konto: Kontowert, Bargeld und Kaufkraft gelesen (Betraege werden nicht ausgegeben)",
|
||||
f"Positionen: {len(bestand['positionen'])}",
|
||||
]
|
||||
zeilen += [f" {kennung} {menge}" for kennung, menge in bestand["positionen"]]
|
||||
zeilen += [f"Offene Orders: {bestand['orders']}", schluss]
|
||||
return "\n".join(zeilen) + "\n"
|
||||
|
||||
|
||||
def unit_text(stand) -> str:
|
||||
"""Ausgabe von systemctl show --property=LoadState,ActiveState: Text als Roh, (LoadState, ActiveState)
|
||||
als Paar, oder nur ActiveState einer geladenen Unit."""
|
||||
if isinstance(stand, Roh):
|
||||
return stand.text
|
||||
laden, aktiv = stand if isinstance(stand, tuple) else ("loaded", stand)
|
||||
return f"LoadState={laden}\nActiveState={aktiv}\n"
|
||||
|
||||
|
||||
class Welt:
|
||||
"""Attrappe fuer alles hinter ab.wsl(): Units, Bestaende, Wrapper, Meldung. Zaehlt jeden Aufruf."""
|
||||
|
||||
def __init__(self, *, units=("inactive", "inactive"), uhr=OFFEN, alpaca=AAPL10, ibkr=IBKR7,
|
||||
alpaca_nachher=FLACH, ibkr_nachher=FLACH, alpaca_fehler="BrokerFehler: Test",
|
||||
wrapper_exit=0, wrapper_wirft=False, echter_wrapper=None):
|
||||
self.units = dict(zip(ab.INSTANZEN, units))
|
||||
self.uhr = uhr
|
||||
self.bestand = {"alpaca": alpaca, "ibkr": ibkr}
|
||||
self.nachher = {"alpaca": alpaca_nachher, "ibkr": ibkr_nachher}
|
||||
self.alpaca_fehler = alpaca_fehler
|
||||
self.wrapper_exit = wrapper_exit
|
||||
self.wrapper_wirft = wrapper_wirft
|
||||
self.echter_wrapper = echter_wrapper
|
||||
self.aufrufe = []
|
||||
self.glatt = []
|
||||
self.ordner = []
|
||||
self.meldungen = []
|
||||
self.schlaf = []
|
||||
self.gelesen = {"alpaca": 0, "ibkr": 0}
|
||||
|
||||
def schlafe(self, sekunden):
|
||||
self.schlaf.append(sekunden)
|
||||
|
||||
def __call__(self, argumente, timeout):
|
||||
argumente = list(argumente)
|
||||
if any(NUL in argument for argument in argumente):
|
||||
raise ValueError("embedded null character")
|
||||
self.aufrufe.append(argumente)
|
||||
if argumente[:2] == ["systemctl", "show"]:
|
||||
instanz = next(name for name, unit in ab.UNITS.items() if unit == argumente[-1])
|
||||
return 0, unit_text(self.units[instanz]), ""
|
||||
if argumente[:2] == [ab.ALPACA_PY, ab.ALPACA_LESEN]:
|
||||
self.gelesen["alpaca"] += 1
|
||||
return 0, alpaca_json(self.uhr, self.bestand["alpaca"], self.alpaca_fehler), ""
|
||||
if argumente[:2] == ["bash", ab.IBKR_LESEN]:
|
||||
self.gelesen["ibkr"] += 1
|
||||
stand = self.bestand["ibkr"]
|
||||
if stand is None:
|
||||
return 4, "", "Voraussetzung fehlt: Gateway nicht erreichbar\n"
|
||||
if isinstance(stand, Roh):
|
||||
return stand.exit, stand.text, ""
|
||||
return 0, ibkr_text(stand), ""
|
||||
if argumente[:2] == ["bash", ab.GLATTSTELLEN]:
|
||||
instanz = argumente[2]
|
||||
self.glatt.append(instanz)
|
||||
self.ordner.append(argumente[3])
|
||||
if self.wrapper_wirft:
|
||||
raise subprocess.TimeoutExpired(argumente, timeout)
|
||||
self.bestand[instanz] = self.nachher[instanz]
|
||||
if self.echter_wrapper is not None:
|
||||
return self.echter_wrapper(argumente[2:])
|
||||
return self.wrapper_exit, WRAPPER_AUSGABE, ""
|
||||
if argumente[:3] == [ab.WSL_PY, ab.WSL_SKRIPT, "--meldung"]:
|
||||
self.meldungen.append((argumente[3], argumente[4]))
|
||||
return 0, '{"meldung": "gesendet"}\n', ""
|
||||
raise AssertionError(f"unbekannter Aufruf: {argumente!r}")
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def attrappe(welt):
|
||||
"""Ersetzt wsl() und schlafe() fuer die Dauer eines Falls und stellt beide danach zurueck."""
|
||||
alt = ab.wsl, ab.schlafe
|
||||
ab.wsl, ab.schlafe = welt, welt.schlafe
|
||||
try:
|
||||
yield welt
|
||||
finally:
|
||||
ab.wsl, ab.schlafe = alt
|
||||
|
||||
|
||||
def lies_json(text):
|
||||
try:
|
||||
return json.loads(text) if text else None
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def lauf_mit(welt, *, jetzt=JETZT, modus="scharf", ergebnis_vorher=None):
|
||||
"""Ein Lauf von ab.main() in Wegwerf-Ordnern fuer Projekt und Vault; gibt alles Sichtbare zurueck."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
projekt, vault = tmp / "projekt", tmp / "vault"
|
||||
projekt.mkdir()
|
||||
(vault / "Inbox").mkdir(parents=True)
|
||||
ablage = projekt / ab.ABLAGE_TEIL
|
||||
if ergebnis_vorher is not None:
|
||||
ablage.mkdir(parents=True)
|
||||
(ablage / ab.ERGEBNIS_NAME).write_text(ergebnis_vorher, encoding="utf-8")
|
||||
puffer = io.StringIO()
|
||||
with attrappe(welt), contextlib.redirect_stdout(puffer):
|
||||
code = ab.main(["--modus", modus, "--jetzt", jetzt, "--projekt", str(projekt), "--vault", str(vault)])
|
||||
ergebnis = ablage / ab.ERGEBNIS_NAME
|
||||
log = ablage / ab.LOG_NAME
|
||||
notizen = sorted((vault / "Inbox").glob("*.md"))
|
||||
text = ergebnis.read_text(encoding="utf-8") if ergebnis.exists() else None
|
||||
return SimpleNamespace(
|
||||
code=code,
|
||||
welt=welt,
|
||||
ausgabe=puffer.getvalue(),
|
||||
dateien=sorted(p.relative_to(tmp).as_posix() for p in tmp.rglob("*") if p.is_file()),
|
||||
ergebnis_text=text,
|
||||
json=lies_json(text),
|
||||
log=log.read_text(encoding="utf-8") if log.exists() else "",
|
||||
notiz=notizen[0].read_text(encoding="utf-8") if notizen else "",
|
||||
)
|
||||
|
||||
|
||||
def titel_von(e) -> list[str]:
|
||||
return [titel for titel, _ in e.welt.meldungen]
|
||||
|
||||
|
||||
def kurz(e) -> str:
|
||||
flach = None if e.json is None else e.json.get("flach")
|
||||
return (
|
||||
f"code={e.code} glatt={e.welt.glatt} ordner={e.welt.ordner} schlaf={e.welt.schlaf} "
|
||||
f"gelesen={e.welt.gelesen} meldungen={titel_von(e)} flach={flach} dateien={e.dateien} "
|
||||
f"ausgabe={e.ausgabe[-400:]!r}"
|
||||
)
|
||||
|
||||
|
||||
def instanzordner(pfad: Path, *, wort=TESTWORT, exit_code=0) -> Path:
|
||||
"""Ein Wegwerf-Instanzordner mit falschem scripts/killswitch.py und, wenn wort gesetzt ist, killswitch.toml."""
|
||||
(pfad / "scripts").mkdir(parents=True)
|
||||
(pfad / "config").mkdir()
|
||||
if wort is not None:
|
||||
(pfad / "config" / "killswitch.toml").write_text(
|
||||
f'[killswitch]\nbestaetigungswort = "{wort}"\n', encoding="utf-8"
|
||||
)
|
||||
(pfad / "scripts" / "killswitch.py").write_text(FALSCHER_KILLSWITCH, encoding="utf-8")
|
||||
(pfad / "exit.txt").write_text(str(exit_code), encoding="utf-8")
|
||||
return pfad
|
||||
|
||||
|
||||
def wrapper(tmp: Path, *argumente: str) -> tuple[int, str]:
|
||||
"""Der echte abbau_glattstellen.sh aus einem leeren Arbeitsordner; stdout und stderr zusammen."""
|
||||
leer = tmp / "leer"
|
||||
leer.mkdir(exist_ok=True)
|
||||
lauf = subprocess.run(
|
||||
["bash", str(WRAPPER), *argumente], cwd=leer, capture_output=True, text=True, timeout=120
|
||||
)
|
||||
return lauf.returncode, lauf.stdout + lauf.stderr
|
||||
|
||||
|
||||
class FalscherAlpaca:
|
||||
"""Ein Broker mit den drei Leseaufrufen; die Betraege sind erfunden und duerfen nie herauskommen."""
|
||||
|
||||
def __init__(self, jetzt, fehler=None):
|
||||
self.jetzt = jetzt
|
||||
self.fehler = fehler
|
||||
|
||||
def marktuhr(self):
|
||||
return SimpleNamespace(is_open=True, next_open=self.jetzt, next_close=self.jetzt + timedelta(minutes=20))
|
||||
|
||||
def positionen(self):
|
||||
if self.fehler is not None:
|
||||
raise RuntimeError(self.fehler)
|
||||
return [
|
||||
SimpleNamespace(symbol="AAPL", menge=10, seite="long", marktwert=1234.56, einstandskurs=98.76),
|
||||
SimpleNamespace(symbol="KO", menge=-20, seite="short", marktwert=-1111.11, einstandskurs=55.55),
|
||||
]
|
||||
|
||||
def offene_orders(self):
|
||||
return [SimpleNamespace(id="o-1", menge=10, limit=98.76)]
|
||||
|
||||
|
||||
# --- Bedienung und Grundlinie ---------------------------------------------------------------------
|
||||
|
||||
def test_ohne_modus_bricht_mit_exit_2_ab_und_ruft_nichts():
|
||||
welt = Welt()
|
||||
codes = []
|
||||
for argumente in (["--jetzt", JETZT], ["--modus", "probe", "--jetzt", JETZT]):
|
||||
with attrappe(welt), contextlib.redirect_stderr(io.StringIO()):
|
||||
try:
|
||||
ab.main(argumente)
|
||||
except SystemExit as ende:
|
||||
codes.append(ende.code)
|
||||
else:
|
||||
codes.append(None)
|
||||
assert codes == [2, 2], codes
|
||||
assert welt.aufrufe == [], welt.aufrufe
|
||||
|
||||
|
||||
def test_grundlinie_je_instanz_genau_ein_eingriff_dann_flach():
|
||||
e = lauf_mit(Welt())
|
||||
assert e.code == 0, kurz(e)
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
assert e.welt.ordner == [ab.ORDNER_WSL["alpaca"], ab.ORDNER_WSL["ibkr"]], kurz(e)
|
||||
assert e.welt.schlaf == [ab.WARTEN_S], kurz(e)
|
||||
assert e.welt.gelesen == {"alpaca": 2, "ibkr": 2}, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: flach"], kurz(e)
|
||||
assert e.json["flach"] is True, kurz(e)
|
||||
assert e.json["instanzen"]["alpaca"]["vorher"] == {"positionen": {"AAPL": 10}, "orders": 0}, kurz(e)
|
||||
assert e.json["instanzen"]["ibkr"]["nachher"] == {"positionen": {}, "orders": 0}, kurz(e)
|
||||
assert "Abbau Restbestand (automatisch, Auftrag Pascal 2026-10-07): flach." in e.notiz, e.notiz
|
||||
assert "AAPL long 10" in e.notiz and "NEE long 20" in e.notiz, e.notiz
|
||||
assert "#needs-review" not in e.notiz, e.notiz
|
||||
|
||||
|
||||
# --- Fenster ---------------------------------------------------------------------------------------
|
||||
|
||||
def test_vor_dem_fenster_eine_sekunde_tut_nichts():
|
||||
e = lauf_mit(Welt(), jetzt="2026-10-08T15:39:59")
|
||||
assert e.code == 0 and e.welt.aufrufe == [] and e.dateien == [], kurz(e)
|
||||
|
||||
|
||||
def test_fensterbeginn_genau_1540_greift():
|
||||
e = lauf_mit(Welt(), jetzt="2026-10-08T15:40:00")
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
|
||||
|
||||
def test_fensterende_genau_2145_am_freitag_greift():
|
||||
e = lauf_mit(Welt(), jetzt="2026-10-09T21:45:00")
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
|
||||
|
||||
def test_nach_dem_fenster_eine_sekunde_tut_nichts():
|
||||
e = lauf_mit(Welt(), jetzt="2026-10-09T21:45:01")
|
||||
assert e.code == 0 and e.welt.aufrufe == [] and e.dateien == [], kurz(e)
|
||||
|
||||
|
||||
# --- Marke flach und Entwarnung --------------------------------------------------------------------
|
||||
|
||||
def test_zweiter_lauf_nach_flach_tut_nichts():
|
||||
vorher = json.dumps({"zeit": "2026-10-08T15:45:00", "flach": True})
|
||||
e = lauf_mit(Welt(), ergebnis_vorher=vorher)
|
||||
assert e.code == 0 and e.welt.aufrufe == [] and e.welt.meldungen == [], kurz(e)
|
||||
assert e.ergebnis_text == vorher and e.notiz == "", kurz(e)
|
||||
assert e.dateien == [ERGEBNIS, LOG], kurz(e)
|
||||
|
||||
|
||||
def test_ergebnisdatei_ohne_echtes_true_laeuft_weiter():
|
||||
for vorher in ('{"flach": false}', '{"flach": "ja"}', '{"flach": 1}', "{kaputt", "[true]"):
|
||||
e = lauf_mit(Welt(), ergebnis_vorher=vorher)
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], (vorher, kurz(e))
|
||||
|
||||
|
||||
def test_erstes_flach_ohne_eingriff_meldet_entwarnung():
|
||||
vorher = json.dumps({"flach": False, "befunde": ["ibkr: Bestand nachher nicht lesbar; Ergebnis offen"]})
|
||||
e = lauf_mit(Welt(alpaca=FLACH, ibkr=FLACH), ergebnis_vorher=vorher)
|
||||
assert e.code == 0 and e.welt.glatt == [] and e.welt.schlaf == [], kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: flach"], kurz(e)
|
||||
assert ": flach." in e.notiz and "#needs-review" not in e.notiz, e.notiz
|
||||
assert e.json["flach"] is True, kurz(e)
|
||||
|
||||
|
||||
def test_gesperrte_instanz_verhindert_flach_auch_bei_flachem_bestand():
|
||||
e = lauf_mit(Welt(units=("inactive", "active"), alpaca=FLACH, ibkr=FLACH))
|
||||
assert e.code == 1 and e.welt.glatt == [], kurz(e)
|
||||
assert e.json["flach"] is False, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Befund"], kurz(e)
|
||||
assert "**Befund** #needs-review" in e.notiz, e.notiz
|
||||
|
||||
|
||||
# --- Modus lesen -----------------------------------------------------------------------------------
|
||||
|
||||
def test_modus_lesen_ruft_den_wrapper_nie_und_schreibt_nichts():
|
||||
e = lauf_mit(Welt(), modus="lesen")
|
||||
assert e.code == 0, kurz(e)
|
||||
assert e.welt.glatt == [] and e.welt.meldungen == [] and e.welt.schlaf == [], kurz(e)
|
||||
assert e.dateien == [], kurz(e)
|
||||
for erwartet in (
|
||||
"Bestand alpaca: 1 Position(en): AAPL long 10; offene Orders 0",
|
||||
"Glattstellen im Modus scharf: alpaca, ibkr",
|
||||
"ERGEBNIS: gelesen, ohne Befund",
|
||||
):
|
||||
assert erwartet in e.ausgabe, (erwartet, e.ausgabe)
|
||||
|
||||
|
||||
def test_modus_lesen_nachts_bei_markt_zu_liest_beide_bestaende_ohne_eingriff():
|
||||
e = lauf_mit(Welt(uhr=ZU), modus="lesen", jetzt="2026-10-08T00:30:00")
|
||||
assert e.code == 0, kurz(e)
|
||||
assert e.welt.glatt == [] and e.welt.meldungen == [] and e.dateien == [], kurz(e)
|
||||
assert e.welt.gelesen == {"alpaca": 1, "ibkr": 1}, kurz(e)
|
||||
for erwartet in (
|
||||
": ausserhalb",
|
||||
"Runner alpaca (trading-paper.service): LoadState=loaded, ActiveState=inactive",
|
||||
"Markt zu",
|
||||
"Bestand ibkr: 7 Position(en): AAPL long 10, CAT long 10, JPM long 10, KO long 20, MSFT long 10, "
|
||||
"NEE long 20, NVDA long 20; offene Orders 0",
|
||||
"Glattstellen im Modus scharf: keine (Markt zu",
|
||||
):
|
||||
assert erwartet in e.ausgabe, (erwartet, e.ausgabe)
|
||||
|
||||
|
||||
def test_modus_lesen_liest_auch_nach_flach():
|
||||
e = lauf_mit(Welt(), modus="lesen", ergebnis_vorher='{"flach": true}')
|
||||
assert e.welt.gelesen == {"alpaca": 1, "ibkr": 1} and e.welt.glatt == [], kurz(e)
|
||||
assert "Ergebnisdatei sagt flach" in e.ausgabe, e.ausgabe
|
||||
assert e.dateien == [ERGEBNIS], kurz(e)
|
||||
|
||||
|
||||
# --- Marktuhr --------------------------------------------------------------------------------------
|
||||
|
||||
def test_markt_zu_scharf_still_ohne_eingriff():
|
||||
e = lauf_mit(Welt(uhr=ZU))
|
||||
assert e.code == 0 and e.welt.glatt == [] and e.welt.meldungen == [], kurz(e)
|
||||
assert e.welt.gelesen == {"alpaca": 1, "ibkr": 0}, kurz(e)
|
||||
assert e.dateien == [LOG] and e.notiz == "", kurz(e)
|
||||
assert "Markt zu" in e.log, e.log
|
||||
|
||||
|
||||
def test_markt_schliesst_in_599_s_still():
|
||||
e = lauf_mit(Welt(uhr={"offen": True, "sekunden_bis_schluss": 599}))
|
||||
assert e.code == 0 and e.welt.glatt == [] and e.dateien == [LOG], kurz(e)
|
||||
|
||||
|
||||
def test_markt_schliesst_in_genau_600_s_greift():
|
||||
e = lauf_mit(Welt(uhr={"offen": True, "sekunden_bis_schluss": 600}))
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
|
||||
|
||||
def test_marktuhr_unlesbar_befund_ohne_eingriff():
|
||||
e = lauf_mit(Welt(uhr=None))
|
||||
assert e.code == 1 and e.welt.glatt == [] and e.welt.schlaf == [], kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Befund"], kurz(e)
|
||||
assert e.json["flach"] is False, kurz(e)
|
||||
assert "**Befund** #needs-review" in e.notiz and "Marktuhr nicht lesbar" in e.notiz, e.notiz
|
||||
|
||||
|
||||
# --- Runner-Units ----------------------------------------------------------------------------------
|
||||
|
||||
def test_laufender_runner_sperrt_genau_seine_instanz():
|
||||
e = lauf_mit(Welt(units=("active", "inactive")))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
erwartet = "alpaca: Runner-Unit trading-paper.service laeuft (LoadState=loaded, ActiveState=active); nicht angefasst"
|
||||
assert erwartet in e.notiz, e.notiz
|
||||
assert "#needs-review" in e.notiz, e.notiz
|
||||
e = lauf_mit(Welt(units=("inactive", "deactivating")))
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca"], kurz(e)
|
||||
|
||||
|
||||
def test_runner_zustand_unlesbar_instanz_nicht_angefasst():
|
||||
for zustand in ("", "unknown", Roh(0, "")):
|
||||
e = lauf_mit(Welt(units=(zustand, "inactive")))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], (zustand, kurz(e))
|
||||
|
||||
|
||||
def test_runner_unit_not_found_sperrt_die_instanz():
|
||||
e = lauf_mit(Welt(units=(("not-found", "inactive"), "inactive")))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
assert "LoadState=not-found, ActiveState=inactive" in e.notiz and "#needs-review" in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_runner_failed_gilt_als_stehend():
|
||||
e = lauf_mit(Welt(units=("failed", "failed")))
|
||||
assert e.code == 0 and e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
|
||||
|
||||
# --- Bestaende -------------------------------------------------------------------------------------
|
||||
|
||||
def test_ibkr_ohne_ergebnis_zeile_gilt_als_unlesbar():
|
||||
roh = Roh(1, "Positionen: 1\n AAPL.XNAS 10\nOffene Orders: 0\n")
|
||||
e = lauf_mit(Welt(ibkr=roh))
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca"], kurz(e)
|
||||
assert "ibkr: Bestand nicht lesbar" in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_ibkr_exit_4_gilt_als_unlesbar():
|
||||
e = lauf_mit(Welt(ibkr=Roh(4, ibkr_text(IBKR7))))
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca"], kurz(e)
|
||||
|
||||
|
||||
def test_ibkr_zahl_und_zeilen_passen_nicht():
|
||||
roh = Roh(0, "Positionen: 2\n AAPL.XNAS 10\nOffene Orders: 0\nERGEBNIS: gelesen\n")
|
||||
e = lauf_mit(Welt(ibkr=roh))
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca"], kurz(e)
|
||||
|
||||
|
||||
def test_ibkr_exit_1_mit_ergebnis_gelesen_gilt_als_gelesen():
|
||||
roh = Roh(1, ibkr_text(IBKR7, schluss="ERGEBNIS: gelesen, mit Befund"))
|
||||
e = lauf_mit(Welt(ibkr=roh))
|
||||
assert e.code == 0 and e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
|
||||
|
||||
def test_alpaca_bestand_unlesbar_nicht_angefasst():
|
||||
e = lauf_mit(Welt(alpaca=None))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
assert "alpaca: Bestand nicht lesbar (BrokerFehler: Test); nicht angefasst" in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_fremder_fehlertext_mit_zeilenumbruch_bleibt_einzeilig():
|
||||
e = lauf_mit(Welt(alpaca=None, alpaca_fehler="BrokerFehler: HTTP 500\n## Zweite Zeile"))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
assert "BrokerFehler: HTTP 500 ## Zweite Zeile" in e.notiz, e.notiz
|
||||
assert not any(zeile.startswith("## Zweite") for zeile in e.notiz.splitlines()), e.notiz
|
||||
assert not any(zeile.startswith("## Zweite") for zeile in e.log.splitlines()), e.log
|
||||
for titel, text in e.welt.meldungen:
|
||||
assert "\n" not in text and "\r" not in text, text
|
||||
|
||||
|
||||
def test_nur_offene_order_ohne_position_wird_glattgestellt():
|
||||
e = lauf_mit(Welt(alpaca={"positionen": [], "orders": 1}, ibkr=FLACH))
|
||||
assert e.code == 0 and e.welt.glatt == ["alpaca"], kurz(e)
|
||||
|
||||
|
||||
def test_flache_instanz_wird_nicht_angefasst():
|
||||
e = lauf_mit(Welt(alpaca=FLACH))
|
||||
assert e.code == 0 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
assert e.json["instanzen"]["alpaca"]["glattstellen"] is None, kurz(e)
|
||||
|
||||
|
||||
# --- Ergebnis nach dem Eingriff --------------------------------------------------------------------
|
||||
|
||||
def test_rest_offen_nach_eingriff():
|
||||
e = lauf_mit(Welt(ibkr_nachher={"positionen": [("NEE.XNYS", 20)], "orders": 0}))
|
||||
assert e.code == 0 and e.json["flach"] is False, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Rest offen"], kurz(e)
|
||||
assert ": Rest offen." in e.notiz and "#needs-review" not in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_offene_order_nachher_ist_nicht_flach():
|
||||
e = lauf_mit(Welt(alpaca_nachher={"positionen": [], "orders": 1}))
|
||||
assert e.json["flach"] is False, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Rest offen"], kurz(e)
|
||||
|
||||
|
||||
def test_wrapper_exit_1_und_danach_flach_heisst_flach_mit_befund():
|
||||
e = lauf_mit(Welt(wrapper_exit=1))
|
||||
assert e.code == 1, kurz(e)
|
||||
assert e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
assert e.json["flach"] is True, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: flach, mit Befund"], kurz(e)
|
||||
assert ": flach, mit Befund #needs-review." in e.notiz, e.notiz
|
||||
assert "alpaca: glattstellen endete mit Exit 1" in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_glattstellen_wirft_exit_none_befund_nachlesen_und_meldung():
|
||||
e = lauf_mit(Welt(wrapper_wirft=True))
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
assert e.welt.schlaf == [ab.WARTEN_S] and e.welt.gelesen == {"alpaca": 2, "ibkr": 2}, kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Befund"], kurz(e)
|
||||
assert "alpaca: glattstellen endete mit Exit None" in e.notiz, e.notiz
|
||||
assert e.json["instanzen"]["alpaca"]["glattstellen"]["exit"] is None, kurz(e)
|
||||
|
||||
|
||||
def test_nachher_unlesbar_ist_befund_und_nicht_flach():
|
||||
e = lauf_mit(Welt(ibkr_nachher=None))
|
||||
assert e.code == 1 and e.json["flach"] is False, kurz(e)
|
||||
assert "ibkr: Bestand nachher nicht lesbar" in e.notiz, e.notiz
|
||||
|
||||
|
||||
def test_ausnahme_traegt_nur_den_typnamen_in_notiz_und_meldung():
|
||||
welt = Welt()
|
||||
|
||||
def wirft(sekunden):
|
||||
raise RuntimeError(f"Order an {KONTO} limit {BETRAG}")
|
||||
|
||||
welt.schlafe = wirft
|
||||
e = lauf_mit(welt)
|
||||
assert e.code == 1, kurz(e)
|
||||
assert "Befund: abgebrochen: RuntimeError" in e.notiz, e.notiz
|
||||
texte = [e.notiz, e.ergebnis_text or ""] + [titel + " " + text for titel, text in e.welt.meldungen]
|
||||
for text in texte:
|
||||
assert KONTO not in text and BETRAG not in text, text
|
||||
assert "Traceback" in e.log, e.log
|
||||
|
||||
|
||||
# --- wsl.exe-Fehler und NUL ------------------------------------------------------------------------
|
||||
|
||||
def test_aufruf_nach_wsl_entfernt_nul_und_setzt_wsl_utf8():
|
||||
gesehen = []
|
||||
|
||||
def falscher_start(befehl, **optionen):
|
||||
gesehen.append((befehl, optionen))
|
||||
return SimpleNamespace(
|
||||
returncode=4294967295,
|
||||
stdout="Es ist keine Distribution".encode("utf-16-le"),
|
||||
stderr="Fehler".encode("utf-16-le"),
|
||||
)
|
||||
|
||||
alt = ab.starte
|
||||
ab.starte = falscher_start
|
||||
try:
|
||||
code, ausgabe, fehler = ab.wsl(["systemctl", "show", "trading-paper.service"], 5)
|
||||
finally:
|
||||
ab.starte = alt
|
||||
assert code == 4294967295, code
|
||||
assert NUL not in ausgabe and NUL not in fehler, (ausgabe, fehler)
|
||||
assert ausgabe == "Es ist keine Distribution" and fehler == "Fehler", (ausgabe, fehler)
|
||||
assert len(gesehen) == 1 and gesehen[0][0][0] == ab.WSL, gesehen
|
||||
assert gesehen[0][1]["env"]["WSL_UTF8"] == "1", gesehen[0][1].get("env", {}).get("WSL_UTF8")
|
||||
|
||||
|
||||
def test_runner_zustand_mit_nul_kommt_sauber_in_notiz_log_und_meldung():
|
||||
roh = Roh(0, "LoadState=loaded\nActiveState=E" + NUL + "s" + NUL + "\n")
|
||||
e = lauf_mit(Welt(units=(roh, "inactive")))
|
||||
assert e.code == 1 and e.welt.glatt == ["ibkr"], kurz(e)
|
||||
assert titel_von(e) == ["Abbau Restbestand: Befund"], kurz(e)
|
||||
assert NUL not in e.notiz and NUL not in e.log, (e.notiz, e.log)
|
||||
welt = Welt()
|
||||
with attrappe(welt):
|
||||
antwort = ab.melde("Abbau Restbestand: Probe" + NUL, "Text" + NUL, ab.SCHARF)
|
||||
assert antwort == "gesendet", antwort
|
||||
assert welt.meldungen == [("Abbau Restbestand: Probe", "Text")], welt.meldungen
|
||||
|
||||
|
||||
# --- Abdichtung ------------------------------------------------------------------------------------
|
||||
|
||||
def test_glattstellen_und_melden_ohne_scharf_rufen_nie_wsl():
|
||||
welt = Welt()
|
||||
with attrappe(welt):
|
||||
try:
|
||||
ab.glattstelle("alpaca", ab.LESEN)
|
||||
except RuntimeError:
|
||||
abgelehnt = True
|
||||
else:
|
||||
abgelehnt = False
|
||||
antwort = ab.melde("Abbau Restbestand: Probe", "Text", ab.LESEN)
|
||||
assert abgelehnt, "glattstelle lief im Modus lesen"
|
||||
assert welt.aufrufe == [], welt.aufrufe
|
||||
assert antwort.startswith("nicht gesendet"), antwort
|
||||
|
||||
|
||||
def test_abdichtung_alle_pfade_nach_draussen_zeigen_ins_leere():
|
||||
assert not Path(ab.WSL).exists(), ab.WSL
|
||||
assert ab.starte is subprocess.run, ab.starte
|
||||
for pfad in (ab.WSL_SKRIPT, ab.ALPACA_LESEN, ab.IBKR_LESEN, ab.GLATTSTELLEN, *ab.ORDNER_WSL.values()):
|
||||
assert pfad.startswith("/nicht-vorhanden/"), pfad
|
||||
try:
|
||||
ab.wsl(["systemctl", "show", "trading-paper.service"], 5)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
raise AssertionError("wsl() startete einen Prozess, obwohl WSL ins Leere zeigt")
|
||||
|
||||
|
||||
# --- WSL: Wrapper, Kette, Helfer -------------------------------------------------------------------
|
||||
|
||||
def test_wsl_wrapper_waehlt_alpaca_ersetzt_das_wort_und_maskiert_konto_und_betraege():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
ordner = instanzordner(tmp / "alpaca", exit_code=1)
|
||||
code, text = wrapper(tmp, "alpaca", str(ordner))
|
||||
angekommen = (ordner / "aufgerufen").read_text(encoding="utf-8")
|
||||
assert code == 1, (code, text)
|
||||
assert angekommen == f"--broker alpaca glattstellen --grund {GRUND} --bestaetigung {TESTWORT}", angekommen
|
||||
assert TESTWORT not in text and "[wort]" in text, text
|
||||
assert BETRAG not in text and "Kontowert [x]" in text, text
|
||||
assert KONTO not in text and "Order an [konto]" in text, text
|
||||
assert BETRAG_OHNE_GELDWORT not in text and "limit [x]" in text, text
|
||||
assert "venvs/trading/bin/python" in text, text
|
||||
|
||||
|
||||
def test_wsl_wrapper_ibkr_mit_attrappe_nimmt_ihr_venv_und_broker_vor_dem_unterbefehl():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
ordner = instanzordner(tmp / "ibkr")
|
||||
code, text = wrapper(tmp, "ibkr", str(ordner), "--broker", "attrappe")
|
||||
angekommen = (ordner / "aufgerufen").read_text(encoding="utf-8")
|
||||
assert code == 0, (code, text)
|
||||
assert angekommen.startswith("--broker attrappe glattstellen --grund "), angekommen
|
||||
assert "venvs/trading-ibkr/bin/python" in text, text
|
||||
|
||||
|
||||
def test_wsl_wrapper_lehnt_anderen_broker_und_fremde_angaben_ab():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
ordner = instanzordner(tmp / "alpaca")
|
||||
faelle = (
|
||||
("alpaca", str(ordner), "--broker", "alpaca"),
|
||||
("alpaca", str(ordner), "--broker"),
|
||||
("alpaca", str(ordner), "--weiter"),
|
||||
("live", str(ordner)),
|
||||
("alpaca",),
|
||||
)
|
||||
codes = [wrapper(tmp, *fall)[0] for fall in faelle]
|
||||
aufgerufen = (ordner / "aufgerufen").exists()
|
||||
assert codes == [2, 2, 2, 2, 2], codes
|
||||
assert not aufgerufen, "killswitch.py wurde trotz Bedienfehler gerufen"
|
||||
|
||||
|
||||
def test_wsl_wrapper_ohne_wort_ruft_killswitch_nicht():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
ordner = instanzordner(tmp / "alpaca", wort=None)
|
||||
code, text = wrapper(tmp, "alpaca", str(ordner))
|
||||
aufgerufen = (ordner / "aufgerufen").exists()
|
||||
assert code == 4 and not aufgerufen, (code, aufgerufen, text)
|
||||
|
||||
|
||||
def test_wsl_kette_wort_konto_und_betrag_stehen_nirgends():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp = Path(tmp)
|
||||
ordner = {instanz: instanzordner(tmp / instanz, exit_code=1) for instanz in ab.INSTANZEN}
|
||||
|
||||
def echter_wrapper(argumente):
|
||||
assert Path(argumente[1]).resolve().is_relative_to(tmp.resolve()), argumente
|
||||
code, text = wrapper(tmp, *argumente)
|
||||
return code, text, ""
|
||||
|
||||
alt = ab.ORDNER_WSL
|
||||
ab.ORDNER_WSL = {instanz: str(pfad) for instanz, pfad in ordner.items()}
|
||||
try:
|
||||
e = lauf_mit(Welt(echter_wrapper=echter_wrapper))
|
||||
finally:
|
||||
ab.ORDNER_WSL = alt
|
||||
angekommen = (ordner["ibkr"] / "aufgerufen").read_text(encoding="utf-8")
|
||||
assert e.code == 1 and e.welt.glatt == ["alpaca", "ibkr"], kurz(e)
|
||||
assert angekommen.startswith("--broker ibkr glattstellen --grund "), angekommen
|
||||
assert angekommen.endswith("--bestaetigung " + TESTWORT), angekommen
|
||||
assert "[wort]" in e.ergebnis_text and "Kontowert [x]" in e.notiz, (e.ergebnis_text, e.notiz)
|
||||
assert "[konto]" in e.notiz, e.notiz
|
||||
texte = [e.ausgabe, e.ergebnis_text, e.log, e.notiz] + [titel + " " + text for titel, text in e.welt.meldungen]
|
||||
for text in texte:
|
||||
for geheim in (TESTWORT, KONTO, BETRAG, BETRAG_OHNE_GELDWORT):
|
||||
assert geheim not in text, (geheim, text)
|
||||
|
||||
|
||||
def test_wsl_alpaca_helfer_gibt_stueck_ohne_betrag():
|
||||
nur_in_wsl()
|
||||
import abbau_alpaca_lesen as helfer
|
||||
|
||||
jetzt = datetime(2026, 10, 8, 14, 0, tzinfo=timezone.utc)
|
||||
daten = helfer.lies(FalscherAlpaca(jetzt), jetzt)
|
||||
text = json.dumps(daten)
|
||||
assert daten["uhr"] == {"offen": True, "sekunden_bis_schluss": 1200}, daten
|
||||
assert daten["positionen"] == [{"symbol": "AAPL", "menge": 10}, {"symbol": "KO", "menge": -20}], daten
|
||||
assert daten["orders"] == 1 and daten["bestand_fehler"] is None, daten
|
||||
for betrag in ("1234.56", "98.76", "1111.11", "55.55"):
|
||||
assert betrag not in text, (betrag, text)
|
||||
|
||||
|
||||
def test_wsl_alpaca_helfer_maskiert_konto_und_betraege_im_fehlertext():
|
||||
nur_in_wsl()
|
||||
import abbau_alpaca_lesen as helfer
|
||||
|
||||
jetzt = datetime(2026, 10, 8, 14, 0, tzinfo=timezone.utc)
|
||||
fehler = f"Order an {KONTO}: limit {BETRAG_OHNE_GELDWORT}, Kontowert {BETRAG} USD"
|
||||
daten = helfer.lies(FalscherAlpaca(jetzt, fehler=fehler), jetzt)
|
||||
text = json.dumps(daten)
|
||||
assert daten["positionen"] is None and daten["orders"] is None, daten
|
||||
assert daten["bestand_fehler"].startswith("RuntimeError: "), daten
|
||||
assert "Order an [konto]" in daten["bestand_fehler"] and "limit [x]" in daten["bestand_fehler"], daten
|
||||
for geheim in (KONTO, BETRAG, BETRAG_OHNE_GELDWORT):
|
||||
assert geheim not in text, (geheim, text)
|
||||
assert daten["uhr"] == {"offen": True, "sekunden_bis_schluss": 1200}, daten
|
||||
|
||||
|
||||
def test_wsl_ibkr_lesen_ohne_erreichbaren_ordner_startet_nichts():
|
||||
nur_in_wsl()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
leer = Path(tmp)
|
||||
ohne = subprocess.run(["bash", str(IBKR_SKRIPT)], cwd=leer, capture_output=True, text=True, timeout=60)
|
||||
fehlt = subprocess.run(
|
||||
["bash", str(IBKR_SKRIPT), "/nicht-vorhanden/ibkr"], cwd=leer, capture_output=True, text=True, timeout=60
|
||||
)
|
||||
assert ohne.returncode == 2, (ohne.returncode, ohne.stderr)
|
||||
assert fehlt.returncode == 4, (fehlt.returncode, fehlt.stderr)
|
||||
|
||||
|
||||
# --- Lauf als Skript -------------------------------------------------------------------------------
|
||||
|
||||
def main() -> int:
|
||||
faelle = [(name, wert) for name, wert in globals().items() if name.startswith("test_") and callable(wert)]
|
||||
abweichungen = 0
|
||||
uebersprungen = 0
|
||||
for name, fall in faelle:
|
||||
try:
|
||||
fall()
|
||||
except Uebersprungen as grund:
|
||||
uebersprungen += 1
|
||||
print(f"UEBERSPRUNGEN {name}: {grund}")
|
||||
continue
|
||||
except (Exception, SystemExit) as ausnahme:
|
||||
abweichungen += 1
|
||||
print(f"FEHL {name}: {type(ausnahme).__name__}: {str(ausnahme)[:400]!r}")
|
||||
continue
|
||||
print(f"OK {name}")
|
||||
print(f"ERGEBNIS: {len(faelle)} Faelle, {abweichungen} Abweichung(en), {uebersprungen} uebersprungen")
|
||||
return 1 if abweichungen else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user