Phase 3.1: Integration Tests mit Datenbank implementiert

26 Integration Tests hinzugefügt für Service-Layer mit DB:

 ProgressService (12 Tests)
- get_or_create_global_progress (creates/returns existing)
- get/set_current_lesson_index
- get_or_create_lesson_progress (creates/returns existing)
- update_lesson_progress
- reset_lesson_progress
- get_lesson_progress_data
- complete_lesson
- multiple_lessons_independent

 StatisticsService (14 Tests)
- save_lesson_statistics
- has_lesson_statistics
- get_lesson_statistics_data
- delete_lesson_statistics (single/preserve others)
- get_all_statistics
- update_daily_practice (creates/accumulates)
- get_full_30_days_practice (with gaps filled)
- daily_practice_separate_days
- end_to_end_statistic_save

Test-Setup:
- In-Memory SQLite DB für isolation
- Flask app_context für DB-Operationen
- setUp/tearDown für DB-Lifecycle
- os.environ['FLASK_TESTING'] für Test-Modus

Änderung app.py:
- DB-Check nur wenn nicht FLASK_TESTING=true
- os import hinzugefügt
- Ermöglicht Tests ohne Produktions-DB

Alle 26 Integration Tests bestehen 
Ausführung: python -m unittest tests.test_integration_*
This commit is contained in:
jamulix 2025-10-29 09:41:17 +01:00
commit 94819bee71
3 changed files with 578 additions and 15 deletions

32
app.py
View file

