feat: Add Excel, PDF export, and user generation modules

- Implemented `excel_export.py` for generating library item exports in Excel format.
- Created `pdf_export.py` for generating audit reports compliant with DIN 5008 standards, including detailed event tables and signature blocks.
- Developed `generate_user.py` for interactive user creation with validation for usernames and passwords.
- Introduced `module_registry.py` for managing module states and path matching.
- Added a basic `__init__.py` in the `terminplaner` module for initialization.
This commit is contained in:
2026-05-20 15:30:23 +02:00
parent b3fc470c88
commit 0c27d7ac86
42 changed files with 122 additions and 1055 deletions
+30 -1
View File
@@ -150,11 +150,25 @@ jobs:
${{ steps.meta.outputs.image }} ${{ steps.meta.outputs.image }}
ghcr.io/aiirondev/legendary-octo-garbanzo:latest ghcr.io/aiirondev/legendary-octo-garbanzo:latest
- name: Build and push development image
uses: docker/build-push-action@v6
with:
context: .
file: ./Dockerfile
push: true
tags: |
ghcr.io/aiirondev/legendary-octo-garbanzo:development
- name: Build local image tar for offline deploy - name: Build local image tar for offline deploy
run: | run: |
docker pull "${{ steps.meta.outputs.image }}" docker pull "${{ steps.meta.outputs.image }}"
docker save "${{ steps.meta.outputs.image }}" | gzip > "inventarsystem-image-${{ steps.meta.outputs.tag }}.tar.gz" docker save "${{ steps.meta.outputs.image }}" | gzip > "inventarsystem-image-${{ steps.meta.outputs.tag }}.tar.gz"
- name: Build local development image tar for offline deploy
run: |
docker pull ghcr.io/aiirondev/legendary-octo-garbanzo:development || true
docker save ghcr.io/aiirondev/legendary-octo-garbanzo:development | gzip > "inventarsystem-image-development.tar.gz" || true
- name: Create release-only docker bundle - name: Create release-only docker bundle
run: | run: |
mkdir -p release-bundle mkdir -p release-bundle
@@ -238,6 +252,20 @@ jobs:
cp MULTITENANT_DEPLOYMENT.md release-bundle/MULTITENANT_DEPLOYMENT.md cp MULTITENANT_DEPLOYMENT.md release-bundle/MULTITENANT_DEPLOYMENT.md
cp MULTITENANT_PYTHON_API.md release-bundle/MULTITENANT_PYTHON_API.md cp MULTITENANT_PYTHON_API.md release-bundle/MULTITENANT_PYTHON_API.md
# Include optional development image tar and helper note
if [ -f "inventarsystem-image-development.tar.gz" ]; then
cp inventarsystem-image-development.tar.gz release-bundle/ || true
fi
cat > release-bundle/DEVELOPMENT.md <<'EOF'
This bundle contains an optional development image tar: inventarsystem-image-development.tar.gz
To install the development build on a target host, extract the bundle and run:
./update.sh development
EOF
chmod +x release-bundle/start.sh release-bundle/stop.sh release-bundle/restart.sh release-bundle/backup.sh release-bundle/update.sh release-bundle/manage-tenant.sh release-bundle/run-tenant-cmd.sh chmod +x release-bundle/start.sh release-bundle/stop.sh release-bundle/restart.sh release-bundle/backup.sh release-bundle/update.sh release-bundle/manage-tenant.sh release-bundle/run-tenant-cmd.sh
tar -czf inventarsystem-docker-bundle.tar.gz -C release-bundle . tar -czf inventarsystem-docker-bundle.tar.gz -C release-bundle .
@@ -248,5 +276,6 @@ jobs:
files: | files: |
inventarsystem-docker-bundle.tar.gz inventarsystem-docker-bundle.tar.gz
inventarsystem-image-${{ steps.meta.outputs.tag }}.tar.gz inventarsystem-image-${{ steps.meta.outputs.tag }}.tar.gz
fail_on_unmatched_files: true inventarsystem-image-development.tar.gz
fail_on_unmatched_files: false
generate_release_notes: true generate_release_notes: true
+10 -10
View File
@@ -29,12 +29,12 @@ from flask import Flask, render_template, request, redirect, url_for, session, f
from werkzeug.utils import secure_filename from werkzeug.utils import secure_filename
from werkzeug.middleware.proxy_fix import ProxyFix from werkzeug.middleware.proxy_fix import ProxyFix
from jinja2 import TemplateNotFound from jinja2 import TemplateNotFound
import user as us import Web.modules.database.user as us
import items as it import Web.modules.database.items as it
import ausleihung as au import Web.modules.database.ausleihung as au
import audit_log as al import Web.modules.logs.audit_log as al
import push_notifications as pn import push_notifications as pn
import pdf_export import Web.modules.inventarsystem.pdf_export as pdf_export
import datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
from bson.objectid import ObjectId from bson.objectid import ObjectId
@@ -69,7 +69,7 @@ import uuid
from PIL import Image, ImageOps from PIL import Image, ImageOps
import mimetypes import mimetypes
import subprocess import subprocess
from data_protection import ( from Web.modules.inventarsystem.data_protection import (
decrypt_document_fields, decrypt_document_fields,
encrypt_document_fields, encrypt_document_fields,
encrypt_soft_deleted_media_pack, encrypt_soft_deleted_media_pack,
@@ -77,8 +77,8 @@ from data_protection import (
# Set base directory and centralized settings # Set base directory and centralized settings
BASE_DIR = os.path.dirname(os.path.abspath(__file__)) BASE_DIR = os.path.dirname(os.path.abspath(__file__))
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
from tenant import get_tenant_context from tenant import get_tenant_context
@@ -2724,7 +2724,7 @@ def library_export_excel(scope):
username = session['username'] username = session['username']
is_admin_user = us.check_admin(username) is_admin_user = us.check_admin(username)
import excel_export import Web.modules.inventarsystem.excel_export as excel_export
client = MongoClient(MONGODB_HOST, MONGODB_PORT) client = MongoClient(MONGODB_HOST, MONGODB_PORT)
db = client[cfg.MONGODB_DB] db = client[cfg.MONGODB_DB]
items_collection = db['items'] items_collection = db['items']
@@ -10104,7 +10104,7 @@ def reset_item(id):
try: try:
# Import the ausleihung module # Import the ausleihung module
import ausleihung as au import Web.modules.database.ausleihung as au
result = au.reset_item_completely(id) result = au.reset_item_completely(id)
-149
View File
@@ -1,149 +0,0 @@
"""
Tamper-evident audit logging helpers.
The audit chain stores each entry with a hash of the previous entry.
Any mutation in history breaks the chain verification.
"""
import datetime
import hashlib
import json
import random
import time
from pymongo.errors import DuplicateKeyError
def _stable_json(value):
"""Serialize dictionaries in a stable way for deterministic hashing."""
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str)
def _entry_hash(prev_hash, payload):
"""Build the chained entry hash from previous hash + canonical payload."""
base = f"{prev_hash}|{_stable_json(payload)}"
return hashlib.sha256(base.encode("utf-8")).hexdigest()
def append_audit_event(db, event_type, actor, payload, request_ip=None, source="web", max_retries=5):
"""
Append an audit event to a tamper-evident chain.
Args:
db: MongoDB database handle.
event_type (str): Event category.
actor (str): User/system who performed the action.
payload (dict): Event details.
request_ip (str, optional): Request origin.
source (str): Source subsystem.
Returns:
dict: Inserted audit entry.
"""
logs = db["audit_log"]
attempts = 0
while attempts <= max_retries:
previous = logs.find_one(sort=[("chain_index", -1)])
prev_hash = previous.get("entry_hash", "") if previous else ""
chain_index = int(previous.get("chain_index", 0)) + 1 if previous else 1
timestamp = datetime.datetime.utcnow()
entry_payload = {
"event_type": event_type,
"actor": actor or "system",
"source": source,
"ip": request_ip or "",
"payload": payload or {},
"timestamp": timestamp.isoformat() + "Z",
}
entry_hash = _entry_hash(prev_hash, entry_payload)
entry = {
**entry_payload,
"created_at": timestamp,
"prev_hash": prev_hash,
"entry_hash": entry_hash,
"chain_index": chain_index,
}
try:
logs.insert_one(entry)
return entry
except DuplicateKeyError:
attempts += 1
if attempts > max_retries:
raise
# Exponential backoff with jitter to avoid retry storms.
delay = min(0.25, (0.005 * (2 ** attempts)) + random.random() * 0.01)
time.sleep(delay)
def ensure_audit_indexes(db):
"""Create indexes required for fast and safe audit operations."""
logs = db["audit_log"]
logs.create_index("chain_index", unique=True, name="audit_chain_index_unique")
logs.create_index("created_at", name="audit_created_at_idx")
logs.create_index("event_type", name="audit_event_type_idx")
def verify_audit_chain(db):
"""Verify hash chain integrity across all stored audit entries."""
logs = db["audit_log"]
entries = list(logs.find({}, {"_id": 1, "event_type": 1, "actor": 1, "source": 1, "ip": 1, "payload": 1, "timestamp": 1, "prev_hash": 1, "entry_hash": 1, "chain_index": 1}).sort("chain_index", 1))
previous_hash = ""
previous_index = 0
mismatches = []
for entry in entries:
chain_index = int(entry.get("chain_index", 0))
prev_hash = entry.get("prev_hash", "")
entry_hash = entry.get("entry_hash", "")
payload = {
"event_type": entry.get("event_type", ""),
"actor": entry.get("actor", ""),
"source": entry.get("source", ""),
"ip": entry.get("ip", ""),
"payload": entry.get("payload", {}),
"timestamp": entry.get("timestamp", ""),
}
expected_hash = _entry_hash(previous_hash, payload)
if chain_index != previous_index + 1:
mismatches.append({
"chain_index": chain_index,
"error": "chain_index_gap",
"expected": previous_index + 1,
"found": chain_index,
})
if prev_hash != previous_hash:
mismatches.append({
"chain_index": chain_index,
"error": "prev_hash_mismatch",
"expected": previous_hash,
"found": prev_hash,
})
if entry_hash != expected_hash:
mismatches.append({
"chain_index": chain_index,
"error": "entry_hash_mismatch",
"expected": expected_hash,
"found": entry_hash,
})
previous_hash = entry_hash
previous_index = chain_index
return {
"ok": len(mismatches) == 0,
"count": len(entries),
"last_chain_index": previous_index,
"last_hash": previous_hash,
"mismatches": mismatches,
}
-45
View File
@@ -1,45 +0,0 @@
'''
Copyright 2025-2026 AIIrondev
Licensed under the Inventarsystem EULA (Endbenutzer-Lizenzvertrag).
See Legal/LICENSE for the full license text.
Unauthorized commercial use, SaaS hosting, or removal of branding is prohibited.
For commercial licensing inquiries: https://github.com/AIIrondev
'''
"""
Funktion zum Protokollieren von Statusänderungen bei Ausleihungen
"""
import os
import datetime
from bson.objectid import ObjectId
def log_status_change(ausleihung_id, old_status, new_status, user=None):
"""
Protokolliert eine Statusänderung einer Ausleihung in einer Log-Datei.
Args:
ausleihung_id: Die ID der Ausleihung
old_status: Der alte Status
new_status: Der neue Status
user: Der Benutzer, der die Änderung vorgenommen hat (optional)
"""
try:
# Erstelle Log-Verzeichnis, falls es nicht existiert
log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'logs')
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# Log-Datei für Statusänderungen
log_file = os.path.join(log_dir, 'ausleihungen_status_changes.log')
# Protokolliere die Änderung
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
user_info = f" by {user}" if user else ""
with open(log_file, 'a', encoding='utf-8') as f:
f.write(f"{timestamp}: Ausleihung {ausleihung_id} - Status changed from '{old_status}' to '{new_status}'{user_info}\n")
return True
except Exception as e:
print(f"Fehler beim Protokollieren der Statusänderung: {e}")
return False
+4
View File
@@ -0,0 +1,4 @@
"""
"""
# Log initialization
@@ -32,8 +32,8 @@ from datetime import timezone
import os import os
import json import json
import shutil import shutil
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
# Add this helper function after imports # Add this helper function after imports
def ensure_timezone_aware(dt): def ensure_timezone_aware(dt):
@@ -103,7 +103,7 @@ def get_current_status(ausleihung, log_changes=False, user=None):
if log_changes and new_status != original_status and '_id' in ausleihung: if log_changes and new_status != original_status and '_id' in ausleihung:
try: try:
# Importieren Sie das Modul nur bei Bedarf, um zirkuläre Importe zu vermeiden # Importieren Sie das Modul nur bei Bedarf, um zirkuläre Importe zu vermeiden
import ausleihung_log import Web.modules.logs.ausleihung_log as ausleihung_log
ausleihung_log.log_status_change( ausleihung_log.log_status_change(
str(ausleihung['_id']), str(ausleihung['_id']),
original_status, original_status,
@@ -28,8 +28,8 @@ Collection Structure:
''' '''
from bson.objectid import ObjectId from bson.objectid import ObjectId
import datetime import datetime
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
LIBRARY_ITEM_TYPES = ('book', 'cd', 'dvd', 'media') LIBRARY_ITEM_TYPES = ('book', 'cd', 'dvd', 'media')
@@ -211,7 +211,7 @@ class _TenantAwareBool:
return f"_TenantAwareBool(module_name={self.module_name!r}, value={self.resolve()!r})" return f"_TenantAwareBool(module_name={self.module_name!r}, value={self.resolve()!r})"
from module_registry import registry as MODULES from Web.modules.module_registry import registry as MODULES
INVENTORY_MODULE_ENABLED = _TenantAwareBool('inventory', _get(_conf, ['modules', 'inventory', 'enabled'], DEFAULTS['modules']['inventory']['enabled'])) INVENTORY_MODULE_ENABLED = _TenantAwareBool('inventory', _get(_conf, ['modules', 'inventory', 'enabled'], DEFAULTS['modules']['inventory']['enabled']))
LIBRARY_MODULE_ENABLED = _TenantAwareBool('library', _get(_conf, ['modules', 'library', 'enabled'], DEFAULTS['modules']['library']['enabled'])) LIBRARY_MODULE_ENABLED = _TenantAwareBool('library', _get(_conf, ['modules', 'library', 'enabled'], DEFAULTS['modules']['library']['enabled']))
@@ -1,5 +1,5 @@
from pymongo import MongoClient from pymongo import MongoClient
import Web.settings as cfg import Web.modules.database.settings as cfg
def get_filter_names(): def get_filter_names():
client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT)
+2 -2
View File
@@ -17,8 +17,8 @@ import re
import secrets import secrets
import string import string
from bson.objectid import ObjectId from bson.objectid import ObjectId
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
logger = logging.getLogger('app') logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
+6
View File
@@ -0,0 +1,6 @@
"""
Inventar System Funktionen
"""
@@ -10,7 +10,7 @@ from datetime import datetime
from cryptography.fernet import Fernet, InvalidToken from cryptography.fernet import Fernet, InvalidToken
import settings as cfg import Web.modules.database.settings as cfg
_ENC_PREFIX = "enc::" _ENC_PREFIX = "enc::"
@@ -6,7 +6,7 @@
Unauthorized commercial use, SaaS hosting, or removal of branding is prohibited. Unauthorized commercial use, SaaS hosting, or removal of branding is prohibited.
For commercial licensing inquiries: https://github.com/AIIrondev For commercial licensing inquiries: https://github.com/AIIrondev
''' '''
import user import Web.modules.database.user as user
import sys import sys
import getpass import getpass
import re import re
@@ -18,7 +18,7 @@ from reportlab.pdfgen import canvas
from reportlab.lib.enums import TA_LEFT, TA_RIGHT, TA_CENTER, TA_JUSTIFY from reportlab.lib.enums import TA_LEFT, TA_RIGHT, TA_CENTER, TA_JUSTIFY
from reportlab.pdfbase import pdfmetrics from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont from reportlab.pdfbase.ttfonts import TTFont
import settings as cfg import Web.modules.database.settings as cfg
__version__ = cfg.APP_VERSION __version__ = cfg.APP_VERSION
+1
View File
@@ -0,0 +1 @@
print("hello")
+2 -2
View File
@@ -11,8 +11,8 @@ import requests
import hashlib import hashlib
import logging import logging
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
+2 -2
View File
@@ -12,8 +12,8 @@ from functools import wraps
import logging import logging
import os import os
import re import re
import settings as cfg import Web.modules.database.settings as cfg
from settings import MongoClient from Web.modules.database.settings import MongoClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
-1
View File
@@ -1 +0,0 @@
394722
-5
View File
@@ -1,5 +0,0 @@
#!/bin/bash
# Wrapper to run tenant management fully containerized via docker-compose
PROJECT_NAME=${COMPOSE_PROJECT_NAME:-$(basename "$PWD" | tr '[:upper:]' '[:lower:]')}
# Pass the absolute working directory to the container so docker compose can resolve paths correctly
docker compose -f docker-compose-multitenant.yml --profile tools run --rm -e COMPOSE_PROJECT_NAME="$PROJECT_NAME" -e HOST_WORKDIR="$PWD" tenant-manager "$@"
Binary file not shown.
Binary file not shown.
Binary file not shown.
-588
View File
@@ -1,588 +0,0 @@
"""
Test Suite for Ausleihung (Borrowing) System
Tests all core functionality of the borrowing/lending module
Run with: pytest test_ausleihung.py -v
"""
import pytest
import datetime
from bson.objectid import ObjectId
import sys
import os
# Add Web directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'Web'))
import ausleihung
import settings as cfg
from settings import MongoClient
@pytest.fixture(scope='session')
def db_client():
"""Create MongoDB connection for tests"""
client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT)
yield client
client.close()
@pytest.fixture(scope='session')
def test_db(db_client):
"""Get test database"""
return db_client[cfg.MONGODB_DB]
@pytest.fixture(autouse=True)
def cleanup_test_data(test_db):
"""Clean up test data before and after each test"""
yield
# Clean up after test
test_db['ausleihungen'].delete_many({})
@pytest.fixture
def sample_ausleihung_data():
"""Fixture with sample borrowing data"""
now = datetime.datetime.now()
return {
'item_id': str(ObjectId()),
'user': 'test_user',
'start_date': now,
'end_date': now + datetime.timedelta(days=1),
'notes': 'Test borrowing',
'period': None
}
# ============================================================================
# Status Determination Tests
# ============================================================================
class TestGetCurrentStatus:
"""Test status determination based on dates"""
def test_planned_status_future_date(self):
"""Test that future borrowing is marked as 'planned'"""
future_time = datetime.datetime.now() + datetime.timedelta(days=1)
ausleihung_doc = {
'Status': 'planned',
'Start': future_time,
'End': future_time + datetime.timedelta(hours=1)
}
status = ausleihung.get_current_status(ausleihung_doc)
assert status == 'planned'
def test_active_status_during_borrowing(self):
"""Test that current borrowing is marked as 'active'"""
now = datetime.datetime.now()
start = now - datetime.timedelta(hours=1)
end = now + datetime.timedelta(hours=1)
ausleihung_doc = {
'Status': 'active',
'Start': start,
'End': end
}
status = ausleihung.get_current_status(ausleihung_doc)
assert status == 'active'
def test_completed_status_after_end_time(self):
"""Test that past borrowing is marked as 'completed'"""
now = datetime.datetime.now()
start = now - datetime.timedelta(days=2)
end = now - datetime.timedelta(hours=1)
ausleihung_doc = {
'Status': 'active',
'Start': start,
'End': end
}
status = ausleihung.get_current_status(ausleihung_doc)
assert status == 'completed'
def test_cancelled_status_remains_cancelled(self):
"""Test that cancelled status is never changed"""
future_time = datetime.datetime.now() + datetime.timedelta(days=1)
ausleihung_doc = {
'Status': 'cancelled',
'Start': future_time,
'End': future_time + datetime.timedelta(hours=1)
}
status = ausleihung.get_current_status(ausleihung_doc)
assert status == 'cancelled'
def test_active_with_no_end_time(self):
"""Test that borrowing without end time stays active if started"""
now = datetime.datetime.now()
start = now - datetime.timedelta(hours=1)
ausleihung_doc = {
'Status': 'active',
'Start': start,
'End': None
}
status = ausleihung.get_current_status(ausleihung_doc)
assert status == 'active'
# ============================================================================
# Create and Update Tests
# ============================================================================
class TestCreateAusleihung:
"""Test creating new borrowings"""
def test_create_active_ausleihung(self, test_db):
"""Test creating an immediately active borrowing"""
item_id = str(ObjectId())
user = 'test_user'
start = datetime.datetime.now()
end = start + datetime.timedelta(hours=2)
result = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=start,
end_date=end,
notes='Test active',
status='active'
)
assert result is not None
# Verify in database
ausleihung_col = test_db['ausleihungen']
stored = ausleihung_col.find_one({'_id': result})
assert stored is not None
assert stored['Item'] == item_id # Correct field name
assert stored['User'] == user
assert stored['Status'] == 'active'
def test_create_planned_ausleihung(self, test_db):
"""Test creating a future/planned borrowing"""
item_id = str(ObjectId())
user = 'test_user'
future_start = datetime.datetime.now() + datetime.timedelta(days=1)
future_end = future_start + datetime.timedelta(hours=2)
result = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=future_start,
end_date=future_end,
notes='Test planned',
status='planned'
)
assert result is not None
ausleihung_col = test_db['ausleihungen']
stored = ausleihung_col.find_one({'_id': result})
assert stored['Status'] == 'planned'
# Check approximately equal (within 1 second for datetime precision)
assert abs((stored['Start'] - future_start).total_seconds()) < 1
class TestUpdateAusleihung:
"""Test updating existing borrowings"""
def test_update_ausleihung_dates(self, test_db, sample_ausleihung_data):
"""Test updating borrowing dates"""
# Create initial borrowing
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
assert ausleihung_id is not None
# Update dates
new_start = datetime.datetime.now() + datetime.timedelta(days=2)
new_end = new_start + datetime.timedelta(hours=1)
ausleihung.update_ausleihung(
id=ausleihung_id,
start=new_start,
end=new_end
)
# Verify update (within 1 second tolerance for datetime precision)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert abs((stored['Start'] - new_start).total_seconds()) < 1
assert abs((stored['End'] - new_end).total_seconds()) < 1
def test_update_ausleihung_status(self, test_db, sample_ausleihung_data):
"""Test updating borrowing status"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
ausleihung.update_ausleihung(id=ausleihung_id, status='completed')
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Status'] == 'completed'
def test_update_ausleihung_notes(self, test_db, sample_ausleihung_data):
"""Test updating borrowing notes"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
new_notes = 'Updated notes'
ausleihung.update_ausleihung(id=ausleihung_id, notes=new_notes)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Notes'] == new_notes
# ============================================================================
# Complete and Cancel Tests
# ============================================================================
class TestCompleteAusleihung:
"""Test completing borrowings"""
def test_complete_ausleihung(self, test_db, sample_ausleihung_data):
"""Test marking a borrowing as completed"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
end_time = datetime.datetime.now()
ausleihung.complete_ausleihung(ausleihung_id, end_time=end_time)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Status'] == 'completed'
assert stored['End'] == end_time or stored['End'] is not None
class TestCancelAusleihung:
"""Test canceling borrowings"""
def test_cancel_ausleihung(self, test_db, sample_ausleihung_data):
"""Test canceling a borrowing"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
ausleihung.cancel_ausleihung(ausleihung_id)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Status'] == 'cancelled'
# ============================================================================
# Query Tests
# ============================================================================
class TestGetAusleihung:
"""Test retrieving borrowings"""
def test_get_ausleihung_by_id(self, test_db, sample_ausleihung_data):
"""Test fetching a borrowing by ID"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
retrieved = ausleihung.get_ausleihung(ausleihung_id)
assert retrieved is not None
assert retrieved['_id'] == ausleihung_id
assert retrieved['User'] == sample_ausleihung_data['user']
def test_get_ausleihung_by_user(self, test_db):
"""Test retrieving all borrowings for a user"""
user = 'test_user_xyz'
item1 = str(ObjectId())
item2 = str(ObjectId())
now = datetime.datetime.now()
# Create multiple borrowings for same user
ausleihung.add_ausleihung(
item_id=item1,
user=user,
start_date=now,
end_date=now + datetime.timedelta(hours=1),
status='active'
)
ausleihung.add_ausleihung(
item_id=item2,
user=user,
start_date=now + datetime.timedelta(days=1),
end_date=now + datetime.timedelta(days=1, hours=1),
status='planned'
)
# Retrieve all for user
borrowings = ausleihung.get_ausleihung_by_user(user)
assert len(borrowings) >= 2
assert all(b['User'] == user for b in borrowings)
def test_get_ausleihungen_by_status(self, test_db):
"""Test retrieving borrowings by status"""
item_id = str(ObjectId())
user = 'test_user'
now = datetime.datetime.now()
# Create active
active_id = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=now - datetime.timedelta(hours=1),
end_date=now + datetime.timedelta(hours=1),
status='active'
)
# Get active borrowings
active_borrowings = ausleihung.get_active_ausleihungen()
assert any(b['_id'] == active_id for b in active_borrowings)
# ============================================================================
# Conflict Detection Tests
# ============================================================================
class TestConflictDetection:
"""Test detecting overlapping/conflicting borrowings"""
def test_no_conflict_different_items(self, test_db):
"""Test that different items don't conflict"""
item1 = str(ObjectId())
item2 = str(ObjectId())
now = datetime.datetime.now()
start = now
end = now + datetime.timedelta(hours=1)
# Create first borrowing
ausleihung.add_ausleihung(
item_id=item1,
user='user1',
start_date=start,
end_date=end,
status='active'
)
# Check conflict on different item (should be no conflict)
conflict = ausleihung.check_ausleihung_conflict(
item_id=item2,
start_date=start,
end_date=end
)
assert conflict is False
def test_conflict_same_item_overlapping(self, test_db):
"""Test that overlapping borrowings on same item are detected"""
item_id = str(ObjectId())
now = datetime.datetime.now()
# Create first borrowing
ausleihung.add_ausleihung(
item_id=item_id,
user='user1',
start_date=now,
end_date=now + datetime.timedelta(hours=2),
status='active'
)
# Try to create overlapping borrowing
conflict = ausleihung.check_ausleihung_conflict(
item_id=item_id,
start_date=now + datetime.timedelta(minutes=30),
end_date=now + datetime.timedelta(hours=3)
)
assert conflict is True or conflict == item_id # Depending on implementation
def test_no_conflict_different_times(self, test_db):
"""Test that non-overlapping borrowings don't conflict"""
item_id = str(ObjectId())
now = datetime.datetime.now()
# Create first borrowing
ausleihung.add_ausleihung(
item_id=item_id,
user='user1',
start_date=now,
end_date=now + datetime.timedelta(hours=1),
status='active'
)
# Check borrowing after first ends (should be no conflict)
conflict = ausleihung.check_ausleihung_conflict(
item_id=item_id,
start_date=now + datetime.timedelta(hours=2),
end_date=now + datetime.timedelta(hours=3)
)
assert conflict is False or conflict is None
# ============================================================================
# Period-based Borrowing Tests
# ============================================================================
class TestPeriodBookings:
"""Test period-based borrowings (school periods)"""
def test_create_period_booking(self, test_db):
"""Test creating a borrowing for a specific school period"""
item_id = str(ObjectId())
user = 'test_user'
today = datetime.datetime.now().date()
result = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=datetime.datetime.combine(today, datetime.time(8, 0)),
end_date=datetime.datetime.combine(today, datetime.time(9, 0)),
period=1, # Assuming period 1 is first period
status='active'
)
assert result is not None
stored = test_db['ausleihungen'].find_one({'_id': result})
assert stored.get('Period') == 1
# ============================================================================
# Remove Tests
# ============================================================================
class TestRemoveAusleihung:
"""Test removing/deleting borrowings"""
def test_remove_ausleihung(self, test_db, sample_ausleihung_data):
"""Test deleting a borrowing record (soft delete)"""
ausleihung_id = ausleihung.add_ausleihung(**sample_ausleihung_data)
stored_before = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored_before is not None
# Remove (soft delete - adds DeletedAt timestamp)
ausleihung.remove_ausleihung(ausleihung_id)
# Verify it's marked as deleted (soft delete)
stored_after = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored_after is not None # Still exists
assert 'DeletedAt' in stored_after or stored_after.get('Status') == 'deleted'
# ============================================================================
# Integration Tests
# ============================================================================
class TestAusleihungLifecycle:
"""Test complete borrowing lifecycle"""
def test_full_lifecycle_active_to_complete(self, test_db):
"""Test a complete borrowing lifecycle: active → complete"""
item_id = str(ObjectId())
user = 'test_user'
now = datetime.datetime.now()
# 1. Create active borrowing
ausleihung_id = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=now - datetime.timedelta(hours=1),
end_date=now + datetime.timedelta(hours=1),
status='active'
)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
status = ausleihung.get_current_status(stored)
assert status == 'active'
# 2. Complete the borrowing
ausleihung.complete_ausleihung(ausleihung_id)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
status = ausleihung.get_current_status(stored)
assert status == 'completed'
def test_full_lifecycle_planned_to_active_to_complete(self, test_db):
"""Test complete lifecycle: planned → active → complete"""
item_id = str(ObjectId())
user = 'test_user'
now = datetime.datetime.now()
future = now + datetime.timedelta(hours=1)
# 1. Create planned borrowing
ausleihung_id = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=future,
end_date=future + datetime.timedelta(hours=1),
status='planned'
)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert ausleihung.get_current_status(stored) == 'planned'
# 2. Update to active (simulate time passing or manual activation)
ausleihung.update_ausleihung(id=ausleihung_id, status='active')
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Status'] == 'active'
# 3. Complete
ausleihung.complete_ausleihung(ausleihung_id)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert ausleihung.get_current_status(stored) == 'completed'
def test_cancel_planned_borrowing(self, test_db):
"""Test canceling a planned borrowing"""
item_id = str(ObjectId())
user = 'test_user'
future = datetime.datetime.now() + datetime.timedelta(days=1)
# Create planned
ausleihung_id = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=future,
end_date=future + datetime.timedelta(hours=1),
status='planned'
)
# Cancel
ausleihung.cancel_ausleihung(ausleihung_id)
stored = test_db['ausleihungen'].find_one({'_id': ausleihung_id})
assert stored['Status'] == 'cancelled'
# ============================================================================
# Edge Cases
# ============================================================================
class TestEdgeCases:
"""Test edge cases and boundary conditions"""
def test_borrowing_with_same_start_and_end(self, test_db):
"""Test borrowing where start equals end"""
item_id = str(ObjectId())
user = 'test_user'
now = datetime.datetime.now()
result = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=now,
end_date=now, # Same time
status='active'
)
assert result is not None
def test_borrowing_without_end_date(self, test_db):
"""Test creating borrowing without end date"""
item_id = str(ObjectId())
user = 'test_user'
now = datetime.datetime.now()
result = ausleihung.add_ausleihung(
item_id=item_id,
user=user,
start_date=now,
end_date=None,
status='active'
)
assert result is not None
stored = test_db['ausleihungen'].find_one({'_id': result})
# End field should not exist or be None if not provided
assert 'End' not in stored or stored.get('End') is None
def test_get_nonexistent_borrowing(self, test_db):
"""Test retrieving a nonexistent borrowing"""
fake_id = ObjectId()
result = ausleihung.get_ausleihung(fake_id)
assert result is None or result == {} or result == []
# ============================================================================
# Run Tests
# ============================================================================
if __name__ == '__main__':
pytest.main([__file__, '-v', '--tb=short'])
-5
View File
@@ -1,5 +0,0 @@
from Web.app import app
with app.test_client() as client:
resp = client.get('/terminplan')
print(resp.status_code)
# print(resp.data.decode('utf-8')[:200])
+55
View File
@@ -18,6 +18,7 @@ DIST_DIR="$PROJECT_DIR/dist"
COMPOSE_FILE="docker-compose-multitenant.yml" COMPOSE_FILE="docker-compose-multitenant.yml"
MIN_ROOT_FREE_MB="${INVENTAR_MIN_ROOT_FREE_MB:-2048}" MIN_ROOT_FREE_MB="${INVENTAR_MIN_ROOT_FREE_MB:-2048}"
DIST_KEEP_COUNT="${INVENTAR_DIST_KEEP_COUNT:-2}" DIST_KEEP_COUNT="${INVENTAR_DIST_KEEP_COUNT:-2}"
MODE="release"
mkdir -p "$LOG_DIR" mkdir -p "$LOG_DIR"
chmod 777 "$LOG_DIR" 2>/dev/null || true chmod 777 "$LOG_DIR" 2>/dev/null || true
@@ -152,6 +153,7 @@ Usage: $0 [options]
Options: Options:
--multitenant Use docker-compose-multitenant.yml (default) --multitenant Use docker-compose-multitenant.yml (default)
development Install development build from GHCR or local dist
-h, --help Show this help message -h, --help Show this help message
EOF EOF
} }
@@ -163,6 +165,10 @@ parse_args() {
COMPOSE_FILE="docker-compose-multitenant.yml" COMPOSE_FILE="docker-compose-multitenant.yml"
shift shift
;; ;;
development|dev)
MODE="development"
shift
;;
-h|--help) -h|--help)
usage usage
exit 0 exit 0
@@ -480,6 +486,55 @@ main() {
archive_logs archive_logs
create_backup create_backup
# If user requested a development install, perform a simple dev deploy flow
if [ "$MODE" = "development" ]; then
log_message "Requested development install"
local tag="development"
local app_image="$APP_IMAGE_REPO:$tag"
local compose_path="$PROJECT_DIR/$COMPOSE_FILE"
if [ ! -f "$compose_path" ]; then
log_message "ERROR: compose file not found: $compose_path"
exit 1
fi
# Ensure ENV_FILE contains the development image
if [ ! -f "$ENV_FILE" ]; then
cat > "$ENV_FILE" <<EOF
NUITKA_BUILD=0
INVENTAR_HTTP_PORT=10000
INVENTAR_APP_IMAGE=$app_image
EOF
elif grep -q '^INVENTAR_APP_IMAGE=' "$ENV_FILE"; then
sed -i "s|^INVENTAR_APP_IMAGE=.*|INVENTAR_APP_IMAGE=$app_image|" "$ENV_FILE"
else
printf '\nINVENTAR_APP_IMAGE=%s\n' "$app_image" >> "$ENV_FILE"
fi
# Try local dist first, then pull from GHCR
if ! load_local_dist_image "$tag"; then
log_message "Attempting to pull development image $app_image"
if ! docker pull "$app_image" >> "$LOG_FILE" 2>&1; then
log_message "ERROR: Could not obtain development image $app_image"
exit 1
fi
fi
# Bring up stack
docker compose -f "$compose_path" --env-file "$ENV_FILE" pull app mongodb >> "$LOG_FILE" 2>&1 || true
docker compose -f "$compose_path" --env-file "$ENV_FILE" up -d --remove-orphans >> "$LOG_FILE" 2>&1
docker tag "$app_image" "$APP_IMAGE_REPO:latest" >> "$LOG_FILE" 2>&1 || true
if ! verify_stack_health; then
log_message "ERROR: Development deployment failed health check"
exit 1
fi
echo "$tag" > "$STATE_FILE"
log_message "Development update completed successfully"
exit 0
fi
local tmp_dir meta_file latest_tag current_tag bundle_url local tmp_dir meta_file latest_tag current_tag bundle_url
tmp_dir="$(mktemp -d)" tmp_dir="$(mktemp -d)"
meta_file="$tmp_dir/release.json" meta_file="$tmp_dir/release.json"
-235
View File
@@ -1,235 +0,0 @@
#!/usr/bin/env python3
"""
Invario PDF Audit Export - Implementation Verification Script
This script verifies that all components of the PDF audit export system
are correctly installed and configured.
"""
import sys
import os
def verify_files():
"""Verify all required files exist."""
print("=" * 60)
print("INVARIO PDF AUDIT EXPORT - INSTALLATION VERIFICATION")
print("=" * 60)
print()
files_to_check = {
"Core Module": [
"Web/pdf_audit_export.py",
],
"Flask Integration": [
"Web/app.py", # Should contain new routes
"Web/templates/admin_audit.html", # Should contain new buttons
],
"Documentation": [
"PDF_AUDIT_EXPORT_DOCUMENTATION.md",
"PDF_IMPLEMENTATION_GUIDE.md",
"QUICK_START_PDF_EXPORT.md",
]
}
base_path = os.path.dirname(os.path.abspath(__file__))
all_ok = True
for category, files in files_to_check.items():
print(f"\n[{category}]")
for filename in files:
filepath = os.path.join(base_path, filename)
exists = os.path.exists(filepath)
status = "" if exists else ""
print(f" {status} {filename}")
if not exists:
all_ok = False
return all_ok
def verify_dependencies():
"""Verify Python dependencies are installed."""
print("\n[Python Dependencies]")
dependencies = [
("reportlab", "PDF generation"),
("qrcode", "QR code generation"),
("pillow", "Image processing"),
("flask", "Web framework"),
("pymongo", "MongoDB driver"),
]
all_ok = True
for package, description in dependencies:
try:
__import__(package)
print(f"{package:15} - {description}")
except ImportError:
print(f"{package:15} - {description}")
all_ok = False
return all_ok
def verify_routes():
"""Verify new routes are defined in app.py."""
print("\n[Flask Routes]")
routes_to_check = [
"/admin/audit/export/pdf/quick",
"/admin/audit/export/pdf/official",
]
app_py_path = "Web/app.py"
if not os.path.exists(app_py_path):
print(" ✗ app.py not found")
return False
with open(app_py_path, 'r') as f:
app_content = f.read()
all_ok = True
for route in routes_to_check:
if route in app_content:
print(f"{route}")
else:
print(f"{route}")
all_ok = False
return all_ok
def verify_template():
"""Verify template updates are in place."""
print("\n[HTML Template Updates]")
template_path = "Web/templates/admin_audit.html"
if not os.path.exists(template_path):
print(" ✗ admin_audit.html not found")
return False
with open(template_path, 'r') as f:
template_content = f.read()
checks = {
"DIN 5008 Info Box": "DIN 5008",
"PDF Quick-Check Button": "admin_audit_export_pdf_quick",
"PDF Official Button": "admin_audit_export_pdf_official",
}
all_ok = True
for check_name, check_string in checks.items():
if check_string in template_content:
print(f"{check_name}")
else:
print(f"{check_name}")
all_ok = False
return all_ok
def get_statistics():
"""Get implementation statistics."""
print("\n[Implementation Statistics]")
stats = {
"Web/pdf_audit_export.py": "PDF Export Module",
"PDF_AUDIT_EXPORT_DOCUMENTATION.md": "Technical Documentation",
"PDF_IMPLEMENTATION_GUIDE.md": "Implementation Guide",
"QUICK_START_PDF_EXPORT.md": "Quick Start Guide",
}
total_lines = 0
for filename, description in stats.items():
if os.path.exists(filename):
with open(filename, 'r') as f:
lines = len(f.readlines())
print(f" {filename:45} {lines:5} lines - {description}")
total_lines += lines
print(f"\n Total documentation: {total_lines} lines")
def print_next_steps():
"""Print next steps for the user."""
print("\n" + "=" * 60)
print("NEXT STEPS")
print("=" * 60)
print("""
1. CONFIGURE SCHOOL INFO (Optional)
Edit config.json and add:
{
"school": {
"name": "Your School Name",
"address": "School Address",
"postal_code": "12345",
"city": "City Name",
"school_number": "123456",
"it_admin": "Admin Name"
}
}
2. RESTART THE SYSTEM
./start.sh
or: python Web/app.py
3. TEST PDF EXPORTS
- Login to http://localhost:8000/admin/audit
- Click "🚀 Schnell-Check" for compact PDF
- Click "📋 Amtlicher Bericht" for full DIN 5008 report
4. VERIFY COMPLIANCE
- Check PDF opens in your PDF reader
- Verify school info is correct
- Test signature fields
- Verify barrierefreiheit (accessibility)
5. DEPLOY TO PRODUCTION
- Update your deployment scripts
- Document new export procedures
- Train staff on Quick-Check vs Official Report
""")
def main():
"""Run all verifications."""
print()
checks = [
("File Structure", verify_files),
("Dependencies", verify_dependencies),
("Flask Routes", verify_routes),
("HTML Template", verify_template),
]
all_passed = True
for name, check_func in checks:
try:
passed = check_func()
if not passed:
all_passed = False
except Exception as e:
print(f"\n✗ Error during {name} check: {e}")
all_passed = False
# Get statistics
get_statistics()
# Print results
print("\n" + "=" * 60)
if all_passed:
print("✓ ALL VERIFICATION CHECKS PASSED!")
print("=" * 60)
print_next_steps()
else:
print("✗ SOME VERIFICATION CHECKS FAILED")
print("=" * 60)
print("\nPlease check the errors above and verify:")
print("1. All files are in the correct locations")
print("2. All dependencies are installed")
print("3. app.py and admin_audit.html have been updated")
sys.exit(1)
print("\n" + "=" * 60)
print("For detailed documentation, see:")
print(" - QUICK_START_PDF_EXPORT.md (Start here!)")
print(" - PDF_IMPLEMENTATION_GUIDE.md (Technical details)")
print(" - PDF_AUDIT_EXPORT_DOCUMENTATION.md (Requirements)")
print("=" * 60 + "\n")
if __name__ == "__main__":
main()