aufräumen
This commit is contained in:
8
backend/development/crontab-example
Normal file
8
backend/development/crontab-example
Normal file
@ -0,0 +1,8 @@
|
||||
# MYP Backend Cron-Jobs
|
||||
# Installiere mit: crontab crontab-example
|
||||
|
||||
# Prüfe alle 5 Minuten auf abgelaufene Reservierungen und schalte Steckdosen aus
|
||||
*/5 * * * * cd /pfad/zum/projektarbeit-myp/backend && /pfad/zur/venv/bin/flask check-jobs >> /pfad/zum/projektarbeit-myp/backend/logs/cron.log 2>&1
|
||||
|
||||
# Tägliche Sicherung der Datenbank um 3:00 Uhr
|
||||
0 3 * * * cd /pfad/zum/projektarbeit-myp/backend && cp instance/myp.db instance/backups/myp-$(date +\%Y\%m\%d).db
|
||||
84
backend/development/initialize_myp_database.sh
Normal file
84
backend/development/initialize_myp_database.sh
Normal file
@ -0,0 +1,84 @@
|
||||
#!/bin/bash
|
||||
|
||||
# MYP Datenbank Initialisierungs-Skript
|
||||
# Dieses Skript erstellt die erforderlichen Datenbanktabellen für das MYP Backend
|
||||
|
||||
echo "=== MYP Datenbank Initialisierung ==="
|
||||
echo ""
|
||||
|
||||
# Prüfe, ob sqlite3 installiert ist
|
||||
if ! command -v sqlite3 &> /dev/null; then
|
||||
echo "FEHLER: sqlite3 ist nicht installiert."
|
||||
echo "Bitte installiere sqlite3 mit deinem Paketmanager, z.B. 'apt install sqlite3'"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Erstelle Instance-Ordner, falls nicht vorhanden
|
||||
echo "Erstelle instance-Ordner, falls nicht vorhanden..."
|
||||
mkdir -p instance/backups
|
||||
|
||||
# Prüfen, ob die Datenbank bereits existiert
|
||||
if [ -f "instance/myp.db" ]; then
|
||||
echo "Datenbank existiert bereits."
|
||||
echo "Erstelle Backup in instance/backups..."
|
||||
cp instance/myp.db "instance/backups/myp_$(date '+%Y%m%d_%H%M%S').db"
|
||||
fi
|
||||
|
||||
# Erstelle die Datenbank und ihre Tabellen
|
||||
echo "Erstelle neue Datenbank..."
|
||||
sqlite3 instance/myp.db <<EOF
|
||||
PRAGMA foreign_keys = ON;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS user (
|
||||
id TEXT PRIMARY KEY,
|
||||
username TEXT UNIQUE NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
display_name TEXT,
|
||||
email TEXT UNIQUE,
|
||||
role TEXT DEFAULT 'user'
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS session (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
expires_at TIMESTAMP NOT NULL,
|
||||
FOREIGN KEY (user_id) REFERENCES user (id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS printer (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL,
|
||||
status INTEGER DEFAULT 0,
|
||||
ip_address TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS print_job (
|
||||
id TEXT PRIMARY KEY,
|
||||
printer_id TEXT NOT NULL,
|
||||
user_id TEXT NOT NULL,
|
||||
start_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
duration_in_minutes INTEGER NOT NULL,
|
||||
comments TEXT,
|
||||
aborted INTEGER DEFAULT 0,
|
||||
abort_reason TEXT,
|
||||
FOREIGN KEY (printer_id) REFERENCES printer (id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (user_id) REFERENCES user (id) ON DELETE CASCADE
|
||||
);
|
||||
EOF
|
||||
|
||||
# Setze Berechtigungen für die Datenbankdatei
|
||||
chmod 644 instance/myp.db
|
||||
|
||||
echo ""
|
||||
echo "=== Datenbank-Initialisierung abgeschlossen ==="
|
||||
echo ""
|
||||
echo "Du kannst jetzt einen Admin-Benutzer über die Web-Oberfläche registrieren."
|
||||
echo "Der erste registrierte Benutzer wird automatisch zum Admin."
|
||||
echo ""
|
||||
echo "Starte den Server mit:"
|
||||
echo "python app.py"
|
||||
echo ""
|
||||
echo "Alternativ kannst du einen Admin-Benutzer über die API erstellen mit:"
|
||||
echo "curl -X POST http://localhost:5000/api/create-initial-admin -H \"Content-Type: application/json\" -d '{\"username\":\"admin\",\"password\":\"password\",\"displayName\":\"Administrator\"}'"
|
||||
echo ""
|
||||
73
backend/development/install.sh
Normal file
73
backend/development/install.sh
Normal file
@ -0,0 +1,73 @@
|
||||
#!/bin/bash
|
||||
|
||||
# Installation Script für MYP Backend
|
||||
|
||||
echo "=== MYP Backend Installation ==="
|
||||
echo ""
|
||||
|
||||
# Prüfe Python-Version
|
||||
python_version=$(python3 --version 2>&1 | awk '{print $2}')
|
||||
echo "Python-Version: $python_version"
|
||||
|
||||
# Prüfe, ob die Python-Version mindestens 3.8 ist
|
||||
required_version="3.8.0"
|
||||
if [[ "$(printf '%s\n' "$required_version" "$python_version" | sort -V | head -n1)" != "$required_version" ]]; then
|
||||
echo "FEHLER: Python $required_version oder höher wird benötigt"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Prüfe, ob sqlite3 installiert ist
|
||||
if ! command -v sqlite3 &> /dev/null; then
|
||||
echo "FEHLER: sqlite3 ist nicht installiert."
|
||||
echo "Bitte installiere sqlite3 mit deinem Paketmanager, z.B. 'apt install sqlite3'"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Erstelle virtuelle Umgebung
|
||||
echo ""
|
||||
echo "Erstelle virtuelle Python-Umgebung..."
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate
|
||||
|
||||
# Installiere Abhängigkeiten
|
||||
echo ""
|
||||
echo "Installiere Abhängigkeiten..."
|
||||
pip install --upgrade pip
|
||||
pip install -r requirements.txt
|
||||
|
||||
# Erstelle .env-Datei
|
||||
echo ""
|
||||
echo "Erstelle .env-Datei..."
|
||||
if [ ! -f .env ]; then
|
||||
cp .env.example .env
|
||||
echo "Die .env-Datei wurde aus der Beispieldatei erstellt."
|
||||
echo "Bitte passe die Konfiguration an, falls nötig."
|
||||
else
|
||||
echo ".env-Datei existiert bereits."
|
||||
fi
|
||||
|
||||
# Erstelle Logs-Ordner
|
||||
echo ""
|
||||
echo "Erstelle logs-Ordner..."
|
||||
mkdir -p logs
|
||||
|
||||
# Initialisiere die Datenbank
|
||||
echo ""
|
||||
echo "Initialisiere die Datenbank..."
|
||||
bash initialize_myp_database.sh
|
||||
|
||||
echo ""
|
||||
echo "=== Installation abgeschlossen ==="
|
||||
echo ""
|
||||
echo "Wichtige Schritte vor dem Start:"
|
||||
echo "1. Passe die Konfigurationen in der .env-Datei an"
|
||||
echo "2. Konfiguriere die Tapo-Steckdosen-Zugangsdaten in der .env-Datei (optional)"
|
||||
echo "3. Passe die crontab-example an und installiere den Cron-Job (optional)"
|
||||
echo ""
|
||||
echo "Starte den Server mit:"
|
||||
echo "source venv/bin/activate"
|
||||
echo "python app.py"
|
||||
echo ""
|
||||
echo "Oder mit Gunicorn für Produktion:"
|
||||
echo "gunicorn --bind 0.0.0.0:5000 app:app"
|
||||
echo ""
|
||||
95
backend/development/tests/api-test.drucker.py
Normal file
95
backend/development/tests/api-test.drucker.py
Normal file
@ -0,0 +1,95 @@
|
||||
import requests
|
||||
import json
|
||||
|
||||
# Basis-URL inkl. Token
|
||||
url = "http://192.168.0.102:80/app?token=9DFAC92C53CEC92E67A9CB2E00B3CB2F"
|
||||
|
||||
# HTTP-Header wie in der Originalanfrage
|
||||
headers = {
|
||||
"Referer": "http://192.168.0.102:80",
|
||||
"Accept": "application/json",
|
||||
"requestByApp": "true",
|
||||
"Content-Type": "application/json; charset=UTF-8",
|
||||
"Host": "192.168.0.102",
|
||||
"Connection": "Keep-Alive",
|
||||
"Accept-Encoding": "gzip",
|
||||
"User-Agent": "okhttp/3.14.9"
|
||||
}
|
||||
|
||||
# Liste der Payloads (als Python-Dictionaries)
|
||||
payloads = [
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"ZC4CHp6bbfBO1rtmuH6I+TStBIiFRfQpayYPwet5NBmL35dib5xXHeEeLM7c0OSQSyxO6fnbXrC1\n"
|
||||
"gXdfowwwq4Fum9ispgt8yT7cgbDcqnoVrhxEtHIDfuwLh8YAGmDSfTMo/JlsGspWPYMKd1EWXtb5\n"
|
||||
"gP9FA9LHnV2kxKsNSPQ=\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"k111EbfCcfVzAouNbu1vyos9Ltsg+a97n4xUUQMviQVJfhqxvKOhv1SrvEk2LvpD0LwNVUNPZdwU\n"
|
||||
"6pH5E/NOwdc1WzTPeqHiY760GpUuqn0tToHEHEyO2HaSKdrAYnw2gN410bvHb0pM3gYWS43eOA==\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"7/uYVDwyNfFhg9y7rHyp+4AGKBYQPyaBN6cFMl9j4ER/JpJTcGBdaUteSmx8P8Fkz+b2kkNLjYa2\n"
|
||||
"wQr2gA3m6vEq9jpnAF2V3fv9c4Yg9gja9MlTIZqM6EdMi7YbfbhLme34Bh8kMcohDR3u1F4DwFDz\n"
|
||||
"hNZPckf/CegbY9KGFeGwT4rWyX3BTk9+FE7ldtJn\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"EjWZb+YYS9tihgLdX4x+Wwx7q+e5X/ZHicr4jOnYmpFToDANzpm5ZpzD49BITcTCdQMOHlJBis85\n"
|
||||
"9GX6Hv8j66OITyH0XmfG9dQo2tgIykyagCZIofr/BpAWYX4aRaOkU4z14mVa2XpDtHJQjc+pXYkh\n"
|
||||
"JuWvLE+h01U5RoyPtvE=\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"OwyTsm5HdB/ReJMhVRrkjnV0NLTanw6iXOxVrDDexT456edWuwKiBOsZUyBHmUyJKgiPQzOXqyWWi220bX8IjLX4q8YNgPwRlj+7nRbfzpC/I57wBZBTWIt626pSdIH0vpiuPq84KMfPD5BB2p78/LjsqlzyeLGYzkSsGRBMT8TnLMDFzZE864nfDUZ9muH2kk8NRMN9l6xoCXBJqGA9q8XxIWRTpsl0kTx52kUszY69hYlfFSrrCDIls1ykul14/T1NtOVF8KOgiwaSGOZf7L4QlbhYvRj9kkVVkrxhlwt8jtMqfJKEqq+CIPh3Mp4440WYMLRo6VNIEJ3pWjplkJmc+htnYC4FwVgT7mHZ8eeGGKBvsJz+78gTaHnGBnwZ26I8UdFparyp6QXpOhK9zFmGVh0yapiTHo6jOOI+4Q3Ru+aPnidX/ZASPmR7CZO70CUpvv9zIKJnrAaoTMmH7A6+kcmCRLgLFaTaM+4DFmiz6JGP+4W7MmVPJxxvn0IFlo1P/xwNDuL3T6GLUIEVNk89JG5roBm7AdchUZJO38dGZ0eFiiTK/NhKPvjj+fk9A4FGh7EDshXZhL2u50cdLcdUtcP/CAMDjgWlMm4Kk3vxMQO+UGE+jsB7NkaulmTW1jcl+PSnAE5P71oqVVQ0ng==\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"7/uYVDwyNfFhg9y7rHyp+4AGKBYQPyaBN6cFMl9j4ER/JpJTcGBdaUteSmx8P8FkURmv/LWV1FpO\n"
|
||||
"M3RWvsiC5UAsei2G+vwTVuQpOPjKKAx+qwftr9Qs2mSkPNjNLpWHK68EZkIw+h04TQkt0Q99Dirg\n"
|
||||
"0BcrPgHTVKjiK8mdZ6w6gcld/h/FOKYMqJrP0Z+2\n"
|
||||
)
|
||||
}
|
||||
},
|
||||
{
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": (
|
||||
"ZE/+XlUmTA9D3DFfp4x3xhS3vdsQ+60tz4TOodtZDby/4DPoqk9EBvJZ1JtUCr5c0AHuv/sfwcvN\n"
|
||||
"Vx1zJP9RkltrAKVTWoaESAeewLozpXt/x0s/jkYC1rh7eTrxm+nYTZ5LJgNtcQq8yJxhEPez1w==\n"
|
||||
)
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
# Sende die Payloads sequenziell per POST-Anfrage
|
||||
for idx, payload in enumerate(payloads, start=1):
|
||||
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
||||
print(f"Anfrage {idx}:")
|
||||
print("Status Code:", response.status_code)
|
||||
print("Response Text:", response.text)
|
||||
print("-" * 60)
|
||||
BIN
backend/development/tests/capture.pcap
Normal file
BIN
backend/development/tests/capture.pcap
Normal file
Binary file not shown.
128
backend/development/tests/handshake.py
Normal file
128
backend/development/tests/handshake.py
Normal file
@ -0,0 +1,128 @@
|
||||
import requests
|
||||
import json
|
||||
|
||||
# Constants from the Wireshark capture
|
||||
PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
|
||||
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCMl89OZsjqE8yZ9TQhUb9h539WTX3U8Y5YCNdp
|
||||
OhuXvLFYcAT5mvC074VFROmD0xhvw5hrwESOisqpPPU9r78JpLuYUKd+/aidvykqBT8OW5rDLb6d
|
||||
O9FO6Gc+bV8L8ttHVlBFoX69EqiRhcreGPG6FQz4JqGJF4T1nFi0EvALXwIDAQAB
|
||||
-----END PUBLIC KEY-----"""
|
||||
|
||||
# Vorbereitete verschlüsselte Befehle (aus Wireshark extrahiert)
|
||||
COMMAND_ON = """ps0Puxc37EK4PhfcevceL3lyyDrjwLT1+443DDXNbcNRsltlgCQ6+oXgsrE2Pl5OhV73ZI/oM5Nj
|
||||
37cWEaHpXPiHdr1W0cD3aJ5qJ55TfTRkHP9xcMNQJHCn6aWPEHpR7xvvXW9WbJWfShnE2Xdvmw==
|
||||
"""
|
||||
|
||||
COMMAND_OFF = """FlO5i3DRcrUmu2ZwIIv8b68EisGu8VCuqfGOydaR+xCA0n3f2W/EcqVj8MurRBFXYTrZ/uwa1W26
|
||||
ftCfvhdXNebBRwHr9Rj3id4bVfltJ8eT5/R3xY8kputklW2mrw9UfdISzAJqOPp9KZcU4K9p8g==
|
||||
"""
|
||||
|
||||
class TapoP115Controller:
|
||||
def __init__(self, device_ip):
|
||||
self.device_ip = device_ip
|
||||
self.session_id = None
|
||||
self.token = None
|
||||
|
||||
def perform_handshake(self):
|
||||
"""Führt den ersten Handshake durch und speichert die Session-ID"""
|
||||
handshake_data = {
|
||||
"method": "handshake",
|
||||
"params": {
|
||||
"key": PUBLIC_KEY
|
||||
},
|
||||
"requestTimeMils": 0
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Referer": f"http://{self.device_ip}:80",
|
||||
"Accept": "application/json",
|
||||
"requestByApp": "true",
|
||||
"Content-Type": "application/json; charset=UTF-8"
|
||||
}
|
||||
|
||||
response = requests.post(
|
||||
f"http://{self.device_ip}/app",
|
||||
json=handshake_data,
|
||||
headers=headers
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
if data["error_code"] == 0:
|
||||
# Session-ID aus dem Cookie extrahieren
|
||||
self.session_id = response.cookies.get("TP_SESSIONID")
|
||||
print(f"Handshake erfolgreich, Session-ID: {self.session_id}")
|
||||
|
||||
# In einem echten Szenario würden wir hier den verschlüsselten Schlüssel entschlüsseln
|
||||
# Da wir keinen privaten Schlüssel haben, speichern wir nur die Antwort
|
||||
encrypted_key = data["result"]["key"]
|
||||
print(f"Verschlüsselter Schlüssel: {encrypted_key}")
|
||||
return True
|
||||
|
||||
print("Handshake fehlgeschlagen")
|
||||
return False
|
||||
|
||||
def send_command(self, encrypted_command):
|
||||
"""Sendet einen vorbereiteten verschlüsselten Befehl"""
|
||||
if not self.session_id:
|
||||
print("Keine Session-ID. Bitte zuerst Handshake durchführen.")
|
||||
return None
|
||||
|
||||
# Token aus der Wireshark-Aufnahme (könnte sich ändern, oder vom Gerät abhängen)
|
||||
token = "9DFAC92C53CEC92E67A9CB2E00B3CB2F"
|
||||
|
||||
secure_data = {
|
||||
"method": "securePassthrough",
|
||||
"params": {
|
||||
"request": encrypted_command
|
||||
}
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Referer": f"http://{self.device_ip}:80",
|
||||
"Accept": "application/json",
|
||||
"requestByApp": "true",
|
||||
"Content-Type": "application/json; charset=UTF-8",
|
||||
"Cookie": f"TP_SESSIONID={self.session_id}"
|
||||
}
|
||||
|
||||
response = requests.post(
|
||||
f"http://{self.device_ip}/app?token={token}",
|
||||
json=secure_data,
|
||||
headers=headers
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
if data["error_code"] == 0:
|
||||
# In einem echten Szenario würden wir die Antwort entschlüsseln
|
||||
encrypted_response = data["result"]["response"]
|
||||
print("Befehl erfolgreich gesendet")
|
||||
return encrypted_response
|
||||
|
||||
print("Fehler beim Senden des Befehls")
|
||||
return None
|
||||
|
||||
def turn_on(self):
|
||||
"""Schaltet die Steckdose ein"""
|
||||
return self.send_command(COMMAND_ON)
|
||||
|
||||
def turn_off(self):
|
||||
"""Schaltet die Steckdose aus"""
|
||||
return self.send_command(COMMAND_OFF)
|
||||
|
||||
# Verwendungsbeispiel
|
||||
if __name__ == "__main__":
|
||||
controller = TapoP115Controller("192.168.0.102")
|
||||
|
||||
# Handshake durchführen
|
||||
if controller.perform_handshake():
|
||||
# Steckdose einschalten
|
||||
controller.turn_on()
|
||||
|
||||
# Kurze Pause (im echten Code mit time.sleep)
|
||||
print("Steckdose ist eingeschaltet")
|
||||
|
||||
# Steckdose ausschalten
|
||||
controller.turn_off()
|
||||
print("Steckdose ist ausgeschaltet")
|
||||
9
backend/development/tests/tapo.py
Normal file
9
backend/development/tests/tapo.py
Normal file
@ -0,0 +1,9 @@
|
||||
from PyP100 import PyP100
|
||||
|
||||
p100 = PyP100.P100("192.168.0.102", "till.tomczak@mercedes-benz.com", "Agent045") #Creates a P100 plug object
|
||||
|
||||
p100.handshake() #Creates the cookies required for further methods
|
||||
p100.login() #Sends credentials to the plug and creates AES Key and IV for further methods
|
||||
|
||||
p100.turnOn() #Turns the connected plug on
|
||||
p100.turnOff() #Turns the connected plug off
|
||||
253
backend/development/tests/tests.py
Normal file
253
backend/development/tests/tests.py
Normal file
@ -0,0 +1,253 @@
|
||||
import unittest
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from app import app, db, User, Printer, PrintJob
|
||||
|
||||
class MYPBackendTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# Temporäre Datenbank für Tests
|
||||
self.db_fd, app.config['DATABASE'] = tempfile.mkstemp()
|
||||
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + app.config['DATABASE']
|
||||
app.config['TESTING'] = True
|
||||
self.app = app.test_client()
|
||||
|
||||
# Datenbank-Tabellen erstellen und Test-Daten einfügen
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
|
||||
# Admin-Benutzer erstellen
|
||||
admin = User(username='admin_test', email='admin@test.com', role='admin')
|
||||
admin.set_password('admin')
|
||||
db.session.add(admin)
|
||||
|
||||
# Normaler Benutzer erstellen
|
||||
user = User(username='user_test', email='user@test.com', role='user')
|
||||
user.set_password('user')
|
||||
db.session.add(user)
|
||||
|
||||
# Drucker erstellen
|
||||
printer1 = Printer(name='Printer 1', location='Room A', type='3D',
|
||||
status='available', description='Test printer 1')
|
||||
printer2 = Printer(name='Printer 2', location='Room B', type='3D',
|
||||
status='busy', description='Test printer 2')
|
||||
db.session.add(printer1)
|
||||
db.session.add(printer2)
|
||||
|
||||
# Job erstellen
|
||||
start_time = datetime.utcnow()
|
||||
end_time = start_time + timedelta(minutes=60)
|
||||
job = PrintJob(title='Test Job', start_time=start_time, end_time=end_time,
|
||||
duration=60, status='active', comments='Test job',
|
||||
user_id=2, printer_id=2)
|
||||
db.session.add(job)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
def tearDown(self):
|
||||
# Aufräumen nach dem Test
|
||||
os.close(self.db_fd)
|
||||
os.unlink(app.config['DATABASE'])
|
||||
|
||||
def get_token(self, username, password):
|
||||
response = self.app.post('/api/auth/login',
|
||||
data=json.dumps({'username': username, 'password': password}),
|
||||
content_type='application/json')
|
||||
data = json.loads(response.data)
|
||||
return data.get('token')
|
||||
|
||||
def test_login(self):
|
||||
# Test: Erfolgreicher Login
|
||||
response = self.app.post('/api/auth/login',
|
||||
data=json.dumps({'username': 'admin_test', 'password': 'admin'}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertIn('token', data)
|
||||
self.assertIn('user', data)
|
||||
|
||||
# Test: Fehlgeschlagener Login (falsches Passwort)
|
||||
response = self.app.post('/api/auth/login',
|
||||
data=json.dumps({'username': 'admin_test', 'password': 'wrong'}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
def test_register(self):
|
||||
# Test: Erfolgreiche Registrierung
|
||||
response = self.app.post('/api/auth/register',
|
||||
data=json.dumps({
|
||||
'username': 'new_user',
|
||||
'email': 'new@test.com',
|
||||
'password': 'password'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 201)
|
||||
|
||||
# Test: Doppelte Registrierung
|
||||
response = self.app.post('/api/auth/register',
|
||||
data=json.dumps({
|
||||
'username': 'new_user',
|
||||
'email': 'another@test.com',
|
||||
'password': 'password'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_get_printers(self):
|
||||
# Test: Drucker abrufen
|
||||
response = self.app.get('/api/printers')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(len(data), 2)
|
||||
|
||||
def test_get_single_printer(self):
|
||||
# Test: Einzelnen Drucker abrufen
|
||||
response = self.app.get('/api/printers/1')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['name'], 'Printer 1')
|
||||
|
||||
def test_create_printer(self):
|
||||
# Als Admin einen Drucker erstellen
|
||||
token = self.get_token('admin_test', 'admin')
|
||||
response = self.app.post('/api/printers',
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
data=json.dumps({
|
||||
'name': 'New Printer',
|
||||
'location': 'Room C',
|
||||
'type': '3D',
|
||||
'description': 'New test printer'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 201)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['name'], 'New Printer')
|
||||
|
||||
def test_update_printer(self):
|
||||
# Als Admin einen Drucker aktualisieren
|
||||
token = self.get_token('admin_test', 'admin')
|
||||
response = self.app.put('/api/printers/1',
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
data=json.dumps({
|
||||
'name': 'Updated Printer',
|
||||
'location': 'Room D'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['name'], 'Updated Printer')
|
||||
self.assertEqual(data['location'], 'Room D')
|
||||
|
||||
def test_delete_printer(self):
|
||||
# Als Admin einen Drucker löschen
|
||||
token = self.get_token('admin_test', 'admin')
|
||||
response = self.app.delete('/api/printers/1',
|
||||
headers={'Authorization': f'Bearer {token}'})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Überprüfen, ob der Drucker wirklich gelöscht wurde
|
||||
response = self.app.get('/api/printers/1')
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
def test_get_jobs_as_admin(self):
|
||||
# Als Admin alle Jobs abrufen
|
||||
token = self.get_token('admin_test', 'admin')
|
||||
response = self.app.get('/api/jobs',
|
||||
headers={'Authorization': f'Bearer {token}'})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(len(data), 1)
|
||||
|
||||
def test_get_jobs_as_user(self):
|
||||
# Als normaler Benutzer nur eigene Jobs abrufen
|
||||
token = self.get_token('user_test', 'user')
|
||||
response = self.app.get('/api/jobs',
|
||||
headers={'Authorization': f'Bearer {token}'})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(len(data), 1) # Der Benutzer hat einen Job
|
||||
|
||||
def test_create_job(self):
|
||||
# Als Benutzer einen Job erstellen
|
||||
token = self.get_token('user_test', 'user')
|
||||
response = self.app.post('/api/jobs',
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
data=json.dumps({
|
||||
'title': 'New Job',
|
||||
'printer_id': 1,
|
||||
'duration': 30,
|
||||
'comments': 'Test job creation'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 201)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['title'], 'New Job')
|
||||
self.assertEqual(data['duration'], 30)
|
||||
|
||||
def test_update_job(self):
|
||||
# Als Benutzer den eigenen Job aktualisieren
|
||||
token = self.get_token('user_test', 'user')
|
||||
response = self.app.put('/api/jobs/1',
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
data=json.dumps({
|
||||
'comments': 'Updated comments',
|
||||
'duration': 15 # Verlängerung
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['comments'], 'Updated comments')
|
||||
self.assertEqual(data['duration'], 75) # 60 + 15
|
||||
|
||||
def test_complete_job(self):
|
||||
# Als Benutzer einen Job als abgeschlossen markieren
|
||||
token = self.get_token('user_test', 'user')
|
||||
response = self.app.put('/api/jobs/1',
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
data=json.dumps({
|
||||
'status': 'completed'
|
||||
}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['status'], 'completed')
|
||||
|
||||
# Überprüfen, ob der Drucker wieder verfügbar ist
|
||||
response = self.app.get('/api/printers/2')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['status'], 'available')
|
||||
|
||||
def test_get_remaining_time(self):
|
||||
# Test: Verbleibende Zeit für einen aktiven Job abrufen
|
||||
response = self.app.get('/api/job/1/remaining-time')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertIn('remaining_minutes', data)
|
||||
# Der genaue Wert kann nicht überprüft werden, da er von der Zeit abhängt
|
||||
|
||||
def test_stats(self):
|
||||
# Als Admin Statistiken abrufen
|
||||
token = self.get_token('admin_test', 'admin')
|
||||
response = self.app.get('/api/stats',
|
||||
headers={'Authorization': f'Bearer {token}'})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertIn('printers', data)
|
||||
self.assertIn('jobs', data)
|
||||
self.assertIn('users', data)
|
||||
self.assertEqual(data['printers']['total'], 2)
|
||||
self.assertEqual(data['jobs']['total'], 1)
|
||||
self.assertEqual(data['users']['total'], 2)
|
||||
|
||||
def test_test_endpoint(self):
|
||||
# Test: API-Test-Endpunkt
|
||||
response = self.app.get('/api/test')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = json.loads(response.data)
|
||||
self.assertEqual(data['message'], 'MYP Backend API funktioniert!')
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user