21a844cb8a
A previous `git pull --rebase origin main` dropped 177 local commits,
losing 3400+ files across admin-v2, backend, studio-v2, website,
klausur-service, and many other services. The partial restore attempt
(660295e2) only recovered some files.
This commit restores all missing files from pre-rebase ref 98933f5e
while preserving post-rebase additions (night-scheduler, night-mode UI,
NightModeWidget dashboard integration).
Restored features include:
- AI Module Sidebar (FAB), OCR Labeling, OCR Compare
- GPU Dashboard, RAG Pipeline, Magic Help
- Klausur-Korrektur (8 files), Abitur-Archiv (5+ files)
- Companion, Zeugnisse-Crawler, Screen Flow
- Full backend, studio-v2, website, klausur-service
- All compliance SDKs, agent-core, voice-service
- CI/CD configs, documentation, scripts
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
228 lines
9.3 KiB
Python
228 lines
9.3 KiB
Python
"""
|
|
SQLAlchemy Models fuer Test Registry.
|
|
|
|
Definiert die Datenbank-Tabellen fuer persistente Test-Speicherung:
|
|
- TestRunDB: Jeder Test-Durchlauf
|
|
- TestResultDB: Einzelne Test-Ergebnisse
|
|
- FailedTestBacklogDB: Persistenter Backlog fuer zu fixende Tests
|
|
- TestFixHistoryDB: Historie aller Fix-Versuche
|
|
- TestServiceStatsDB: Aggregierte Statistiken pro Service
|
|
"""
|
|
from datetime import datetime
|
|
from sqlalchemy import (
|
|
Column, Integer, String, Float, Text, DateTime, Boolean,
|
|
ForeignKey, UniqueConstraint, Index
|
|
)
|
|
from sqlalchemy.orm import relationship
|
|
|
|
# Nutze die gleiche Base wie Classroom Engine fuer konsistente Migrations
|
|
from classroom_engine.database import Base
|
|
|
|
|
|
class TestRunDB(Base):
|
|
"""
|
|
Speichert jeden Test-Durchlauf.
|
|
Enthaelt Metadaten und Aggregat-Statistiken.
|
|
"""
|
|
__tablename__ = 'test_runs'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
run_id = Column(String(50), unique=True, nullable=False, index=True)
|
|
service = Column(String(100), nullable=False, index=True)
|
|
framework = Column(String(50), nullable=False)
|
|
started_at = Column(DateTime, nullable=False, index=True)
|
|
completed_at = Column(DateTime, nullable=True)
|
|
status = Column(String(20), nullable=False) # queued, running, completed, failed
|
|
total_tests = Column(Integer, default=0)
|
|
passed_tests = Column(Integer, default=0)
|
|
failed_tests = Column(Integer, default=0)
|
|
skipped_tests = Column(Integer, default=0)
|
|
duration_seconds = Column(Float, default=0)
|
|
git_commit = Column(String(40), nullable=True)
|
|
git_branch = Column(String(100), nullable=True)
|
|
triggered_by = Column(String(50), nullable=True) # manual, ci, schedule
|
|
output = Column(Text, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationship zu einzelnen Test-Ergebnissen
|
|
results = relationship("TestResultDB", back_populates="run", cascade="all, delete-orphan")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.run_id,
|
|
"run_id": self.run_id,
|
|
"service": self.service,
|
|
"framework": self.framework,
|
|
"started_at": self.started_at.isoformat() if self.started_at else None,
|
|
"completed_at": self.completed_at.isoformat() if self.completed_at else None,
|
|
"status": self.status,
|
|
"total_tests": self.total_tests,
|
|
"passed_tests": self.passed_tests,
|
|
"failed_tests": self.failed_tests,
|
|
"skipped_tests": self.skipped_tests,
|
|
"duration_seconds": self.duration_seconds,
|
|
"git_commit": self.git_commit,
|
|
"git_branch": self.git_branch,
|
|
"triggered_by": self.triggered_by,
|
|
}
|
|
|
|
|
|
class TestResultDB(Base):
|
|
"""
|
|
Speichert einzelne Test-Ergebnisse pro Run.
|
|
Ermoeglicht detaillierte Analyse fehlgeschlagener Tests.
|
|
"""
|
|
__tablename__ = 'test_results'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
run_id = Column(String(50), ForeignKey('test_runs.run_id', ondelete='CASCADE'), nullable=False, index=True)
|
|
test_name = Column(String(500), nullable=False, index=True)
|
|
test_file = Column(String(500), nullable=True)
|
|
line_number = Column(Integer, nullable=True)
|
|
status = Column(String(20), nullable=False, index=True) # passed, failed, skipped, error
|
|
duration_ms = Column(Float, nullable=True)
|
|
error_message = Column(Text, nullable=True)
|
|
error_type = Column(String(100), nullable=True)
|
|
output = Column(Text, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationship zum Run
|
|
run = relationship("TestRunDB", back_populates="results")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"run_id": self.run_id,
|
|
"test_name": self.test_name,
|
|
"test_file": self.test_file,
|
|
"line_number": self.line_number,
|
|
"status": self.status,
|
|
"duration_ms": self.duration_ms,
|
|
"error_message": self.error_message,
|
|
"error_type": self.error_type,
|
|
}
|
|
|
|
|
|
class FailedTestBacklogDB(Base):
|
|
"""
|
|
Persistenter Backlog fuer fehlgeschlagene Tests.
|
|
Aggregiert Fehler ueber mehrere Runs hinweg.
|
|
"""
|
|
__tablename__ = 'failed_tests_backlog'
|
|
__table_args__ = (
|
|
UniqueConstraint('test_name', 'service', name='uq_backlog_test_service'),
|
|
)
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
test_name = Column(String(500), nullable=False)
|
|
test_file = Column(String(500), nullable=True)
|
|
service = Column(String(100), nullable=False, index=True)
|
|
framework = Column(String(50), nullable=True)
|
|
error_message = Column(Text, nullable=True)
|
|
error_type = Column(String(100), nullable=True)
|
|
first_failed_at = Column(DateTime, nullable=False)
|
|
last_failed_at = Column(DateTime, nullable=False)
|
|
failure_count = Column(Integer, default=1)
|
|
status = Column(String(30), default='open', index=True) # open, in_progress, fixed, wont_fix, flaky
|
|
priority = Column(String(20), default='medium', index=True) # critical, high, medium, low
|
|
assigned_to = Column(String(100), nullable=True)
|
|
fix_suggestion = Column(Text, nullable=True)
|
|
notes = Column(Text, nullable=True)
|
|
# Resolution-Felder (auto-close wenn Tests bestehen)
|
|
resolved_at = Column(DateTime, nullable=True)
|
|
resolution_commit = Column(String(50), nullable=True)
|
|
resolution_notes = Column(Text, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# Relationship zu Fix-Historie
|
|
fixes = relationship("TestFixHistoryDB", back_populates="backlog_item", cascade="all, delete-orphan")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"test_name": self.test_name,
|
|
"test_file": self.test_file,
|
|
"service": self.service,
|
|
"framework": self.framework,
|
|
"error_message": self.error_message,
|
|
"error_type": self.error_type,
|
|
"first_failed_at": self.first_failed_at.isoformat() if self.first_failed_at else None,
|
|
"last_failed_at": self.last_failed_at.isoformat() if self.last_failed_at else None,
|
|
"failure_count": self.failure_count,
|
|
"status": self.status,
|
|
"priority": self.priority,
|
|
"assigned_to": self.assigned_to,
|
|
"fix_suggestion": self.fix_suggestion,
|
|
"notes": self.notes,
|
|
"resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
|
|
"resolution_commit": self.resolution_commit,
|
|
"resolution_notes": self.resolution_notes,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class TestFixHistoryDB(Base):
|
|
"""
|
|
Historie aller Fix-Versuche fuer einen Backlog-Eintrag.
|
|
Ermoeglicht Tracking von Auto-Fix und manuellen Fixes.
|
|
"""
|
|
__tablename__ = 'test_fixes_history'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
backlog_id = Column(Integer, ForeignKey('failed_tests_backlog.id', ondelete='CASCADE'), nullable=False, index=True)
|
|
fix_type = Column(String(50), nullable=True) # manual, auto_claude, auto_script
|
|
fix_description = Column(Text, nullable=True)
|
|
commit_hash = Column(String(40), nullable=True)
|
|
success = Column(Boolean, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationship zum Backlog-Item
|
|
backlog_item = relationship("FailedTestBacklogDB", back_populates="fixes")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"backlog_id": self.backlog_id,
|
|
"fix_type": self.fix_type,
|
|
"fix_description": self.fix_description,
|
|
"commit_hash": self.commit_hash,
|
|
"success": self.success,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
}
|
|
|
|
|
|
class TestServiceStatsDB(Base):
|
|
"""
|
|
Aggregierte Statistiken pro Service.
|
|
Wird nach jedem Test-Run aktualisiert fuer schnelle Abfragen.
|
|
"""
|
|
__tablename__ = 'test_service_stats'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
service = Column(String(100), unique=True, nullable=False)
|
|
total_tests = Column(Integer, default=0)
|
|
passed_tests = Column(Integer, default=0)
|
|
failed_tests = Column(Integer, default=0)
|
|
skipped_tests = Column(Integer, default=0)
|
|
pass_rate = Column(Float, default=0.0)
|
|
last_run_id = Column(String(50), nullable=True)
|
|
last_run_at = Column(DateTime, nullable=True)
|
|
last_status = Column(String(20), nullable=True)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"service": self.service,
|
|
"total_tests": self.total_tests,
|
|
"passed_tests": self.passed_tests,
|
|
"failed_tests": self.failed_tests,
|
|
"skipped_tests": self.skipped_tests,
|
|
"pass_rate": round(self.pass_rate, 1) if self.pass_rate else 0.0,
|
|
"last_run_id": self.last_run_id,
|
|
"last_run_at": self.last_run_at.isoformat() if self.last_run_at else None,
|
|
"last_status": self.last_status,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|