@ -4,6 +4,7 @@ Modularized version with Blueprints
"""
import logging
import os
from flask import Flask
from flask_wtf.csrf import CSRFProtect
from models import db
@ -61,25 +62,26 @@ app.register_blueprint(utils_bp)
csrf.exempt(utils_bp)
# Datenbank-Status prüfen und Lektionen laden
with app.app_context():
db_ready = helpers.check_database_status(db.engine)
# Datenbank-Status prüfen und Lektionen laden (nur wenn nicht im Test-Modus)
if not app.config.get('TESTING', False) and os.environ.get('FLASK_TESTING') != 'true':
with app.app_context():
db_ready = helpers.check_database_status(db.engine)
if not db_ready:
logger.error("Anwendung kann nicht gestartet werden - Datenbank nicht bereit")
logger.error("Siehe Warnungen oben für Lösungsschritte")
import sys
sys.exit(1)
if not db_ready:
logger.error("Anwendung kann nicht gestartet werden - Datenbank nicht bereit")
logger.error("Siehe Warnungen oben für Lösungsschritte")
import sys
sys.exit(1)
# Versuche Lektionen zu laden
helpers.load_lessons()
# Versuche Lektionen zu laden
helpers.load_lessons()
# Falls keine Lektionen geladen wurden, erstelle Beispieldaten
if not helpers.lessons:
logger.warning("Keine Lektionen verfügbar, erstelle Beispiel-Lektionen...")
helpers.lessons = helpers.create_sample_lessons()
# Falls keine Lektionen geladen wurden, erstelle Beispieldaten
if not helpers.lessons:
logger.warning("Keine Lektionen verfügbar, erstelle Beispiel-Lektionen...")
helpers.lessons = helpers.create_sample_lessons()
logger.info(f"{len(helpers.lessons)} Lektionen geladen")
logger.info(f"{len(helpers.lessons)} Lektionen geladen")
if __name__ == '__main__':

View file

@ -0,0 +1,259 @@
"""
Integration Tests für ProgressService mit Datenbank
"""
import unittest
import os
os.environ['FLASK_TESTING'] = 'true' # Set before importing app
from app import app, db
from models import Progress, LessonProgress
from services.progress_service import ProgressService
class TestProgressServiceIntegration(unittest.TestCase):
"""Integration Tests für ProgressService mit echter Datenbank"""
def setUp(self):
"""Setup: Test-App und In-Memory-Datenbank"""
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
app.config['WTF_CSRF_ENABLED'] = False
self.app = app
self.app_context = self.app.app_context()
self.app_context.push()
db.create_all()
def tearDown(self):
"""Cleanup: Datenbank löschen"""
db.session.remove()
db.drop_all()
self.app_context.pop()
# ========================
# Global Progress Tests
# ========================
def test_get_or_create_global_progress_creates_new(self):
"""Test: Erstellt neuen GlobalProgress wenn keiner existiert"""
# Sicherstellen, dass DB leer ist
self.assertIsNone(Progress.query.first())
progress = ProgressService.get_or_create_global_progress()
# Sollte neuen Eintrag erstellt haben
self.assertIsNotNone(progress)
self.assertEqual(progress.current_text_index, 0)
self.assertEqual(progress.current_position, 0)
self.assertEqual(Progress.query.count(), 1)
def test_get_or_create_global_progress_returns_existing(self):
"""Test: Gibt existierenden GlobalProgress zurück"""
# Erstelle existierenden Eintrag
existing = Progress(current_text_index=5, current_position=100)
db.session.add(existing)
db.session.commit()
progress = ProgressService.get_or_create_global_progress()
# Sollte existierenden Eintrag zurückgeben
self.assertEqual(progress.id, existing.id)
self.assertEqual(progress.current_text_index, 5)
self.assertEqual(Progress.query.count(), 1) # Kein neuer Eintrag
def test_get_current_lesson_index(self):
"""Test: Liest aktuellen Lektionsindex"""
# Erstelle Progress mit Lektion 3
progress = Progress(current_text_index=3)
db.session.add(progress)
db.session.commit()
index = ProgressService.get_current_lesson_index()
self.assertEqual(index, 3)
def test_set_current_lesson_index(self):
"""Test: Setzt aktuellen Lektionsindex"""
# Erstelle initialen Progress
ProgressService.get_or_create_global_progress()
ProgressService.set_current_lesson_index(7)
# Verifiziere Änderung
progress = Progress.query.first()
self.assertEqual(progress.current_text_index, 7)
# ========================
# Lesson Progress Tests
# ========================
def test_get_or_create_lesson_progress_creates_new(self):
"""Test: Erstellt neuen LessonProgress"""
self.assertIsNone(LessonProgress.query.first())
lesson_progress = ProgressService.get_or_create_lesson_progress(5)
self.assertIsNotNone(lesson_progress)
self.assertEqual(lesson_progress.lesson_index, 5)
self.assertEqual(lesson_progress.current_position, 0)
self.assertEqual(lesson_progress.last_text, "")
self.assertTrue(lesson_progress.is_paused)
self.assertEqual(LessonProgress.query.count(), 1)
def test_get_or_create_lesson_progress_returns_existing(self):
"""Test: Gibt existierenden LessonProgress zurück"""
# Erstelle existierenden Eintrag
existing = LessonProgress(
lesson_index=2,
current_position=50,
last_text="test input"
)
db.session.add(existing)
db.session.commit()
lesson_progress = ProgressService.get_or_create_lesson_progress(2)
self.assertEqual(lesson_progress.id, existing.id)
self.assertEqual(lesson_progress.current_position, 50)
self.assertEqual(lesson_progress.last_text, "test input")
self.assertEqual(LessonProgress.query.count(), 1)
def test_update_lesson_progress(self):
"""Test: Aktualisiert LessonProgress"""
# Erstelle initialen Progress
ProgressService.get_or_create_lesson_progress(3)
ProgressService.update_lesson_progress(
lesson_index=3,
cursor_position=100,
user_input="hello world",
total_elapsed_time=5000, # 5 Sekunden in ms
key_stroke_count=25,
is_paused=False
)
# Verifiziere Änderungen
lesson_progress = LessonProgress.query.filter_by(lesson_index=3).first()
self.assertEqual(lesson_progress.current_position, 100)
self.assertEqual(lesson_progress.last_text, "hello world")
self.assertEqual(lesson_progress.total_elapsed_time, 5000)
self.assertEqual(lesson_progress.key_stroke_count, 25)
self.assertFalse(lesson_progress.is_paused)
def test_reset_lesson_progress(self):
"""Test: Setzt LessonProgress zurück"""
# Erstelle Progress mit Daten
lesson_progress = LessonProgress(
lesson_index=4,
current_position=200,
last_text="some text",
total_elapsed_time=10000,
key_stroke_count=50,
is_paused=False
)
db.session.add(lesson_progress)
db.session.commit()
ProgressService.reset_lesson_progress(4)
# Verifiziere Reset
lesson_progress = LessonProgress.query.filter_by(lesson_index=4).first()
self.assertEqual(lesson_progress.current_position, 0)
self.assertEqual(lesson_progress.last_text, "")
self.assertEqual(lesson_progress.total_elapsed_time, 0)
self.assertEqual(lesson_progress.key_stroke_count, 0)
self.assertTrue(lesson_progress.is_paused)
def test_get_lesson_progress_data(self):
"""Test: Holt LessonProgress-Daten als Dict"""
# Erstelle Progress
lesson_progress = LessonProgress(
lesson_index=1,
current_position=75,
last_text="test",
total_elapsed_time=3000,
key_stroke_count=15,
is_paused=True
)
db.session.add(lesson_progress)
db.session.commit()
data = ProgressService.get_lesson_progress_data(1)
self.assertEqual(data['current_position'], 75)
self.assertEqual(data['last_text'], "test")
self.assertEqual(data['total_elapsed_time'], 3000)
self.assertEqual(data['key_stroke_count'], 15)
self.assertTrue(data['is_paused'])
def test_get_lesson_progress_data_creates_if_not_exists(self):
"""Test: Erstellt neuen LessonProgress wenn nicht vorhanden"""
# DB ist leer
self.assertIsNone(LessonProgress.query.filter_by(lesson_index=99).first())
data = ProgressService.get_lesson_progress_data(99)
# Sollte neuen Eintrag erstellt haben mit Default-Werten
self.assertEqual(data['current_position'], 0)
self.assertEqual(data['last_text'], "")
self.assertEqual(data['total_elapsed_time'], 0)
self.assertEqual(data['key_stroke_count'], 0)
self.assertTrue(data['is_paused'])
# Verifiziere DB-Eintrag
lesson_progress = LessonProgress.query.filter_by(lesson_index=99).first()
self.assertIsNotNone(lesson_progress)
def test_complete_lesson(self):
"""Test: Schließt Lektion ab (reset)"""
# Erstelle aktiven Progress
lesson_progress = LessonProgress(
lesson_index=6,
current_position=300,
last_text="completed text",
total_elapsed_time=20000,
key_stroke_count=100,
is_paused=False
)
db.session.add(lesson_progress)
db.session.commit()
ProgressService.complete_lesson(6)
# Sollte zurückgesetzt sein
lesson_progress = LessonProgress.query.filter_by(lesson_index=6).first()
self.assertEqual(lesson_progress.current_position, 0)
self.assertEqual(lesson_progress.last_text, "")
self.assertEqual(lesson_progress.total_elapsed_time, 0)
self.assertEqual(lesson_progress.key_stroke_count, 0)
self.assertTrue(lesson_progress.is_paused)
def test_multiple_lessons_independent(self):
"""Test: Mehrere Lektionen sind unabhängig"""
# Erstelle Progress für mehrere Lektionen
ProgressService.update_lesson_progress(1, 10, "text1", 1000, 5, False)
ProgressService.update_lesson_progress(2, 20, "text2", 2000, 10, False)
ProgressService.update_lesson_progress(3, 30, "text3", 3000, 15, False)
# Verifiziere dass alle existieren und unabhängig sind
lesson1 = LessonProgress.query.filter_by(lesson_index=1).first()
lesson2 = LessonProgress.query.filter_by(lesson_index=2).first()
lesson3 = LessonProgress.query.filter_by(lesson_index=3).first()
self.assertEqual(lesson1.current_position, 10)
self.assertEqual(lesson2.current_position, 20)
self.assertEqual(lesson3.current_position, 30)
# Reset einer Lektion sollte andere nicht beeinflussen
ProgressService.reset_lesson_progress(2)
lesson1 = LessonProgress.query.filter_by(lesson_index=1).first()
lesson2 = LessonProgress.query.filter_by(lesson_index=2).first()
lesson3 = LessonProgress.query.filter_by(lesson_index=3).first()
self.assertEqual(lesson1.current_position, 10) # Unverändert
self.assertEqual(lesson2.current_position, 0) # Zurückgesetzt
self.assertEqual(lesson3.current_position, 30) # Unverändert
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,302 @@
"""
Integration Tests für StatisticsService mit Datenbank
"""
import unittest
import os
os.environ['FLASK_TESTING'] = 'true' # Set before importing app
from datetime import datetime, timedelta, date
from app import app, db
from models import LessonStatistic, DailyPractice
from services.statistics_service import StatisticsService
class TestStatisticsServiceIntegration(unittest.TestCase):
"""Integration Tests für StatisticsService mit echter Datenbank"""
def setUp(self):
"""Setup: Test-App und In-Memory-Datenbank"""
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
app.config['WTF_CSRF_ENABLED'] = False
self.app = app
self.app_context = self.app.app_context()
self.app_context.push()
db.create_all()
def tearDown(self):
"""Cleanup: Datenbank löschen"""
db.session.remove()
db.drop_all()
self.app_context.pop()
# ========================
# Lesson Statistics Tests
# ========================
def test_save_lesson_statistics(self):
"""Test: Speichert Lektionsstatistiken"""
stat = StatisticsService.save_lesson_statistics(
lesson_index=1,
chars_per_minute=120.5,
error_rate=5.2,
wpm=24.1,
training_duration=300.0
)
self.assertIsNotNone(stat)
self.assertEqual(stat.lesson_index, 1)
self.assertEqual(stat.chars_per_minute, 120.5)
self.assertEqual(stat.error_rate, 5.2)
self.assertEqual(stat.wpm, 24.1)
self.assertEqual(stat.training_duration, 300.0)
# Verifiziere DB-Eintrag
db_stat = LessonStatistic.query.first()
self.assertIsNotNone(db_stat)
self.assertEqual(db_stat.lesson_index, 1)
def test_has_lesson_statistics_true(self):
"""Test: Prüft ob Statistiken für Lektion existieren (True)"""
# Erstelle Statistik
stat = LessonStatistic(
lesson_index=2,
chars_per_minute=100.0,
error_rate=3.0,
wpm=20.0,
training_duration=200.0
)
db.session.add(stat)
db.session.commit()
has_stats = StatisticsService.has_lesson_statistics(2)
self.assertTrue(has_stats)
def test_has_lesson_statistics_false(self):
"""Test: Prüft ob Statistiken für Lektion existieren (False)"""
# DB ist leer
has_stats = StatisticsService.has_lesson_statistics(99)
self.assertFalse(has_stats)
def test_get_lesson_statistics_data(self):
"""Test: Holt Statistiken für eine Lektion"""
# Erstelle mehrere Statistiken für eine Lektion
for i in range(3):
stat = LessonStatistic(
lesson_index=3,
chars_per_minute=100.0 + i * 10,
error_rate=5.0 - i,
wpm=20.0 + i * 2,
training_duration=180.0
)
db.session.add(stat)
db.session.commit()
data = StatisticsService.get_lesson_statistics_data(3)
self.assertEqual(len(data['dates']), 3)
self.assertEqual(len(data['chars_per_minute']), 3)
self.assertEqual(len(data['error_rate']), 3)
self.assertEqual(len(data['wpm']), 3)
# Älteste zuerst (aufsteigende Reihenfolge nach created_at)
self.assertEqual(data['chars_per_minute'][0], 100.0)
self.assertEqual(data['chars_per_minute'][-1], 120.0)
self.assertEqual(data['error_rate'][0], 5.0)
self.assertEqual(data['error_rate'][-1], 3.0)
def test_get_lesson_statistics_data_empty(self):
"""Test: Holt Statistiken für Lektion ohne Daten"""
data = StatisticsService.get_lesson_statistics_data(99)
self.assertEqual(len(data['dates']), 0)
self.assertEqual(len(data['chars_per_minute']), 0)
self.assertEqual(len(data['error_rate']), 0)
self.assertEqual(len(data['wpm']), 0)
def test_delete_lesson_statistics(self):
"""Test: Löscht alle Statistiken für eine Lektion"""
# Erstelle mehrere Statistiken
for i in range(5):
stat = LessonStatistic(
lesson_index=4,
chars_per_minute=100.0,
error_rate=5.0,
wpm=20.0,
training_duration=180.0
)
db.session.add(stat)
db.session.commit()
# Verifiziere dass sie existieren
self.assertEqual(LessonStatistic.query.filter_by(lesson_index=4).count(), 5)
deleted_count = StatisticsService.delete_lesson_statistics(4)
self.assertEqual(deleted_count, 5)
self.assertEqual(LessonStatistic.query.filter_by(lesson_index=4).count(), 0)
def test_delete_lesson_statistics_preserves_others(self):
"""Test: Löschen einer Lektion lässt andere Lektionen intakt"""
# Erstelle Statistiken für mehrere Lektionen
for lesson in [1, 2, 3]:
for i in range(2):
stat = LessonStatistic(
lesson_index=lesson,
chars_per_minute=100.0,
error_rate=5.0,
wpm=20.0,
training_duration=180.0
)
db.session.add(stat)
db.session.commit()
# Lösche nur Lektion 2
StatisticsService.delete_lesson_statistics(2)
# Lektion 1 und 3 sollten noch existieren
self.assertEqual(LessonStatistic.query.filter_by(lesson_index=1).count(), 2)
self.assertEqual(LessonStatistic.query.filter_by(lesson_index=2).count(), 0)
self.assertEqual(LessonStatistic.query.filter_by(lesson_index=3).count(), 2)
def test_get_all_statistics(self):
"""Test: Holt alle Statistiken"""
# Erstelle Statistiken für verschiedene Lektionen
for lesson in range(3):
stat = LessonStatistic(
lesson_index=lesson,
chars_per_minute=100.0 + lesson * 10,
error_rate=5.0,
wpm=20.0,
training_duration=180.0
)
db.session.add(stat)
db.session.commit()
all_stats = StatisticsService.get_all_statistics()
self.assertEqual(len(all_stats), 3)
# Sollte nach created_at DESC sortiert sein
self.assertIsInstance(all_stats[0], LessonStatistic)
# ========================
# Daily Practice Tests
# ========================
def test_update_daily_practice_creates_new(self):
"""Test: Erstellt neuen DailyPractice-Eintrag"""
today = date.today()
StatisticsService.update_daily_practice(15.5) # 15.5 Minuten
# Verifiziere DB-Eintrag
practice = DailyPractice.query.filter_by(date=today).first()
self.assertIsNotNone(practice)
self.assertEqual(practice.minutes_practiced, 15.5)
def test_update_daily_practice_accumulates(self):
"""Test: Addiert zu existierendem DailyPractice"""
today = date.today()
# Erste Session: 10 Minuten
StatisticsService.update_daily_practice(10.0)
# Zweite Session: 5 Minuten
StatisticsService.update_daily_practice(5.0)
# Sollte addiert haben
practice = DailyPractice.query.filter_by(date=today).first()
self.assertEqual(practice.minutes_practiced, 15.0)
def test_get_full_30_days_practice(self):
"""Test: Holt 30 Tage Übungsdaten (mit Lücken gefüllt)"""
today = date.today()
# Erstelle Einträge für heute und vor 5 Tagen
practice_today = DailyPractice(date=today, minutes_practiced=20.0)
practice_5_days_ago = DailyPractice(
date=today - timedelta(days=5),
minutes_practiced=15.0
)
db.session.add(practice_today)
db.session.add(practice_5_days_ago)
db.session.commit()
full_data = StatisticsService.get_full_30_days_practice()
# Sollte 30 Einträge haben (mit 0 für Tage ohne Übung)
self.assertEqual(len(full_data), 30)
# Heute sollte 20 Minuten haben
self.assertEqual(full_data[-1]['minutes'], 20.0)
self.assertEqual(full_data[-1]['date'], today.strftime('%Y-%m-%d'))
# Vor 5 Tagen sollte 15 Minuten haben
self.assertEqual(full_data[-6]['minutes'], 15.0)
# Andere Tage sollten 0 haben
self.assertEqual(full_data[-2]['minutes'], 0.0) # Gestern
self.assertEqual(full_data[-3]['minutes'], 0.0) # Vorgestern
def test_get_full_30_days_practice_all_zeros(self):
"""Test: 30 Tage ohne Übung gibt nur Nullen"""
full_data = StatisticsService.get_full_30_days_practice()
self.assertEqual(len(full_data), 30)
# Alle sollten 0 Minuten haben
for day in full_data:
self.assertEqual(day['minutes'], 0.0)
def test_daily_practice_separate_days(self):
"""Test: Verschiedene Tage werden separat gespeichert"""
today = date.today()
yesterday = today - timedelta(days=1)
StatisticsService.update_daily_practice(10.0) # Heute
# Simuliere gestern durch direkten DB-Eintrag
practice_yesterday = DailyPractice(date=yesterday, minutes_practiced=20.0)
db.session.add(practice_yesterday)
db.session.commit()
# Verifiziere separate Einträge
practice_today = DailyPractice.query.filter_by(date=today).first()
practice_yesterday_db = DailyPractice.query.filter_by(date=yesterday).first()
self.assertEqual(practice_today.minutes_practiced, 10.0)
self.assertEqual(practice_yesterday_db.minutes_practiced, 20.0)
# ========================
# Statistics Calculations (Integration)
# ========================
def test_end_to_end_statistic_save(self):
"""Test: Kompletter Workflow - Berechnen und Speichern"""
# Berechne Statistiken
stats = StatisticsService.calculate_typing_statistics(
user_input="hello world",
current_text="hello world",
elapsed_time=10.0
)
# Speichere in DB
saved_stat = StatisticsService.save_lesson_statistics(
lesson_index=5,
chars_per_minute=stats['typing_speed'],
error_rate=stats['error_rate'],
wpm=stats['words_per_minute'],
training_duration=10.0
)
# Verifiziere dass gespeicherte Daten abrufbar sind
has_stats = StatisticsService.has_lesson_statistics(5)
self.assertTrue(has_stats)
data = StatisticsService.get_lesson_statistics_data(5)
self.assertEqual(len(data['dates']), 1)
self.assertEqual(data['error_rate'][0], 0.0) # Perfekte Eingabe
if __name__ == '__main__':
unittest.main()