This repository has been archived on 2026-02-15. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
breakpilot-pwa/backend/tests/test_dsms_webui.py
Benjamin Admin 21a844cb8a fix: Restore all files lost during destructive rebase
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>
2026-02-09 09:51:32 +01:00

377 lines
13 KiB
Python

"""
Unit Tests für DSMS WebUI Funktionalität
Tests für die WebUI-bezogenen Endpoints und Datenstrukturen
"""
import pytest
import json
import hashlib
from unittest.mock import AsyncMock, patch, MagicMock
from fastapi.testclient import TestClient
# ==================== DSMS WebUI API Response Tests ====================
class TestDsmsWebUINodeInfo:
"""Tests für die Node-Info API die vom WebUI verwendet wird"""
def test_node_info_response_structure(self):
"""Test: Node-Info Response hat alle WebUI-relevanten Felder"""
# Diese Struktur wird vom WebUI erwartet
expected_fields = [
"node_id",
"protocol_version",
"agent_version",
"repo_size",
"storage_max",
"num_objects",
"addresses"
]
# Mock response wie sie vom DSMS Gateway kommt
mock_response = {
"node_id": "QmTestNodeId123",
"protocol_version": "ipfs/0.1.0",
"agent_version": "kubo/0.24.0",
"repo_size": 1048576,
"storage_max": 10737418240,
"num_objects": 42,
"addresses": ["/ip4/127.0.0.1/tcp/4001"]
}
for field in expected_fields:
assert field in mock_response, f"Feld {field} fehlt in der Response"
def test_node_info_formats_repo_size(self):
"""Test: Repo-Größe wird korrekt formatiert"""
def format_bytes(size_bytes):
"""Formatiert Bytes in lesbare Einheiten (wie im WebUI)"""
if size_bytes is None:
return "N/A"
for unit in ['B', 'KB', 'MB', 'GB']:
if size_bytes < 1024:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024
return f"{size_bytes:.1f} TB"
assert format_bytes(1024) == "1.0 KB"
assert format_bytes(1048576) == "1.0 MB"
assert format_bytes(1073741824) == "1.0 GB"
assert format_bytes(None) == "N/A"
class TestDsmsWebUIDocumentList:
"""Tests für die Dokumentenlisten-API die vom WebUI verwendet wird"""
def test_document_list_response_structure(self):
"""Test: Document List Response hat alle WebUI-relevanten Felder"""
mock_response = {
"documents": [
{
"cid": "QmTestCid123",
"metadata": {
"document_type": "legal_document",
"document_id": "privacy-policy",
"version": "1.0",
"created_at": "2024-01-01T00:00:00"
},
"filename": "datenschutz.html"
}
],
"total": 1
}
assert "documents" in mock_response
assert "total" in mock_response
assert mock_response["total"] == len(mock_response["documents"])
doc = mock_response["documents"][0]
assert "cid" in doc
assert "metadata" in doc
def test_document_list_empty(self):
"""Test: Leere Dokumentenliste wird korrekt behandelt"""
mock_response = {
"documents": [],
"total": 0
}
assert mock_response["total"] == 0
assert len(mock_response["documents"]) == 0
class TestDsmsWebUIVerification:
"""Tests für die Verifizierungs-API die vom WebUI verwendet wird"""
def test_verify_response_valid_integrity(self):
"""Test: Verifizierungs-Response bei gültiger Integrität"""
content = "Test content"
checksum = hashlib.sha256(content.encode('utf-8')).hexdigest()
mock_response = {
"cid": "QmTestCid123",
"exists": True,
"integrity_valid": True,
"metadata": {
"document_type": "legal_document",
"checksum": checksum
},
"stored_checksum": checksum,
"calculated_checksum": checksum,
"verified_at": "2024-01-01T10:00:00"
}
assert mock_response["exists"] is True
assert mock_response["integrity_valid"] is True
assert mock_response["stored_checksum"] == mock_response["calculated_checksum"]
def test_verify_response_invalid_integrity(self):
"""Test: Verifizierungs-Response bei ungültiger Integrität"""
mock_response = {
"cid": "QmTestCid123",
"exists": True,
"integrity_valid": False,
"metadata": {
"document_type": "legal_document"
},
"stored_checksum": "fake_checksum",
"calculated_checksum": "real_checksum",
"verified_at": "2024-01-01T10:00:00"
}
assert mock_response["exists"] is True
assert mock_response["integrity_valid"] is False
assert mock_response["stored_checksum"] != mock_response["calculated_checksum"]
def test_verify_response_not_found(self):
"""Test: Verifizierungs-Response wenn Dokument nicht existiert"""
mock_response = {
"cid": "QmNonExistent",
"exists": False,
"error": "Dokument nicht gefunden",
"verified_at": "2024-01-01T10:00:00"
}
assert mock_response["exists"] is False
assert "error" in mock_response
class TestDsmsWebUIUpload:
"""Tests für die Upload-API die vom WebUI verwendet wird"""
def test_upload_response_structure(self):
"""Test: Upload Response hat alle WebUI-relevanten Felder"""
mock_response = {
"cid": "QmNewDocCid123",
"size": 1024,
"metadata": {
"document_type": "legal_document",
"document_id": None,
"version": None,
"language": "de",
"created_at": "2024-01-01T10:00:00",
"checksum": "abc123def456",
"encrypted": False
},
"gateway_url": "http://dsms-node:8080/ipfs/QmNewDocCid123",
"timestamp": "2024-01-01T10:00:00"
}
assert "cid" in mock_response
assert "size" in mock_response
assert "gateway_url" in mock_response
assert mock_response["cid"].startswith("Qm")
def test_checksum_calculation(self):
"""Test: Checksum wird korrekt berechnet (wie im Gateway)"""
content = b"Test document content"
expected_checksum = hashlib.sha256(content).hexdigest()
# Simuliert die Checksum-Berechnung wie im DSMS Gateway
calculated = hashlib.sha256(content).hexdigest()
assert calculated == expected_checksum
assert len(calculated) == 64 # SHA-256 hat immer 64 Hex-Zeichen
class TestDsmsWebUIHealthCheck:
"""Tests für den Health Check der vom WebUI verwendet wird"""
def test_health_response_online(self):
"""Test: Health Response wenn Node online ist"""
mock_response = {
"status": "healthy",
"ipfs_connected": True,
"timestamp": "2024-01-01T10:00:00"
}
assert mock_response["status"] == "healthy"
assert mock_response["ipfs_connected"] is True
def test_health_response_offline(self):
"""Test: Health Response wenn Node offline ist"""
mock_response = {
"status": "degraded",
"ipfs_connected": False,
"timestamp": "2024-01-01T10:00:00"
}
assert mock_response["status"] == "degraded"
assert mock_response["ipfs_connected"] is False
class TestDsmsWebUIDataTransformation:
"""Tests für Daten-Transformationen die das WebUI durchführt"""
def test_format_timestamp(self):
"""Test: ISO-Timestamp wird für Anzeige formatiert"""
def format_timestamp(iso_string):
"""Formatiert ISO-Timestamp für die Anzeige"""
from datetime import datetime
try:
dt = datetime.fromisoformat(iso_string.replace('Z', '+00:00'))
return dt.strftime("%d.%m.%Y %H:%M")
except:
return iso_string
assert format_timestamp("2024-01-15T10:30:00") == "15.01.2024 10:30"
assert format_timestamp("invalid") == "invalid"
def test_truncate_cid(self):
"""Test: CID wird für Anzeige gekürzt"""
def truncate_cid(cid, max_length=20):
"""Kürzt CID für die Anzeige"""
if len(cid) <= max_length:
return cid
return cid[:8] + "..." + cid[-8:]
long_cid = "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG"
truncated = truncate_cid(long_cid)
assert len(truncated) < len(long_cid)
assert truncated.startswith("Qm")
assert "..." in truncated
def test_status_badge_class(self):
"""Test: Status-Badge-Klasse wird korrekt ermittelt"""
def get_status_badge_class(status):
"""Gibt die CSS-Klasse für den Status zurück"""
status_classes = {
"healthy": "success",
"degraded": "warning",
"offline": "danger",
True: "success",
False: "danger"
}
return status_classes.get(status, "secondary")
assert get_status_badge_class("healthy") == "success"
assert get_status_badge_class("degraded") == "warning"
assert get_status_badge_class(True) == "success"
assert get_status_badge_class(False) == "danger"
assert get_status_badge_class("unknown") == "secondary"
class TestDsmsWebUIErrorHandling:
"""Tests für die Fehlerbehandlung im WebUI"""
def test_network_error_message(self):
"""Test: Netzwerkfehler wird benutzerfreundlich angezeigt"""
def get_error_message(error_type, details=None):
"""Gibt benutzerfreundliche Fehlermeldung zurück"""
messages = {
"network": "DSMS Node ist nicht erreichbar. Bitte überprüfen Sie die Verbindung.",
"auth": "Authentifizierung fehlgeschlagen. Bitte erneut anmelden.",
"not_found": "Dokument nicht gefunden.",
"upload": f"Upload fehlgeschlagen: {details}" if details else "Upload fehlgeschlagen.",
"unknown": "Ein unbekannter Fehler ist aufgetreten."
}
return messages.get(error_type, messages["unknown"])
assert "nicht erreichbar" in get_error_message("network")
assert "nicht gefunden" in get_error_message("not_found")
assert "Test Error" in get_error_message("upload", "Test Error")
def test_validation_cid_format(self):
"""Test: CID-Format wird validiert"""
def is_valid_cid(cid):
"""Prüft ob CID ein gültiges Format hat"""
if not cid:
return False
# CIDv0 beginnt mit Qm und hat 46 Zeichen
if cid.startswith("Qm") and len(cid) == 46:
return True
# CIDv1 beginnt mit b und ist base32 encoded
if cid.startswith("b") and len(cid) > 40:
return True
return False
assert is_valid_cid("QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG") is True
assert is_valid_cid("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") is True
assert is_valid_cid("invalid") is False
assert is_valid_cid("") is False
assert is_valid_cid(None) is False
class TestDsmsWebUIIntegration:
"""Integrationstests für WebUI-Workflows"""
def test_upload_and_verify_workflow(self):
"""Test: Upload und anschließende Verifizierung"""
# Simuliert den Upload-Workflow
upload_content = b"Test document for verification"
expected_checksum = hashlib.sha256(upload_content).hexdigest()
# Upload Response
upload_response = {
"cid": "QmNewDoc123456789012345678901234567890123456",
"size": len(upload_content),
"metadata": {
"checksum": expected_checksum
}
}
# Verifizierung
verify_response = {
"cid": upload_response["cid"],
"exists": True,
"integrity_valid": True,
"stored_checksum": expected_checksum,
"calculated_checksum": expected_checksum
}
assert verify_response["integrity_valid"] is True
assert verify_response["stored_checksum"] == upload_response["metadata"]["checksum"]
def test_node_status_determines_ui_state(self):
"""Test: Node-Status bestimmt UI-Zustand"""
def get_ui_state(health_response):
"""Ermittelt UI-Zustand basierend auf Health Check"""
if health_response.get("ipfs_connected"):
return {
"status": "online",
"upload_enabled": True,
"explore_enabled": True,
"message": None
}
else:
return {
"status": "offline",
"upload_enabled": False,
"explore_enabled": False,
"message": "DSMS Node ist nicht erreichbar"
}
online_state = get_ui_state({"ipfs_connected": True, "status": "healthy"})
assert online_state["upload_enabled"] is True
offline_state = get_ui_state({"ipfs_connected": False, "status": "degraded"})
assert offline_state["upload_enabled"] is False
assert offline_state["message"] is not None
# ==================== Run Tests ====================
if __name__ == "__main__":
pytest.main([__file__, "-v"])