185 lines
9.0 KiB
Python
185 lines
9.0 KiB
Python
"""Laufzeiten isoliert aktualisieren; Rollback bleibt ohne Zusatzpakete nutzbar."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
|
|
from .setup import get_latest_mustang_version, install_mustang
|
|
|
|
|
|
def point(link: Path, target: Path) -> None:
|
|
temporary = link.with_name(link.name + '.next')
|
|
temporary.unlink(missing_ok=True)
|
|
temporary.symlink_to(target)
|
|
os.replace(temporary, link)
|
|
|
|
|
|
@contextmanager
|
|
def locked(state: Path):
|
|
state.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
with (state / 'update.lock').open('a') as handle:
|
|
try:
|
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError as exc:
|
|
raise RuntimeError('Ein anderer Update-/Rollback-Lauf ist aktiv.') from exc
|
|
yield
|
|
|
|
|
|
class Maintenance:
|
|
def __init__(self, root: Path, log):
|
|
self.root = root.resolve()
|
|
self.state = self.root / '.runtime'
|
|
self.log = log
|
|
|
|
def run(self, args, *, cwd=None, env=None, timeout=1200):
|
|
self.log.info('Aufruf: %s', ' '.join(map(str, args)))
|
|
result = subprocess.run(list(map(str, args)), cwd=cwd or self.root, env=env,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, timeout=timeout)
|
|
self.log.info('%s', result.stdout.rstrip())
|
|
if result.returncode:
|
|
raise RuntimeError(f'Befehl fehlgeschlagen (Exit {result.returncode}): {args[0]}')
|
|
return result.stdout
|
|
|
|
def health(self, release: Path):
|
|
python = release / 'venv/bin/python'
|
|
env = os.environ.copy()
|
|
for key in ('PYTHONPATH', 'JR_EINVOICE_PRESETS', 'JR_EINVOICE_HOME'):
|
|
env.pop(key, None)
|
|
env['JR_EINVOICE_MUSTANG_JAR'] = str(release / 'Mustang-CLI.jar')
|
|
env['JR_EINVOICE_INTEGRATION'] = '1'
|
|
env['QT_QPA_PLATFORM'] = 'offscreen'
|
|
self.log.info('PRÜFUNG: pip check, WeasyPrint, CLI und echte Rechnungs-/Negativtests')
|
|
self.run([python, '-I', '-m', 'pip', 'check'], env=env)
|
|
with tempfile.TemporaryDirectory(prefix='einvoice-health-') as folder:
|
|
env['JR_EINVOICE_HOME'] = folder
|
|
self.run([python, '-I', '-c', 'from weasyprint import HTML; HTML(string="<p>PDF-Test</p>").write_pdf()'], cwd=folder, env=env)
|
|
for module in ('jr_einvoice.quick_cli', 'jr_einvoice.cli'):
|
|
code = ('from jr_einvoice.cli import app; app()' if module.endswith('.cli')
|
|
else 'from jr_einvoice.quick_cli import main; main()')
|
|
self.run([python, '-I', '-c', code, '--help'], cwd=folder, env=env)
|
|
self.run([python, '-I', '-m', 'unittest', 'discover', '-s', self.root / 'tests', '-v'],
|
|
cwd=folder, env=env)
|
|
|
|
def baseline(self):
|
|
baseline = self.state / 'releases' / ('baseline-' + uuid.uuid4().hex[:8])
|
|
baseline.mkdir(parents=True)
|
|
venv = self.root / '.venv'
|
|
jar = Path(os.environ.get('JR_EINVOICE_MUSTANG_JAR', self.root / 'tools/mustang/Mustang-CLI.jar')).resolve()
|
|
if not (venv / 'bin/python').exists() or not jar.is_file():
|
|
raise RuntimeError('Erstinstallation nötig: .venv und Mustang fehlen; siehe README.')
|
|
(baseline / 'venv').symlink_to(venv)
|
|
shutil.copy2(jar, baseline / 'Mustang-CLI.jar')
|
|
return baseline
|
|
|
|
def activate(self, target: Path):
|
|
active = self.state / 'current'
|
|
old = active.resolve() if active.is_symlink() else self.baseline()
|
|
# Rücksprungziel vor dem einzigen aktiven Laufzeitzeiger sichern.
|
|
point(self.state / 'previous', old)
|
|
self.log.info('AKTIVIERUNG: %s', target)
|
|
point(active, target)
|
|
try:
|
|
self.health(active.resolve())
|
|
except BaseException:
|
|
point(active, old)
|
|
self.log.error('ROLLBACK: bisherige Laufzeit wieder aktiv: %s', old)
|
|
raise
|
|
self.log.info('ERFOLG: Laufzeit aktiv. Rollback verfügbar: %s', old)
|
|
|
|
def update(self, apply=False, version=None, gui=False):
|
|
version = version or get_latest_mustang_version(self.log)
|
|
if not re.fullmatch(r'\d+(?:\.\d+)+', version):
|
|
raise ValueError(f'Keine stabile Mustang-Version: {version!r}; --mustang-version verwenden.')
|
|
release = self.state / 'releases' / (datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ-') + uuid.uuid4().hex[:8])
|
|
release.mkdir(parents=True)
|
|
self.log.info('KANDIDAT: %s; Mustang %s', release, version)
|
|
self.run(['java', '-version'])
|
|
self.run([sys.executable, '-m', 'venv', release / 'venv'])
|
|
python = release / 'venv/bin/python'
|
|
# Eine einmal aktivierte Oberfläche auch bei späteren Updates mitnehmen.
|
|
current_manifest = self.state / 'current/manifest.json'
|
|
if current_manifest.is_file():
|
|
gui = gui or json.loads(current_manifest.read_text()).get('gui', False)
|
|
extras = 'pdf,gui' if gui else 'pdf'
|
|
self.run([python, '-m', 'pip', 'install', '--upgrade', 'pip'])
|
|
self.run([python, '-m', 'pip', 'install', '--upgrade', '--upgrade-strategy', 'eager',
|
|
'--report', release / 'install-report.json', f'{self.root}[{extras}]'])
|
|
install_mustang(version, release / 'Mustang-CLI.jar', self.log)
|
|
frozen = self.run([python, '-m', 'pip', 'freeze', '--all'])
|
|
(release / 'requirements.freeze.txt').write_text(frozen)
|
|
packages = json.loads(self.run([python, '-m', 'pip', 'list', '--format=json']))
|
|
old_python = (self.state / 'current/venv/bin/python') if (self.state / 'current').exists() else self.root / '.venv/bin/python'
|
|
old = {p['name'].lower(): p['version'] for p in json.loads(self.run([old_python, '-m', 'pip', 'list', '--format=json']))}
|
|
for package in packages:
|
|
before = old.get(package['name'].lower(), '(neu)')
|
|
if before != package['version']:
|
|
self.log.info('VERSION: %s: %s -> %s', package['name'], before, package['version'])
|
|
manifest = {'mustang': version, 'gui': gui, 'jar_sha256': hashlib.sha256((release / 'Mustang-CLI.jar').read_bytes()).hexdigest(),
|
|
'packages': packages, 'source_commit': self.run(['git', 'rev-parse', 'HEAD']).strip()}
|
|
(release / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
|
|
self.health(release)
|
|
if apply:
|
|
self.activate(release)
|
|
else:
|
|
self.log.info('PRÜFUNG ERFOLGREICH: keine Umschaltung. Zum Übernehmen erneut mit --apply starten.')
|
|
|
|
def rollback(self):
|
|
previous = self.state / 'previous'
|
|
if not previous.exists():
|
|
raise RuntimeError('Kein vorheriger Laufzeitstand vorhanden.')
|
|
target = previous.resolve()
|
|
self.health(target)
|
|
self.activate(target)
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description='Python-Abhängigkeiten und Mustang isoliert aktualisieren und testen.')
|
|
mode = parser.add_mutually_exclusive_group()
|
|
mode.add_argument('--apply', action='store_true', help='Nach erfolgreichen Tests aktivieren')
|
|
mode.add_argument('--rollback', action='store_true', help='Vorherigen Stand prüfen und aktivieren')
|
|
mode.add_argument('--status', action='store_true', help='Aktiven und vorherigen Stand anzeigen, ohne Netzwerk')
|
|
parser.add_argument('--mustang-version', help='Explizite stabile Version statt Maven-Release')
|
|
parser.add_argument('--with-gui', action='store_true', help='Desktop-Oberfläche einschließlich PDF-Anzeige installieren')
|
|
args = parser.parse_args(argv)
|
|
root = Path(__file__).resolve().parent.parent
|
|
state = root / '.runtime'
|
|
try:
|
|
with locked(state):
|
|
logs = state / 'logs'
|
|
logs.mkdir(exist_ok=True)
|
|
logfile = logs / (datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ-') + uuid.uuid4().hex[:8] + '.log')
|
|
logger = logging.getLogger('maintenance')
|
|
logger.setLevel(logging.INFO)
|
|
logger.handlers.clear()
|
|
for handler in (logging.StreamHandler(sys.stdout), logging.FileHandler(logfile)):
|
|
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
|
|
logger.addHandler(handler)
|
|
logger.info('PROTOKOLL: %s', logfile)
|
|
maintenance = Maintenance(root, logger)
|
|
if args.status:
|
|
for name in ('current', 'previous'):
|
|
link = state / name
|
|
logger.info('%s: %s', name, link.resolve() if link.exists() else '(noch nicht verwaltet)')
|
|
elif args.rollback:
|
|
maintenance.rollback()
|
|
else:
|
|
maintenance.update(args.apply, args.mustang_version, args.with_gui)
|
|
except (Exception, KeyboardInterrupt) as exc:
|
|
logging.getLogger('maintenance').error('FEHLER: %s; nicht aktivierte Kandidaten bleiben zur Diagnose erhalten.', exc)
|
|
return 1
|
|
return 0
|