#!/usr/bin/env python3 """ create_dataset.py Legt über Proteus einen deklarativ definierten, reproduzierbaren Testdatensatz an und liest ihn danach FRISCH VOM SERVER zurück (nicht aus dem lokalen Proteus-Cache) - damit automatische Reformatierung durch die Zielversion sichtbar wird. Ergebnis wird als JSON-Snapshot gespeichert. Gedacht, um dieselbe dataset.yaml gegen zwei Tryton-Versionen laufen zu lassen (zwei getrennte, frisch installierte Datenbanken) und die beiden Snapshots danach mit compare_snapshots.py zu vergleichen. Nutzung: python3 create_dataset.py --config dataset.yaml \\ --database mydb_68 --output snapshot_68.json python3 create_dataset.py --config dataset.yaml \\ --database mydb_70 --output snapshot_70.json python3 compare_snapshots.py snapshot_68.json snapshot_70.json \\ --label-a 6.8 --label-b 7.0 WICHTIG: Dieses Skript wurde nicht gegen einen echten Tryton-Server getestet (dafür fehlt hier die Umgebung) - nur auf Syntax geprüft. Bitte bei dir gegenlesen, insbesondere proteus_config.set_trytond(...) an eure tatsächliche Verbindungsart anpassen (lokal via Unix-Socket/Config-Datei oder per XML-RPC/JSON-RPC über Netzwerk - die Proteus-API unterscheidet sich da je nach Version leicht). Feld-Umbenennungen (FIELD_RENAMES): Da dieselbe dataset.yaml gegen mehrere Versionen laufen soll, aber Felder zwischen Versionen umbenannt werden können (bestätigt: stock.move.uom -> stock.move.unit zwischen 5.0 und 8.0), wird pro Modell zuerst geprüft, ob der im YAML angegebene Feldname auf dem jeweiligen Server tatsächlich existiert (via fields_get()). Falls nicht, aber ein bekannter Alias aus FIELD_RENAMES existiert UND der Alias-Name dort existiert, wird der Alias verwendet - der Datensatz landet dann inhaltlich identisch in beiden Datenbanken, nur eben unter dem jeweils richtigen Feldnamen. Im JSON-Snapshot ("stored") wird trotzdem immer der ORIGINALE YAML-Feldname als Key verwendet, damit compare_snapshots.py die Werte über beide Versionen hinweg unter demselben Key vergleicht. Ein wirklich unbekanntes Feld (kein Treffer in FIELD_RENAMES, oder Alias existiert dort ebenfalls nicht) wird NICHT stillschweigend übergangen, sondern lässt den Record ganz normal mit AttributeError fehlschlagen - das ist ja gerade das gesuchte Signal für eine echte Schema-Änderung. """ import argparse import decimal import datetime import json import sys import traceback from pathlib import Path try: import yaml except ImportError: yaml = None from proteus import Model, config as proteus_config # Bekannte Feld-Umbenennungen zwischen Tryton-Versionen: Modell -> {einer # der beiden Feldnamen: der jeweils andere}. Nur eine Richtung eintragen # (z. B. die 5.0->8.0-Richtung) - _bidirectional_field_renames() ergänzt # die Rückrichtung automatisch, damit die Auflösung unabhängig davon # funktioniert, mit welcher Version die dataset.yaml ursprünglich gebaut # wurde. Bei Bedarf ergänzen, sobald ein compare_snapshots.py-Lauf einen # weiteren Fall aufdeckt. FIELD_RENAMES = { "stock.move": {"uom": "unit"}, } def _bidirectional_field_renames(field_renames): result = {} for model_name, renames in field_renames.items(): both = dict(renames) both.update({new: old for old, new in renames.items()}) result[model_name] = both return result BIDIRECTIONAL_FIELD_RENAMES = _bidirectional_field_renames(FIELD_RENAMES) def resolve_field_name(RecordModel, model_name, field_name, fields_cache): """Liefert den auf DIESEM Server tatsächlich existierenden Feldnamen. Existiert field_name direkt (fields_get()), wird er unverändert zurückgegeben. Sonst wird in FIELD_RENAMES nachgeschaut (in beide Richtungen), ob ein bekannter Alias existiert UND auf diesem Server tatsächlich vorhanden ist. Andernfalls bleibt field_name unverändert, damit ein wirklich unbekanntes Feld weiterhin als Fehler auffällt. """ if model_name not in fields_cache: try: fields_cache[model_name] = set(RecordModel.fields_get().keys()) except Exception: fields_cache[model_name] = None # unbekannt - keine Alias-Auflösung möglich available = fields_cache[model_name] if available is None or field_name in available: return field_name alias = BIDIRECTIONAL_FIELD_RENAMES.get(model_name, {}).get(field_name) if alias and alias in available: return alias return field_name def load_spec(path): text = Path(path).read_text(encoding="utf-8") if path.endswith((".yaml", ".yml")): if yaml is None: raise RuntimeError( "PyYAML nicht installiert (pip install pyyaml) - " "oder --config auf eine .json-Datei zeigen lassen." ) return yaml.safe_load(text) return json.loads(text) def resolve_value(value, created, search_cache): """ Löst Referenzen der Form {"$ref": "irgendein_key"} auf bereits in diesem Lauf erzeugte Datensätze auf (z. B. um eine Adresse mit der vorher angelegten Party zu verknüpfen). Referenzen können nur auf Records verweisen, die im YAML VOR der aktuellen Stelle stehen. Versteht außerdem die von rst_to_dataset.py erzeugten Marker für Werte, die kein einfacher JSON-Typ sind: {"$decimal": "12.34"} -> decimal.Decimal("12.34") {"$date": "2024-01-15"} -> datetime.date(2024, 1, 15) {"$search": {"model": "product.uom", "domain": [["name", "=", "Unit"]]}} -> sucht den Record live auf dem Zielserver (Model.find(domain)) - für Stammdaten, die nicht Teil des Testdatensatzes sind, sondern beim Modul-Aktivieren per XML-Fixture angelegt werden (Länder, Währungen, Einheiten, Standard-Lagerorte, ...). Wirft einen Fehler, wenn nichts gefunden wird, statt einen neuen Record anzulegen - ein fehlender Treffer bedeutet meist, dass sich die Fixture-Daten zwischen den Versionen geändert haben. {"$manual": ""} -> Fehler mit Klartext, da diese Stelle noch von Hand ausgefüllt werden muss. """ if isinstance(value, dict) and "$ref" in value: if value["$ref"] not in created: raise KeyError( f"Referenz '{value['$ref']}' unbekannt - Datensatz mit " f"diesem key wurde noch nicht angelegt oder ist " f"fehlgeschlagen." ) return created[value["$ref"]] if isinstance(value, dict) and "$search" in value: spec = value["$search"] cache_key = (spec["model"], json.dumps(spec["domain"], sort_keys=True)) if cache_key not in search_cache: matches = Model.get(spec["model"]).find(spec["domain"]) if not matches: raise ValueError( f"$search fand keinen Treffer: model={spec['model']!r} " f"domain={spec['domain']!r} - Stammdaten auf diesem " f"Server geändert/entfernt?" ) search_cache[cache_key] = matches[0].id return search_cache[cache_key] if isinstance(value, dict) and "$decimal" in value: return decimal.Decimal(value["$decimal"]) if isinstance(value, dict) and "$date" in value: return datetime.date.fromisoformat(value["$date"]) if isinstance(value, dict) and "$manual" in value: raise ValueError( f"Nicht ausgefüllter $manual-Wert: {value['$manual']!r} - " f"bitte im YAML durch einen echten Wert ersetzen, bevor " f"dieser Datensatz angelegt wird." ) if isinstance(value, list): return [resolve_value(v, created, search_cache) for v in value] if isinstance(value, dict): return {k: resolve_value(v, created, search_cache) for k, v in value.items()} return value def serialize(value): """Proteus-/Tryton-Werte in eine JSON-taugliche, vergleichbare Form bringen.""" if value is None: return None if hasattr(value, "id") and hasattr(value, "__class__"): # Verknüpfter Record (Many2One o. ä.) - Modellname + id vergleichen, # nicht das Objekt selbst. return { "__record__": getattr(value, "__class__").__name__, "id": value.id, } if isinstance(value, (list, tuple)): return [serialize(v) for v in value] try: json.dumps(value) return value except TypeError: # Decimal, date, datetime, etc. - als String vergleichen return str(value) def create_records(spec, created, results): fields_cache = {} # model_name -> Set der auf DIESEM Server bekannten Felder search_cache = {} # (model, domain-json) -> id, damit dieselbe Stammdaten-Suche nicht mehrfach läuft for entry in spec["records"]: key = entry["key"] model_name = entry["model"] fields = entry.get("fields", {}) record_result = { "key": key, "model": model_name, "status": "ok", "error": None, "stored": {}, } try: RecordModel = Model.get(model_name) record = RecordModel() # actual_name je YAML-Feldname merken: fürs Zurücklesen unten # brauchen wir denselben (ggf. per Alias aufgelösten) Namen, # im Snapshot aber weiterhin den originalen YAML-Namen als Key. actual_names = {} for field_name, raw_value in fields.items(): actual_name = resolve_field_name(RecordModel, model_name, field_name, fields_cache) actual_names[field_name] = actual_name setattr(record, actual_name, resolve_value(raw_value, created, search_cache)) record.save() created[key] = record.id # Bewusst NEU vom Server laden statt das lokale Proteus-Objekt # weiterzuverwenden - das lokale Objekt kann noch den Wert # zeigen, den WIR gesetzt haben, nicht den, den der Server # tatsächlich abgespeichert (und ggf. reformatiert) hat. reloaded = RecordModel(record.id) for field_name in fields: try: record_result["stored"][field_name] = serialize( getattr(reloaded, actual_names[field_name]) ) except Exception as e: record_result["stored"][field_name] = f"" except Exception as e: record_result["status"] = "error" record_result["error"] = f"{type(e).__name__}: {e}" record_result["traceback"] = traceback.format_exc() results.append(record_result) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", required=True, help="Pfad zur dataset.yaml/json") parser.add_argument("--database", required=True, help="Name der Tryton-Datenbank") parser.add_argument( "--config-file", default=None, help="Pfad zu trytond.conf, falls nicht der Standardpfad verwendet wird", ) parser.add_argument("--output", required=True, help="Wohin der JSON-Snapshot geschrieben wird") args = parser.parse_args() # ACHTUNG: set_trytond() ist die lokale Variante (Proteus + trytond im # selben Prozess/Server, kein separater Client-Server-Aufbau). Läuft # euer trytond als eigener Prozess, braucht ihr stattdessen # proteus_config.set_xmlrpc(url) mit einer URL wie # "http://admin:PASSWORT@localhost:8000/mydb_68/". proteus_config.set_trytond(database=args.database, config_file=args.config_file) spec = load_spec(args.config) created = {} results = [] create_records(spec, created, results) snapshot = {"database": args.database, "results": results} Path(args.output).write_text( json.dumps(snapshot, indent=2, ensure_ascii=False), encoding="utf-8" ) ok = sum(1 for r in results if r["status"] == "ok") err = sum(1 for r in results if r["status"] == "error") print(f"{ok} Datensätze ok, {err} Fehler -> {args.output}") if err: print("Fehlgeschlagene Datensätze:") for r in results: if r["status"] == "error": print(f" - {r['key']} ({r['model']}): {r['error']}") if __name__ == "__main__": sys.exit(main())