refactor: split monolithic app.py into modular Flask application with factory pattern
- Created app/ package with 6 modules: - __init__.py: Application factory (create_app function) - routes.py: All route handlers (473 lines) - models.py: Data persistence layer (41 lines) - validators.py: Input validation functions (48 lines) - email_service.py: Email sending functions (85 lines) - utils.py: Utility functions (30 lines) - Implemented Flask application factory pattern - Created run.py as minimal entry point - Updated all test imports to use new module structure - Fixed template/static folder paths for package structure - All 102 tests passing - Improved maintainability with clear separation of concerns - Follows Flask best practices for scalable applications
This commit is contained in:
@@ -0,0 +1,58 @@
|
||||
"""
|
||||
Flask job application system - Application factory module.
|
||||
"""
|
||||
from pathlib import Path
|
||||
from flask import Flask, redirect, url_for, flash
|
||||
from flask_mail import Mail
|
||||
from flask_wtf.csrf import CSRFProtect, CSRFError
|
||||
from config import Config
|
||||
|
||||
# Initialize extensions (not bound to app yet)
|
||||
mail = Mail()
|
||||
csrf = CSRFProtect()
|
||||
|
||||
|
||||
def create_app(config_class=Config):
|
||||
"""
|
||||
Application factory function.
|
||||
Creates and configures the Flask application instance.
|
||||
|
||||
Args:
|
||||
config_class: Configuration class to use (defaults to Config)
|
||||
|
||||
Returns:
|
||||
Configured Flask application instance
|
||||
"""
|
||||
# Create Flask app instance
|
||||
# Specify template and static folders relative to project root (parent of app/)
|
||||
app = Flask(__name__,
|
||||
template_folder='../templates',
|
||||
static_folder='../static')
|
||||
|
||||
# Load configuration
|
||||
app.config.from_object(config_class)
|
||||
|
||||
# Initialize extensions with app
|
||||
mail.init_app(app)
|
||||
csrf.init_app(app)
|
||||
|
||||
# Ensure applications folder exists
|
||||
Path(app.config['APPLICATIONS_FOLDER']).mkdir(exist_ok=True)
|
||||
|
||||
# Register error handlers
|
||||
@app.errorhandler(CSRFError)
|
||||
def handle_csrf_error(e):
|
||||
"""Handle CSRF validation errors."""
|
||||
flash('Sicherheitsfehler: Die Sitzung ist abgelaufen. Bitte laden Sie die Seite neu und versuchen Sie es erneut.', 'error')
|
||||
return redirect(url_for('page1_email')), 400
|
||||
|
||||
@app.errorhandler(429)
|
||||
def handle_rate_limit_error(e):
|
||||
"""Handle rate limit errors."""
|
||||
return str(e), 429
|
||||
|
||||
# Register routes
|
||||
from app.routes import register_routes
|
||||
register_routes(app, mail)
|
||||
|
||||
return app
|
||||
@@ -0,0 +1,85 @@
|
||||
"""
|
||||
Email service functions for the Flask job application system.
|
||||
"""
|
||||
from datetime import datetime
|
||||
from flask import url_for, current_app
|
||||
from flask_mail import Message
|
||||
from app import mail
|
||||
|
||||
|
||||
def send_resume_email(email, session_id, job_name):
|
||||
"""Send email with resume link"""
|
||||
resume_link = url_for('resume_application', session_id=session_id, _external=True)
|
||||
|
||||
subject = f"Ihre Bewerbung bei {current_app.config['COMPANY_NAME']} - Link zum Fortsetzen"
|
||||
body = f"""Guten Tag,
|
||||
|
||||
vielen Dank für Ihr Interesse an der Position "{job_name}" bei {current_app.config['COMPANY_NAME']}.
|
||||
|
||||
Sie können Ihre Bewerbung jederzeit über den folgenden Link fortsetzen:
|
||||
{resume_link}
|
||||
|
||||
Dieser Link bleibt gültig und Sie können Ihre Bewerbung jederzeit bearbeiten.
|
||||
|
||||
Mit freundlichen Grüßen
|
||||
{current_app.config['COMPANY_NAME']}
|
||||
"""
|
||||
|
||||
try:
|
||||
msg = Message(subject=subject, recipients=[email], body=body)
|
||||
mail.send(msg)
|
||||
return True
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"Failed to send email: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def send_hr_notification(session_id, app_data):
|
||||
"""Send HR notification email when application is submitted"""
|
||||
if not current_app.config.get('HR_EMAIL'):
|
||||
current_app.logger.warning("HR_EMAIL not configured, skipping HR notification")
|
||||
return False
|
||||
|
||||
# Build application URL
|
||||
application_url = f"{current_app.config['APPLICATION_URL_BASE']}/application/{session_id}/"
|
||||
|
||||
# Get personal info
|
||||
personal_info = app_data.get('personal_info', {})
|
||||
firstname = personal_info.get('firstname', 'N/A')
|
||||
name = personal_info.get('name', 'N/A')
|
||||
email = app_data.get('email', 'N/A')
|
||||
job_name = app_data.get('job_name', 'N/A')
|
||||
file_count = len(app_data.get('uploaded_files', []))
|
||||
|
||||
# Format timestamp
|
||||
submitted_at = app_data.get('submitted_at', datetime.now().isoformat())
|
||||
try:
|
||||
timestamp = datetime.fromisoformat(submitted_at).strftime('%d.%m.%Y %H:%M')
|
||||
except (ValueError, TypeError):
|
||||
timestamp = submitted_at
|
||||
|
||||
subject = f"Neue Bewerbung eingegangen: {job_name}"
|
||||
body = f"""Sehr geehrtes HR-Team,
|
||||
|
||||
es ist eine neue Bewerbung eingegangen:
|
||||
|
||||
Position: {job_name}
|
||||
Name: {firstname} {name}
|
||||
E-Mail: {email}
|
||||
Eingereicht am: {timestamp}
|
||||
Anzahl der hochgeladenen Dokumente: {file_count}
|
||||
|
||||
Sie können die vollständige Bewerbung hier einsehen:
|
||||
{application_url}
|
||||
|
||||
Mit freundlichen Grüßen,
|
||||
Ihr Bewerbungssystem
|
||||
"""
|
||||
|
||||
try:
|
||||
msg = Message(subject=subject, recipients=[current_app.config['HR_EMAIL']], body=body)
|
||||
mail.send(msg)
|
||||
return True
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"Failed to send HR notification email: {e}")
|
||||
return False
|
||||
@@ -0,0 +1,41 @@
|
||||
"""
|
||||
Data models and storage functions for the Flask job application system.
|
||||
"""
|
||||
import os
|
||||
import yaml
|
||||
from pathlib import Path
|
||||
from flask import current_app
|
||||
|
||||
|
||||
def get_application_path(session_id):
|
||||
"""Get the path to an application folder"""
|
||||
return os.path.join(current_app.config['APPLICATIONS_FOLDER'], session_id)
|
||||
|
||||
|
||||
def get_data_file_path(session_id):
|
||||
"""Get the path to the application data YAML file"""
|
||||
return os.path.join(get_application_path(session_id), 'data.yaml')
|
||||
|
||||
|
||||
def get_attachments_path(session_id):
|
||||
"""Get the path to the attachments folder"""
|
||||
return os.path.join(get_application_path(session_id), 'attachments')
|
||||
|
||||
|
||||
def load_application_data(session_id):
|
||||
"""Load application data from YAML file"""
|
||||
data_file = get_data_file_path(session_id)
|
||||
if os.path.exists(data_file):
|
||||
with open(data_file, 'r', encoding='utf-8') as f:
|
||||
return yaml.safe_load(f)
|
||||
return None
|
||||
|
||||
|
||||
def save_application_data(session_id, data):
|
||||
"""Save application data to YAML file"""
|
||||
app_path = get_application_path(session_id)
|
||||
Path(app_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
data_file = get_data_file_path(session_id)
|
||||
with open(data_file, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(data, f, allow_unicode=True, default_flow_style=False)
|
||||
+473
@@ -0,0 +1,473 @@
|
||||
"""
|
||||
Route handlers for the Flask job application system.
|
||||
"""
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from flask import render_template, request, redirect, url_for, flash, session, send_from_directory, abort
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
from app.models import (
|
||||
load_application_data,
|
||||
save_application_data,
|
||||
get_attachments_path
|
||||
)
|
||||
from app.validators import (
|
||||
validate_email,
|
||||
validate_phone,
|
||||
validate_year,
|
||||
validate_zip,
|
||||
allowed_file
|
||||
)
|
||||
from app.email_service import send_resume_email, send_hr_notification
|
||||
from app.utils import check_rate_limit
|
||||
|
||||
|
||||
def register_routes(app, mail):
|
||||
"""Register all application routes"""
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
"""Redirect to apply page"""
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
|
||||
@app.route('/apply')
|
||||
def page1_email():
|
||||
"""Page 1: Email capture and session initialization"""
|
||||
job_name = request.args.get('job', 'Offene Position')
|
||||
return render_template('page1_email.html', job_name=job_name)
|
||||
|
||||
|
||||
@app.route('/apply/submit-email', methods=['POST'])
|
||||
def submit_email():
|
||||
"""Process email submission and create session"""
|
||||
# Check rate limit
|
||||
allowed, wait_seconds = check_rate_limit()
|
||||
if not allowed:
|
||||
flash(f'Bitte warten Sie noch {wait_seconds} Sekunden vor der nächsten Eingabe.', 'error')
|
||||
return redirect(url_for('page1_email')), 429
|
||||
|
||||
email = request.form.get('email', '').strip()
|
||||
job_name = request.form.get('job_name', 'Offene Position')
|
||||
|
||||
# Validate email
|
||||
if not validate_email(email):
|
||||
flash('Bitte geben Sie eine gültige E-Mail-Adresse ein.', 'error')
|
||||
return redirect(url_for('page1_email', job=job_name))
|
||||
|
||||
# Generate session ID
|
||||
session_id = str(uuid.uuid4())
|
||||
|
||||
# Create application data
|
||||
app_data = {
|
||||
'session_id': session_id,
|
||||
'email': email,
|
||||
'job_name': job_name,
|
||||
'current_page': 2,
|
||||
'created_at': datetime.now().isoformat(),
|
||||
'updated_at': datetime.now().isoformat(),
|
||||
'personal_info': {},
|
||||
'motivation_answers': {},
|
||||
'uploaded_files': []
|
||||
}
|
||||
|
||||
# Save application data
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
# Send resume email
|
||||
if send_resume_email(email, session_id, job_name):
|
||||
flash('Eine E-Mail mit einem Link zum Fortsetzen wurde an Ihre Adresse gesendet.', 'success')
|
||||
else:
|
||||
flash('Warnung: Die E-Mail konnte nicht gesendet werden. Bitte notieren Sie sich diese URL.', 'warning')
|
||||
|
||||
# Store session ID in Flask session
|
||||
session['application_id'] = session_id
|
||||
|
||||
return redirect(url_for('page2_personal', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/personal')
|
||||
def page2_personal(session_id):
|
||||
"""Page 2: Basic personal information"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
return render_template('page2_personal.html',
|
||||
session_id=session_id,
|
||||
job_name=app_data['job_name'],
|
||||
data=app_data.get('personal_info', {}))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/submit-personal', methods=['POST'])
|
||||
def submit_personal(session_id):
|
||||
"""Process personal information submission"""
|
||||
# Check rate limit
|
||||
allowed, wait_seconds = check_rate_limit()
|
||||
if not allowed:
|
||||
flash(f'Bitte warten Sie noch {wait_seconds} Sekunden vor der nächsten Eingabe.', 'error')
|
||||
return redirect(url_for('page2_personal', session_id=session_id)), 429
|
||||
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
# Get form data
|
||||
name = request.form.get('name', '').strip()
|
||||
firstname = request.form.get('firstname', '').strip()
|
||||
address = request.form.get('address', '').strip()
|
||||
zip_code = request.form.get('zip_code', '').strip()
|
||||
city = request.form.get('city', '').strip()
|
||||
phone = request.form.get('phone', '').strip()
|
||||
birth_year = request.form.get('birth_year', '').strip()
|
||||
civil_status = request.form.get('civil_status', '').strip()
|
||||
|
||||
# Validate required fields
|
||||
errors = []
|
||||
|
||||
if not name or len(name) > app.config['MAX_STRING_LENGTH']:
|
||||
errors.append('Name ist erforderlich und darf maximal 255 Zeichen lang sein.')
|
||||
|
||||
if not firstname or len(firstname) > app.config['MAX_STRING_LENGTH']:
|
||||
errors.append('Vorname ist erforderlich und darf maximal 255 Zeichen lang sein.')
|
||||
|
||||
if not address or len(address) > app.config['MAX_STRING_LENGTH']:
|
||||
errors.append('Adresse ist erforderlich und darf maximal 255 Zeichen lang sein.')
|
||||
|
||||
if not zip_code or not validate_zip(zip_code):
|
||||
errors.append('PLZ ist erforderlich und darf maximal 10 Ziffern lang sein.')
|
||||
|
||||
if not city or len(city) > app.config['MAX_STRING_LENGTH']:
|
||||
errors.append('Stadt ist erforderlich und darf maximal 255 Zeichen lang sein.')
|
||||
|
||||
if not phone or not validate_phone(phone):
|
||||
errors.append('Telefonnummer ist erforderlich und muss im internationalen Format sein (z.B. +41 79 123 45 67).')
|
||||
|
||||
if not birth_year or not validate_year(birth_year):
|
||||
errors.append(f'Geburtsjahr ist erforderlich und muss zwischen {app.config["MIN_BIRTH_YEAR"]} und {app.config["MAX_BIRTH_YEAR"]} liegen.')
|
||||
|
||||
if civil_status and len(civil_status) > app.config['MAX_STRING_LENGTH']:
|
||||
errors.append('Zivilstand darf maximal 255 Zeichen lang sein.')
|
||||
|
||||
if errors:
|
||||
for error in errors:
|
||||
flash(error, 'error')
|
||||
return redirect(url_for('page2_personal', session_id=session_id))
|
||||
|
||||
# Save personal information
|
||||
app_data['personal_info'] = {
|
||||
'name': name,
|
||||
'firstname': firstname,
|
||||
'address': address,
|
||||
'zip_code': zip_code,
|
||||
'city': city,
|
||||
'phone': phone,
|
||||
'birth_year': birth_year,
|
||||
'civil_status': civil_status
|
||||
}
|
||||
app_data['current_page'] = 3
|
||||
app_data['updated_at'] = datetime.now().isoformat()
|
||||
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
return redirect(url_for('page3_motivation', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/motivation')
|
||||
def page3_motivation(session_id):
|
||||
"""Page 3: Motivation and qualification questions"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
return render_template('page3_motivation.html',
|
||||
session_id=session_id,
|
||||
job_name=app_data['job_name'],
|
||||
data=app_data.get('motivation_answers', {}))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/submit-motivation', methods=['POST'])
|
||||
def submit_motivation(session_id):
|
||||
"""Process motivation questions submission"""
|
||||
# Check rate limit
|
||||
allowed, wait_seconds = check_rate_limit()
|
||||
if not allowed:
|
||||
flash(f'Bitte warten Sie noch {wait_seconds} Sekunden vor der nächsten Eingabe.', 'error')
|
||||
return redirect(url_for('page3_motivation', session_id=session_id)), 429
|
||||
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
# Get form data
|
||||
current_job = request.form.get('current_job', '').strip()
|
||||
motivation = request.form.get('motivation', '').strip()
|
||||
qualifications = request.form.get('qualifications', '').strip()
|
||||
salary = request.form.get('salary', '').strip()
|
||||
|
||||
# Validate length
|
||||
errors = []
|
||||
|
||||
if current_job and len(current_job) > app.config['MAX_TEXT_AREA_LENGTH']:
|
||||
errors.append(f'Die Beschreibung Ihrer beruflichen Situation darf maximal {app.config["MAX_TEXT_AREA_LENGTH"]} Zeichen lang sein.')
|
||||
|
||||
if motivation and len(motivation) > app.config['MAX_TEXT_AREA_LENGTH']:
|
||||
errors.append(f'Die Motivationsbeschreibung darf maximal {app.config["MAX_TEXT_AREA_LENGTH"]} Zeichen lang sein.')
|
||||
|
||||
if qualifications and len(qualifications) > app.config['MAX_TEXT_AREA_LENGTH']:
|
||||
errors.append(f'Die Qualifikationsbeschreibung darf maximal {app.config["MAX_TEXT_AREA_LENGTH"]} Zeichen lang sein.')
|
||||
|
||||
if salary and len(salary) > app.config['MAX_TEXT_AREA_LENGTH']:
|
||||
errors.append(f'Die Gehaltsvorstellungen dürfen maximal {app.config["MAX_TEXT_AREA_LENGTH"]} Zeichen lang sein.')
|
||||
|
||||
if errors:
|
||||
for error in errors:
|
||||
flash(error, 'error')
|
||||
return redirect(url_for('page3_motivation', session_id=session_id))
|
||||
|
||||
# Save motivation answers
|
||||
app_data['motivation_answers'] = {
|
||||
'current_job': current_job,
|
||||
'motivation': motivation,
|
||||
'qualifications': qualifications,
|
||||
'salary': salary
|
||||
}
|
||||
app_data['current_page'] = 4
|
||||
app_data['updated_at'] = datetime.now().isoformat()
|
||||
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/upload')
|
||||
def page4_upload(session_id):
|
||||
"""Page 4: Document upload"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
return render_template('page4_upload.html',
|
||||
session_id=session_id,
|
||||
job_name=app_data['job_name'],
|
||||
uploaded_files=app_data.get('uploaded_files', []),
|
||||
max_files=app.config['MAX_FILES'])
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/upload-file', methods=['POST'])
|
||||
def upload_file(session_id):
|
||||
"""Handle file upload"""
|
||||
# Check rate limit
|
||||
allowed, wait_seconds = check_rate_limit()
|
||||
if not allowed:
|
||||
flash(f'Bitte warten Sie noch {wait_seconds} Sekunden vor der nächsten Eingabe.', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id)), 429
|
||||
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
# Check file count
|
||||
if len(app_data.get('uploaded_files', [])) >= app.config['MAX_FILES']:
|
||||
flash(f'Sie können maximal {app.config["MAX_FILES"]} Dokumente hochladen.', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
# Check if file was uploaded
|
||||
if 'file' not in request.files:
|
||||
flash('Keine Datei ausgewählt.', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
file = request.files['file']
|
||||
|
||||
if file.filename == '':
|
||||
flash('Keine Datei ausgewählt.', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
# Validate file
|
||||
if not allowed_file(file.filename):
|
||||
flash('Ungültiger Dateityp. Erlaubte Formate: PDF, DOC, DOCX, TXT, JPG, JPEG, PNG', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
# Check file size
|
||||
file.seek(0, os.SEEK_END)
|
||||
file_size = file.tell()
|
||||
file.seek(0)
|
||||
|
||||
if file_size > app.config['MAX_FILE_SIZE']:
|
||||
flash(f'Die Datei ist zu groß. Maximale Größe: 4 MB', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
# Save file
|
||||
filename = secure_filename(file.filename)
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
unique_filename = f"{timestamp}_{filename}"
|
||||
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
Path(attachments_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
file_path = os.path.join(attachments_path, unique_filename)
|
||||
file.save(file_path)
|
||||
|
||||
# Update application data
|
||||
if 'uploaded_files' not in app_data:
|
||||
app_data['uploaded_files'] = []
|
||||
|
||||
app_data['uploaded_files'].append({
|
||||
'original_name': filename,
|
||||
'stored_name': unique_filename,
|
||||
'uploaded_at': datetime.now().isoformat(),
|
||||
'size': file_size
|
||||
})
|
||||
app_data['updated_at'] = datetime.now().isoformat()
|
||||
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
flash('Datei erfolgreich hochgeladen.', 'success')
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/remove-file/<int:file_index>', methods=['POST'])
|
||||
def remove_file(session_id, file_index):
|
||||
"""Remove an uploaded file"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
uploaded_files = app_data.get('uploaded_files', [])
|
||||
|
||||
if 0 <= file_index < len(uploaded_files):
|
||||
file_info = uploaded_files[file_index]
|
||||
|
||||
# Delete physical file
|
||||
file_path = os.path.join(get_attachments_path(session_id), file_info['stored_name'])
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
|
||||
# Remove from data
|
||||
uploaded_files.pop(file_index)
|
||||
app_data['updated_at'] = datetime.now().isoformat()
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
flash('Datei erfolgreich entfernt.', 'success')
|
||||
else:
|
||||
flash('Datei nicht gefunden.', 'error')
|
||||
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/submit-application', methods=['POST'])
|
||||
def submit_application(session_id):
|
||||
"""Submit final application"""
|
||||
# Check rate limit
|
||||
allowed, wait_seconds = check_rate_limit()
|
||||
if not allowed:
|
||||
flash(f'Bitte warten Sie noch {wait_seconds} Sekunden vor der nächsten Eingabe.', 'error')
|
||||
return redirect(url_for('page4_upload', session_id=session_id)), 429
|
||||
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
# Mark as submitted
|
||||
app_data['current_page'] = 5
|
||||
app_data['submitted_at'] = datetime.now().isoformat()
|
||||
app_data['updated_at'] = datetime.now().isoformat()
|
||||
app_data['status'] = 'submitted'
|
||||
|
||||
save_application_data(session_id, app_data)
|
||||
|
||||
# Send HR notification
|
||||
send_hr_notification(session_id, app_data)
|
||||
|
||||
return redirect(url_for('page5_confirmation', session_id=session_id))
|
||||
|
||||
|
||||
@app.route('/apply/<session_id>/confirmation')
|
||||
def page5_confirmation(session_id):
|
||||
"""Page 5: Confirmation page"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
return render_template('page5_confirmation.html',
|
||||
session_id=session_id,
|
||||
job_name=app_data['job_name'],
|
||||
email=app_data['email'])
|
||||
|
||||
|
||||
@app.route('/application/<session_id>/')
|
||||
def view_application(session_id):
|
||||
"""View complete application details (for HR)"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
abort(404)
|
||||
|
||||
return render_template('application_view.html',
|
||||
session_id=session_id,
|
||||
app_data=app_data)
|
||||
|
||||
|
||||
@app.route('/application/<session_id>/download/<filename>')
|
||||
def download_file(session_id, filename):
|
||||
"""Download an uploaded file"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
abort(404)
|
||||
|
||||
# Verify the file is in the uploaded files list
|
||||
uploaded_files = app_data.get('uploaded_files', [])
|
||||
file_found = False
|
||||
|
||||
for file_info in uploaded_files:
|
||||
if file_info.get('stored_name') == filename:
|
||||
file_found = True
|
||||
break
|
||||
|
||||
if not file_found:
|
||||
abort(404)
|
||||
|
||||
# Serve the file
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
file_path = os.path.join(attachments_path, filename)
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
abort(404)
|
||||
|
||||
return send_from_directory(attachments_path, filename, as_attachment=True)
|
||||
|
||||
|
||||
@app.route('/resume/<session_id>')
|
||||
def resume_application(session_id):
|
||||
"""Resume application from email link"""
|
||||
app_data = load_application_data(session_id)
|
||||
if not app_data:
|
||||
flash('Bewerbung nicht gefunden.', 'error')
|
||||
return redirect(url_for('page1_email'))
|
||||
|
||||
# Store session ID in Flask session
|
||||
session['application_id'] = session_id
|
||||
|
||||
# Redirect to current page
|
||||
current_page = app_data.get('current_page', 2)
|
||||
|
||||
if current_page == 2:
|
||||
return redirect(url_for('page2_personal', session_id=session_id))
|
||||
elif current_page == 3:
|
||||
return redirect(url_for('page3_motivation', session_id=session_id))
|
||||
elif current_page == 4:
|
||||
return redirect(url_for('page4_upload', session_id=session_id))
|
||||
elif current_page == 5:
|
||||
return redirect(url_for('page5_confirmation', session_id=session_id))
|
||||
else:
|
||||
return redirect(url_for('page2_personal', session_id=session_id))
|
||||
@@ -0,0 +1,30 @@
|
||||
"""
|
||||
Utility functions for the Flask job application system.
|
||||
"""
|
||||
from datetime import datetime
|
||||
from flask import session, current_app
|
||||
|
||||
|
||||
def check_rate_limit():
|
||||
"""
|
||||
Check if user is submitting forms too quickly.
|
||||
Returns (allowed: bool, wait_seconds: int)
|
||||
"""
|
||||
now = datetime.now()
|
||||
last_submit_str = session.get('last_submission_time')
|
||||
|
||||
if last_submit_str:
|
||||
try:
|
||||
last_submit = datetime.fromisoformat(last_submit_str)
|
||||
elapsed = (now - last_submit).total_seconds()
|
||||
|
||||
if elapsed < current_app.config['RATE_LIMIT_SECONDS']:
|
||||
wait_seconds = int(current_app.config['RATE_LIMIT_SECONDS'] - elapsed) + 1
|
||||
return False, wait_seconds
|
||||
except (ValueError, TypeError):
|
||||
# Invalid timestamp, allow submission
|
||||
pass
|
||||
|
||||
# Update last submission time
|
||||
session['last_submission_time'] = now.isoformat()
|
||||
return True, 0
|
||||
@@ -0,0 +1,48 @@
|
||||
"""
|
||||
Validation functions for the Flask job application system.
|
||||
"""
|
||||
import re
|
||||
from flask import current_app
|
||||
|
||||
|
||||
def validate_email(email):
|
||||
"""Validate email format"""
|
||||
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
|
||||
return re.match(pattern, email) is not None
|
||||
|
||||
|
||||
def validate_phone(phone):
|
||||
"""Validate international phone format"""
|
||||
# Accepts formats like: +41 79 123 45 67, +41791234567, etc.
|
||||
# Remove spaces and check if it matches the pattern
|
||||
pattern = r'^\+\d{1,3}[\s\d]{1,20}$'
|
||||
if not re.match(pattern, phone):
|
||||
return False
|
||||
# Ensure there are at least some digits after the country code
|
||||
digits_only = re.sub(r'\s', '', phone)
|
||||
return len(digits_only) >= 5 # At least +XX XXX format
|
||||
|
||||
|
||||
def validate_year(year):
|
||||
"""Validate birth year"""
|
||||
try:
|
||||
year_int = int(year)
|
||||
return (current_app.config['MIN_BIRTH_YEAR'] <= year_int <= current_app.config['MAX_BIRTH_YEAR']
|
||||
and len(str(year)) == 4)
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def validate_zip(zip_code):
|
||||
"""Validate ZIP code"""
|
||||
try:
|
||||
zip_int = int(zip_code)
|
||||
return len(str(zip_code)) <= current_app.config['MAX_ZIP_DIGITS']
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def allowed_file(filename):
|
||||
"""Check if file extension is allowed"""
|
||||
return '.' in filename and \
|
||||
filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']
|
||||
@@ -0,0 +1,258 @@
|
||||
<objective>
|
||||
Refactor the monolithic app.py (664 lines) into a well-organized, modular Flask application structure using the application factory pattern. The goal is to improve code maintainability, testability, and separation of concerns while preserving all existing functionality.
|
||||
|
||||
This refactoring will make the codebase easier to navigate, enable better unit testing of individual components, and follow Flask best practices for scalable applications.
|
||||
</objective>
|
||||
|
||||
<context>
|
||||
Current state:
|
||||
- Single app.py file contains routes, validation logic, email functions, storage helpers, and Flask app initialization
|
||||
- 102 passing tests that import directly from app module
|
||||
- Flask application with CSRF protection, rate limiting, file uploads, and email notifications
|
||||
- Tech stack: Flask 3.0.0, Flask-Mail, Flask-WTF, PyYAML
|
||||
|
||||
Target state:
|
||||
- Organized app/ package structure with separated concerns
|
||||
- Flask application factory pattern (create_app function)
|
||||
- Maintained backward compatibility for all existing functionality
|
||||
- All 102 tests updated and passing
|
||||
|
||||
Review the current implementation:
|
||||
@app.py
|
||||
</context>
|
||||
|
||||
<requirements>
|
||||
1. **Create app/ package structure**:
|
||||
- `app/__init__.py` - Application factory (create_app function), Flask app initialization
|
||||
- `app/routes.py` - All route handlers (@app.route decorators)
|
||||
- `app/models.py` - Data models and storage functions (save/load application data)
|
||||
- `app/validators.py` - All validation functions (email, phone, year, zip, file extensions)
|
||||
- `app/email_service.py` - Email-related functions (send_resume_email, send_hr_notification)
|
||||
- `app/utils.py` - Helper functions (path helpers, rate limiting, file handling)
|
||||
|
||||
2. **Implement Flask application factory pattern**:
|
||||
- Create `create_app(config_class=Config)` function in `app/__init__.py`
|
||||
- Initialize extensions (mail, csrf) within factory
|
||||
- Register blueprints or routes within factory
|
||||
- Update main entry point to use factory
|
||||
|
||||
3. **Maintain all existing functionality**:
|
||||
- All routes must work identically
|
||||
- CSRF protection, rate limiting, file uploads, email notifications unchanged
|
||||
- Session handling, flash messages, redirects work as before
|
||||
- Error handlers preserved
|
||||
|
||||
4. **Update test suite**:
|
||||
- Modify test imports to use new module structure
|
||||
- Update fixtures to work with application factory
|
||||
- Ensure all 102 tests pass after refactoring
|
||||
- No changes to test logic, only imports and app initialization
|
||||
|
||||
5. **Update main entry point**:
|
||||
- Create `run.py` or update existing entry point to use `create_app()`
|
||||
- Ensure development server still runs with `python run.py` or `flask run`
|
||||
|
||||
6. **Preserve imports and dependencies**:
|
||||
- Keep all existing imports (Flask, yaml, werkzeug, etc.)
|
||||
- Maintain proper circular import prevention
|
||||
- Use relative imports within app/ package
|
||||
</requirements>
|
||||
|
||||
<implementation>
|
||||
**Module separation strategy**:
|
||||
|
||||
1. **app/__init__.py** should contain:
|
||||
- Import statements for Flask, extensions
|
||||
- Extension instances (mail, csrf) initialized but not bound
|
||||
- `create_app(config_class=Config)` function that:
|
||||
- Creates Flask app instance
|
||||
- Loads configuration
|
||||
- Initializes extensions (mail.init_app, csrf.init_app)
|
||||
- Registers routes (import and register blueprint or routes)
|
||||
- Registers error handlers
|
||||
- Returns app instance
|
||||
|
||||
2. **app/routes.py** should contain:
|
||||
- All @app.route decorated functions
|
||||
- Import necessary validators, models, email_service, utils
|
||||
- Either use Blueprint or accept app parameter for route registration
|
||||
- Preserve all route logic exactly as is
|
||||
|
||||
3. **app/models.py** should contain:
|
||||
- Path helper functions (get_application_path, get_data_file_path, get_attachments_path)
|
||||
- Data persistence functions (load_application_data, save_application_data)
|
||||
- Any data structure or schema-related code
|
||||
|
||||
4. **app/validators.py** should contain:
|
||||
- validate_email, validate_phone, validate_year, validate_zip
|
||||
- allowed_file function
|
||||
- Each function should be independent and easily testable
|
||||
|
||||
5. **app/email_service.py** should contain:
|
||||
- send_resume_email function
|
||||
- send_hr_notification function
|
||||
- Requires access to mail instance and app.config
|
||||
- Use current_app for accessing config when needed
|
||||
|
||||
6. **app/utils.py** should contain:
|
||||
- check_rate_limit function
|
||||
- Any other utility functions not fitting elsewhere
|
||||
|
||||
**Important considerations**:
|
||||
|
||||
- **Circular imports**: To avoid circular imports between routes and models:
|
||||
- Routes import from models, validators, email_service, utils
|
||||
- Models should NOT import from routes
|
||||
- Use `from flask import current_app` to access app instance in helper functions
|
||||
|
||||
- **Extension access**: For mail, csrf in separate modules:
|
||||
- Initialize extensions in `app/__init__.py`
|
||||
- Import and use them in routes/email_service: `from app import mail, csrf`
|
||||
- OR pass as parameters where needed
|
||||
|
||||
- **Configuration access**: Use `current_app.config` in utility functions instead of `app.config`
|
||||
|
||||
- **Route registration**: Two approaches (choose one):
|
||||
- Blueprint approach: Create Blueprint in routes.py, register in factory
|
||||
- Direct approach: Import route functions and call them with app instance
|
||||
|
||||
- **Error handlers**: Register in `create_app()` factory function
|
||||
|
||||
**WHY this structure**:
|
||||
- Separation of concerns makes each module independently testable
|
||||
- Application factory enables multiple app instances for testing
|
||||
- Smaller files are easier to navigate and maintain
|
||||
- Follows Flask community best practices for production applications
|
||||
- Enables future scaling (add more routes, validators, services without bloating single file)
|
||||
</implementation>
|
||||
|
||||
<migration_steps>
|
||||
Execute refactoring in this order to minimize breaking changes:
|
||||
|
||||
1. **Create app/ package**:
|
||||
- Create `app/` directory
|
||||
- Create `app/__init__.py` with imports and extension initialization
|
||||
|
||||
2. **Extract validators** (least dependencies):
|
||||
- Move validation functions to `app/validators.py`
|
||||
- Update imports in app.py
|
||||
- Run tests to verify
|
||||
|
||||
3. **Extract models** (storage layer):
|
||||
- Move path helpers and data functions to `app/models.py`
|
||||
- Update imports in app.py
|
||||
- Run tests to verify
|
||||
|
||||
4. **Extract email service**:
|
||||
- Move email functions to `app/email_service.py`
|
||||
- Ensure mail instance is accessible
|
||||
- Update imports in app.py
|
||||
- Run tests to verify
|
||||
|
||||
5. **Extract utilities**:
|
||||
- Move rate limiting and other utils to `app/utils.py`
|
||||
- Update imports in app.py
|
||||
- Run tests to verify
|
||||
|
||||
6. **Extract routes**:
|
||||
- Move all route handlers to `app/routes.py`
|
||||
- Keep route registration in app.py temporarily
|
||||
- Run tests to verify
|
||||
|
||||
7. **Implement application factory**:
|
||||
- Create `create_app()` in `app/__init__.py`
|
||||
- Move app initialization into factory
|
||||
- Update `app.py` or create `run.py` to use factory
|
||||
- Run tests to verify
|
||||
|
||||
8. **Update all tests**:
|
||||
- Update imports in all test files
|
||||
- Update app fixture in conftest.py to use factory
|
||||
- Run full test suite
|
||||
- Fix any remaining import issues
|
||||
|
||||
9. **Final cleanup**:
|
||||
- Remove old app.py or convert to minimal run.py
|
||||
- Verify all functionality works
|
||||
- Run full test suite one final time
|
||||
</migration_steps>
|
||||
|
||||
<test_updates>
|
||||
Update test files to work with new structure:
|
||||
|
||||
**tests/conftest.py** changes:
|
||||
```python
|
||||
from app import create_app
|
||||
from app import mail
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
"""Create and configure a test Flask application instance."""
|
||||
test_app = create_app()
|
||||
# ... rest of configuration
|
||||
yield test_app
|
||||
```
|
||||
|
||||
**All test files** should update imports:
|
||||
- Change `from app import validate_email` → `from app.validators import validate_email`
|
||||
- Change `from app import send_resume_email` → `from app.email_service import send_resume_email`
|
||||
- Change `from app import load_application_data` → `from app.models import load_application_data`
|
||||
- And so on for all imports
|
||||
|
||||
Run tests after each major change to catch import errors early.
|
||||
</test_updates>
|
||||
|
||||
<output>
|
||||
Create the following new files:
|
||||
|
||||
- `./app/__init__.py` - Application factory and extension initialization
|
||||
- `./app/routes.py` - All route handlers
|
||||
- `./app/models.py` - Data persistence layer
|
||||
- `./app/validators.py` - Validation functions
|
||||
- `./app/email_service.py` - Email sending functions
|
||||
- `./app/utils.py` - Utility functions (rate limiting, etc.)
|
||||
|
||||
Modify:
|
||||
- `./app.py` → Convert to `./run.py` (or keep as minimal entry point using factory)
|
||||
- `./tests/conftest.py` - Update app fixture to use create_app()
|
||||
- `./tests/*.py` - Update all imports to new module structure
|
||||
|
||||
Delete after successful migration:
|
||||
- Original `./app.py` (only if converted to run.py)
|
||||
</output>
|
||||
|
||||
<verification>
|
||||
Before declaring complete, verify:
|
||||
|
||||
1. **All tests pass**: Run `pytest tests/ -v` and confirm all 102 tests pass
|
||||
2. **Application runs**: Start server with `python run.py` or `flask run` and verify pages load
|
||||
3. **No circular imports**: Python can import app package without errors
|
||||
4. **All routes work**: Test critical user flow:
|
||||
- Submit email on page 1
|
||||
- Fill personal info on page 2
|
||||
- Fill motivation on page 3
|
||||
- Upload file and submit on page 4
|
||||
- View confirmation on page 5
|
||||
- Access /application/<session_id>/ view
|
||||
- Download uploaded file
|
||||
5. **Email functionality**: HR notification sent on submission (check logs)
|
||||
6. **No broken imports**: All test files import successfully
|
||||
|
||||
If any tests fail, carefully review:
|
||||
- Import statements in test files
|
||||
- Application context in helper functions (use current_app)
|
||||
- Extension initialization in factory
|
||||
- Route registration in factory
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- app/ package created with 6 modules (__init__, routes, models, validators, email_service, utils)
|
||||
- Application factory pattern implemented (create_app function)
|
||||
- Original app.py converted to minimal run.py entry point
|
||||
- All 102 tests passing with updated imports
|
||||
- Application runs and all features work identically
|
||||
- Code is more maintainable with clear separation of concerns
|
||||
- No circular import errors
|
||||
- Flask development server starts successfully
|
||||
</success_criteria>
|
||||
Completed: Sa 27 Dez 2025 22:29:29 CET
|
||||
@@ -0,0 +1,9 @@
|
||||
"""
|
||||
Entry point for the Flask job application system.
|
||||
"""
|
||||
from app import create_app
|
||||
|
||||
app = create_app()
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(debug=True)
|
||||
+8
-3
@@ -8,7 +8,7 @@ from io import BytesIO
|
||||
from datetime import datetime
|
||||
import pytest
|
||||
import yaml
|
||||
from app import app as flask_app
|
||||
from app import create_app
|
||||
from flask_mail import Mail
|
||||
|
||||
|
||||
@@ -18,6 +18,9 @@ def app():
|
||||
# Create a temporary directory for test applications
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
|
||||
# Create test Flask app using factory
|
||||
flask_app = create_app()
|
||||
|
||||
# Configure app for testing
|
||||
flask_app.config['TESTING'] = True
|
||||
flask_app.config['SECRET_KEY'] = 'test-secret-key'
|
||||
@@ -133,7 +136,9 @@ def mock_mail(app, monkeypatch):
|
||||
self.sent_messages.append(message)
|
||||
|
||||
mock = MockMail()
|
||||
monkeypatch.setattr('app.mail', mock)
|
||||
# Patch the mail instance in the app package (used by email_service)
|
||||
import app.email_service
|
||||
monkeypatch.setattr(app.email_service, 'mail', mock)
|
||||
return mock
|
||||
|
||||
|
||||
@@ -142,7 +147,7 @@ def create_test_application(app, temp_applications_dir):
|
||||
"""Factory fixture to create test application data."""
|
||||
def _create(session_id='test-123', **kwargs):
|
||||
"""Create a test application with the given session ID and data."""
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
|
||||
data = {
|
||||
'session_id': session_id,
|
||||
|
||||
@@ -174,7 +174,7 @@ class TestFileDownload:
|
||||
session_id = 'test-download-success'
|
||||
|
||||
# Create application directory and attachment
|
||||
from app import get_attachments_path
|
||||
from app.models import get_attachments_path
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
Path(attachments_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -215,7 +215,7 @@ class TestFileDownload:
|
||||
session_id = 'test-download-unauthorized'
|
||||
|
||||
# Create application directory and attachment
|
||||
from app import get_attachments_path
|
||||
from app.models import get_attachments_path
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
Path(attachments_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -273,7 +273,7 @@ class TestFileDownload:
|
||||
session_id = 'test-download-multiple'
|
||||
|
||||
# Create application directory
|
||||
from app import get_attachments_path
|
||||
from app.models import get_attachments_path
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
Path(attachments_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -318,7 +318,7 @@ class TestFileDownload:
|
||||
session_id = 'test-download-special-chars'
|
||||
|
||||
# Create application directory
|
||||
from app import get_attachments_path
|
||||
from app.models import get_attachments_path
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
Path(attachments_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ Tests for configuration settings and helper functions.
|
||||
import pytest
|
||||
import os
|
||||
from config import Config
|
||||
from app import get_application_path, get_data_file_path, get_attachments_path
|
||||
from app.models import get_application_path, get_data_file_path, get_attachments_path
|
||||
|
||||
|
||||
class TestConfiguration:
|
||||
@@ -59,12 +59,14 @@ class TestPathHelpers:
|
||||
|
||||
def test_get_application_path(self, app):
|
||||
"""Test get_application_path returns correct path."""
|
||||
with app.app_context():
|
||||
session_id = 'test-123'
|
||||
expected_path = os.path.join(app.config['APPLICATIONS_FOLDER'], session_id)
|
||||
assert get_application_path(session_id) == expected_path
|
||||
|
||||
def test_get_data_file_path(self, app):
|
||||
"""Test get_data_file_path returns correct YAML file path."""
|
||||
with app.app_context():
|
||||
session_id = 'test-456'
|
||||
expected_path = os.path.join(
|
||||
app.config['APPLICATIONS_FOLDER'],
|
||||
@@ -75,6 +77,7 @@ class TestPathHelpers:
|
||||
|
||||
def test_get_attachments_path(self, app):
|
||||
"""Test get_attachments_path returns correct attachments directory path."""
|
||||
with app.app_context():
|
||||
session_id = 'test-789'
|
||||
expected_path = os.path.join(
|
||||
app.config['APPLICATIONS_FOLDER'],
|
||||
@@ -85,6 +88,7 @@ class TestPathHelpers:
|
||||
|
||||
def test_path_helpers_with_special_characters(self, app):
|
||||
"""Test path helpers handle session IDs with special characters."""
|
||||
with app.app_context():
|
||||
session_id = 'test-abc-123-def'
|
||||
app_path = get_application_path(session_id)
|
||||
assert session_id in app_path
|
||||
|
||||
+10
-6
@@ -4,7 +4,7 @@ Tests for email functionality.
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from flask_mail import Message
|
||||
from app import send_resume_email
|
||||
from app.email_service import send_resume_email
|
||||
|
||||
|
||||
class TestEmailSending:
|
||||
@@ -12,7 +12,8 @@ class TestEmailSending:
|
||||
|
||||
def test_send_resume_email_success(self, app):
|
||||
"""Test successfully sending a resume email."""
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with app.app_context():
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
result = send_resume_email('test@example.com', 'test-session-123', 'Software Developer')
|
||||
|
||||
assert result is True
|
||||
@@ -20,7 +21,8 @@ class TestEmailSending:
|
||||
|
||||
def test_send_resume_email_failure(self, app):
|
||||
"""Test handling email sending failure."""
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with app.app_context():
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
mock_send.side_effect = Exception('SMTP error')
|
||||
|
||||
result = send_resume_email('test@example.com', 'test-session-123', 'Test Job')
|
||||
@@ -29,7 +31,8 @@ class TestEmailSending:
|
||||
|
||||
def test_resume_email_content(self, app):
|
||||
"""Test that resume email contains correct content."""
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with app.app_context():
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
send_resume_email('applicant@example.com', 'abc-123', 'Marketing Manager')
|
||||
|
||||
# Get the Message object that was passed to send()
|
||||
@@ -44,7 +47,8 @@ class TestEmailSending:
|
||||
|
||||
def test_resume_link_in_email(self, app):
|
||||
"""Test that resume link is included in email body."""
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with app.app_context():
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
send_resume_email('test@example.com', 'session-xyz', 'Test Position')
|
||||
|
||||
if mock_send.called:
|
||||
@@ -60,7 +64,7 @@ class TestEmailIntegration:
|
||||
|
||||
def test_email_sent_on_initial_submission(self, client, app):
|
||||
"""Test that email is sent when user submits their email address."""
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
response = client.post('/apply/submit-email', data={
|
||||
'email': 'newuser@example.com',
|
||||
'job_name': 'Junior Developer'
|
||||
|
||||
@@ -3,16 +3,15 @@ Tests for HR email notifications functionality.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from app import send_hr_notification
|
||||
from app.email_service import send_hr_notification
|
||||
|
||||
|
||||
class TestHRNotifications:
|
||||
"""Test HR notification email functionality."""
|
||||
|
||||
def test_hr_notification_sent_on_submission(self, client, create_test_application, mock_mail):
|
||||
def test_hr_notification_sent_on_submission(self, app, client, create_test_application, mock_mail):
|
||||
"""Test that HR receives email when application is submitted."""
|
||||
# Set HR_EMAIL in config
|
||||
from app import app
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'http://localhost:5000'
|
||||
|
||||
@@ -59,11 +58,10 @@ class TestHRNotifications:
|
||||
assert f'/application/{session_id}/' in hr_email.body
|
||||
|
||||
def test_hr_notification_not_sent_when_hr_email_not_configured(
|
||||
self, client, create_test_application, mock_mail, caplog
|
||||
self, app, client, create_test_application, mock_mail, caplog
|
||||
):
|
||||
"""Test that HR notification is skipped when HR_EMAIL is not configured."""
|
||||
# Ensure HR_EMAIL is not set
|
||||
from app import app
|
||||
app.config['HR_EMAIL'] = None
|
||||
|
||||
# Create a test application
|
||||
@@ -97,9 +95,8 @@ class TestHRNotifications:
|
||||
# Verify warning was logged
|
||||
assert 'HR_EMAIL not configured' in caplog.text
|
||||
|
||||
def test_hr_notification_includes_file_count(self, client, create_test_application, mock_mail):
|
||||
def test_hr_notification_includes_file_count(self, app, client, create_test_application, mock_mail):
|
||||
"""Test that HR notification includes the count of uploaded files."""
|
||||
from app import app
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'http://localhost:5000'
|
||||
|
||||
@@ -146,9 +143,8 @@ class TestHRNotifications:
|
||||
hr_email = mock_mail.sent_messages[0]
|
||||
assert 'Anzahl der hochgeladenen Dokumente: 2' in hr_email.body
|
||||
|
||||
def test_hr_notification_with_zero_files(self, client, create_test_application, mock_mail):
|
||||
def test_hr_notification_with_zero_files(self, app, client, create_test_application, mock_mail):
|
||||
"""Test that HR notification works even when no files are uploaded."""
|
||||
from app import app
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'http://localhost:5000'
|
||||
|
||||
@@ -184,8 +180,6 @@ class TestHRNotifications:
|
||||
|
||||
def test_send_hr_notification_function_directly(self, app):
|
||||
"""Test the send_hr_notification function directly."""
|
||||
from app import send_hr_notification
|
||||
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'http://localhost:5000'
|
||||
|
||||
@@ -205,15 +199,13 @@ class TestHRNotifications:
|
||||
|
||||
with app.app_context():
|
||||
# Mock mail.send to prevent actual sending
|
||||
with patch('app.mail.send') as mock_send:
|
||||
with patch('app.email_service.mail.send') as mock_send:
|
||||
result = send_hr_notification(session_id, app_data)
|
||||
assert result is True
|
||||
assert mock_send.called
|
||||
|
||||
def test_send_hr_notification_handles_email_failure(self, app, caplog):
|
||||
"""Test that send_hr_notification handles email sending failures gracefully."""
|
||||
from app import send_hr_notification
|
||||
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'http://localhost:5000'
|
||||
|
||||
@@ -231,14 +223,13 @@ class TestHRNotifications:
|
||||
|
||||
with app.app_context():
|
||||
# Mock mail.send to raise an exception
|
||||
with patch('app.mail.send', side_effect=Exception('SMTP error')):
|
||||
with patch('app.email_service.mail.send', side_effect=Exception('SMTP error')):
|
||||
result = send_hr_notification(session_id, app_data)
|
||||
assert result is False
|
||||
assert 'Failed to send HR notification email' in caplog.text
|
||||
|
||||
def test_hr_notification_url_format(self, client, create_test_application, mock_mail):
|
||||
def test_hr_notification_url_format(self, app, client, create_test_application, mock_mail):
|
||||
"""Test that the application URL in HR notification is correctly formatted."""
|
||||
from app import app
|
||||
app.config['HR_EMAIL'] = 'hr@example.com'
|
||||
app.config['APPLICATION_URL_BASE'] = 'https://example.com'
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ Integration tests for the complete application workflow.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import patch
|
||||
from app import load_application_data
|
||||
from app.models import load_application_data
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@@ -13,7 +13,7 @@ class TestCompleteWorkflow:
|
||||
def test_complete_application_workflow(self, client, app, sample_pdf_file):
|
||||
"""Test the complete workflow from email to confirmation."""
|
||||
# Step 1: Submit email
|
||||
with patch('app.mail.send'):
|
||||
with patch('app.email_service.mail.send'):
|
||||
response = client.post('/apply/submit-email', data={
|
||||
'email': 'integration@test.com',
|
||||
'job_name': 'Integration Test Position'
|
||||
@@ -22,7 +22,7 @@ class TestCompleteWorkflow:
|
||||
assert response.status_code == 200
|
||||
|
||||
# For testing, create a known session
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
session_id = 'integration-test-123'
|
||||
save_application_data(session_id, {
|
||||
'session_id': session_id,
|
||||
@@ -130,7 +130,7 @@ class TestCompleteWorkflow:
|
||||
|
||||
def test_multiple_concurrent_applications(self, client, app):
|
||||
"""Test handling multiple concurrent applications with different sessions."""
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
|
||||
# Create multiple applications
|
||||
sessions = []
|
||||
@@ -182,7 +182,7 @@ class TestWorkflowEdgeCases:
|
||||
|
||||
def test_skip_optional_fields(self, client, app):
|
||||
"""Test completing workflow without filling optional fields."""
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
|
||||
session_id = 'optional-test'
|
||||
save_application_data(session_id, {
|
||||
@@ -226,7 +226,7 @@ class TestWorkflowEdgeCases:
|
||||
|
||||
def test_workflow_with_unicode_data(self, client, app):
|
||||
"""Test workflow with German special characters."""
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
|
||||
session_id = 'unicode-test'
|
||||
save_application_data(session_id, {
|
||||
@@ -265,7 +265,7 @@ class TestDataIntegrity:
|
||||
|
||||
def test_data_not_lost_between_pages(self, client, app):
|
||||
"""Test that data persists correctly when navigating between pages."""
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
|
||||
session_id = 'persistence-test'
|
||||
save_application_data(session_id, {
|
||||
|
||||
@@ -3,7 +3,7 @@ Tests for routes and form submissions in the Flask job application system.
|
||||
"""
|
||||
import pytest
|
||||
from flask import session
|
||||
from app import load_application_data
|
||||
from app.models import load_application_data
|
||||
|
||||
|
||||
class TestPage1Email:
|
||||
|
||||
+11
-2
@@ -5,7 +5,7 @@ import pytest
|
||||
import os
|
||||
import yaml
|
||||
from pathlib import Path
|
||||
from app import (
|
||||
from app.models import (
|
||||
save_application_data,
|
||||
load_application_data,
|
||||
get_application_path,
|
||||
@@ -19,6 +19,7 @@ class TestYAMLStorage:
|
||||
|
||||
def test_save_application_data(self, app, sample_application_data):
|
||||
"""Test saving application data to YAML file."""
|
||||
with app.app_context():
|
||||
session_id = sample_application_data['session_id']
|
||||
save_application_data(session_id, sample_application_data)
|
||||
|
||||
@@ -36,6 +37,7 @@ class TestYAMLStorage:
|
||||
|
||||
def test_load_application_data(self, app, sample_application_data):
|
||||
"""Test loading application data from YAML file."""
|
||||
with app.app_context():
|
||||
session_id = sample_application_data['session_id']
|
||||
|
||||
# Save data first
|
||||
@@ -50,11 +52,13 @@ class TestYAMLStorage:
|
||||
|
||||
def test_load_nonexistent_application(self, app):
|
||||
"""Test loading data for non-existent application."""
|
||||
with app.app_context():
|
||||
loaded_data = load_application_data('nonexistent-session')
|
||||
assert loaded_data is None
|
||||
|
||||
def test_unicode_characters_in_yaml(self, app):
|
||||
"""Test that German umlauts and special characters are preserved."""
|
||||
with app.app_context():
|
||||
session_id = 'test-unicode-123'
|
||||
data = {
|
||||
'session_id': session_id,
|
||||
@@ -79,6 +83,7 @@ class TestYAMLStorage:
|
||||
|
||||
def test_update_existing_application(self, app, sample_application_data):
|
||||
"""Test updating existing application data."""
|
||||
with app.app_context():
|
||||
session_id = sample_application_data['session_id']
|
||||
|
||||
# Save initial data
|
||||
@@ -104,6 +109,7 @@ class TestFolderStructure:
|
||||
|
||||
def test_application_folder_created(self, app, sample_application_data):
|
||||
"""Test that application folder is created when saving data."""
|
||||
with app.app_context():
|
||||
session_id = sample_application_data['session_id']
|
||||
save_application_data(session_id, sample_application_data)
|
||||
|
||||
@@ -113,6 +119,7 @@ class TestFolderStructure:
|
||||
|
||||
def test_data_yaml_created(self, app, sample_application_data):
|
||||
"""Test that data.yaml file is created."""
|
||||
with app.app_context():
|
||||
session_id = sample_application_data['session_id']
|
||||
save_application_data(session_id, sample_application_data)
|
||||
|
||||
@@ -123,6 +130,7 @@ class TestFolderStructure:
|
||||
|
||||
def test_attachments_folder_path(self, app):
|
||||
"""Test that attachments folder path is correct."""
|
||||
with app.app_context():
|
||||
session_id = 'test-attachments-123'
|
||||
attachments_path = get_attachments_path(session_id)
|
||||
|
||||
@@ -143,7 +151,7 @@ class TestDataPersistence:
|
||||
|
||||
# Extract session ID from response
|
||||
# For testing, we'll create a known session
|
||||
from app import save_application_data
|
||||
from app.models import save_application_data
|
||||
session_id = 'test-persist-123'
|
||||
save_application_data(session_id, {
|
||||
'session_id': session_id,
|
||||
@@ -189,6 +197,7 @@ class TestDataPersistence:
|
||||
|
||||
def test_empty_optional_fields_preserved(self, app):
|
||||
"""Test that empty optional fields are preserved correctly."""
|
||||
with app.app_context():
|
||||
session_id = 'test-empty-fields'
|
||||
data = {
|
||||
'session_id': session_id,
|
||||
|
||||
@@ -4,7 +4,7 @@ Tests for file upload functionality.
|
||||
import pytest
|
||||
import os
|
||||
from io import BytesIO
|
||||
from app import load_application_data, get_attachments_path
|
||||
from app.models import load_application_data, get_attachments_path
|
||||
|
||||
|
||||
class TestFileUpload:
|
||||
|
||||
+21
-11
@@ -2,7 +2,7 @@
|
||||
Tests for validation functions in the Flask job application system.
|
||||
"""
|
||||
import pytest
|
||||
from app import validate_email, validate_phone, validate_year, validate_zip, allowed_file
|
||||
from app.validators import validate_email, validate_phone, validate_year, validate_zip, allowed_file
|
||||
|
||||
|
||||
class TestEmailValidation:
|
||||
@@ -63,21 +63,24 @@ class TestPhoneValidation:
|
||||
class TestYearValidation:
|
||||
"""Tests for birth year validation."""
|
||||
|
||||
def test_valid_years(self):
|
||||
def test_valid_years(self, app):
|
||||
"""Test that valid birth years are accepted."""
|
||||
with app.app_context():
|
||||
assert validate_year('1940') is True # Minimum
|
||||
assert validate_year('1990') is True # Middle
|
||||
assert validate_year('2010') is True # Maximum
|
||||
|
||||
def test_invalid_years_out_of_range(self):
|
||||
def test_invalid_years_out_of_range(self, app):
|
||||
"""Test that years outside the valid range are rejected."""
|
||||
with app.app_context():
|
||||
assert validate_year('1939') is False # Too old
|
||||
assert validate_year('2011') is False # Too young
|
||||
assert validate_year('1900') is False
|
||||
assert validate_year('2025') is False
|
||||
|
||||
def test_invalid_year_format(self):
|
||||
def test_invalid_year_format(self, app):
|
||||
"""Test that invalid year formats are rejected."""
|
||||
with app.app_context():
|
||||
assert validate_year('') is False
|
||||
assert validate_year('90') is False # 2 digits
|
||||
assert validate_year('990') is False # 3 digits
|
||||
@@ -89,22 +92,25 @@ class TestYearValidation:
|
||||
class TestZipValidation:
|
||||
"""Tests for ZIP code validation."""
|
||||
|
||||
def test_valid_zip_codes(self):
|
||||
def test_valid_zip_codes(self, app):
|
||||
"""Test that valid ZIP codes are accepted."""
|
||||
with app.app_context():
|
||||
assert validate_zip('8001') is True
|
||||
assert validate_zip('12345') is True
|
||||
assert validate_zip('1') is True # Single digit
|
||||
assert validate_zip('1234567890') is True # Max 10 digits
|
||||
|
||||
def test_invalid_zip_codes(self):
|
||||
def test_invalid_zip_codes(self, app):
|
||||
"""Test that invalid ZIP codes are rejected."""
|
||||
with app.app_context():
|
||||
assert validate_zip('') is False
|
||||
assert validate_zip('12345678901') is False # 11 digits (too long)
|
||||
assert validate_zip('abc') is False # Non-numeric
|
||||
assert validate_zip('123a5') is False # Mixed
|
||||
|
||||
def test_zip_edge_cases(self):
|
||||
def test_zip_edge_cases(self, app):
|
||||
"""Test edge cases for ZIP validation."""
|
||||
with app.app_context():
|
||||
assert validate_zip('0') is True # Zero is valid
|
||||
assert validate_zip('00000') is True # Leading zeros
|
||||
|
||||
@@ -112,8 +118,9 @@ class TestZipValidation:
|
||||
class TestFileExtensionValidation:
|
||||
"""Tests for file extension validation."""
|
||||
|
||||
def test_valid_file_extensions(self):
|
||||
def test_valid_file_extensions(self, app):
|
||||
"""Test that files with valid extensions are accepted."""
|
||||
with app.app_context():
|
||||
assert allowed_file('document.pdf') is True
|
||||
assert allowed_file('resume.doc') is True
|
||||
assert allowed_file('cover.docx') is True
|
||||
@@ -122,8 +129,9 @@ class TestFileExtensionValidation:
|
||||
assert allowed_file('image.jpeg') is True
|
||||
assert allowed_file('picture.png') is True
|
||||
|
||||
def test_invalid_file_extensions(self):
|
||||
def test_invalid_file_extensions(self, app):
|
||||
"""Test that files with invalid extensions are rejected."""
|
||||
with app.app_context():
|
||||
assert allowed_file('script.exe') is False
|
||||
assert allowed_file('data.zip') is False
|
||||
assert allowed_file('code.py') is False
|
||||
@@ -134,15 +142,17 @@ class TestFileExtensionValidation:
|
||||
assert allowed_file('noextension') is False
|
||||
assert allowed_file('') is False
|
||||
|
||||
def test_case_insensitive_extensions(self):
|
||||
def test_case_insensitive_extensions(self, app):
|
||||
"""Test that file extension validation is case-insensitive."""
|
||||
with app.app_context():
|
||||
assert allowed_file('document.PDF') is True
|
||||
assert allowed_file('document.Pdf') is True
|
||||
assert allowed_file('image.JPG') is True
|
||||
assert allowed_file('image.JPEG') is True
|
||||
|
||||
def test_multiple_dots_in_filename(self):
|
||||
def test_multiple_dots_in_filename(self, app):
|
||||
"""Test files with multiple dots in the filename."""
|
||||
with app.app_context():
|
||||
assert allowed_file('my.document.pdf') is True
|
||||
assert allowed_file('file.name.with.dots.jpg') is True
|
||||
assert allowed_file('test.tar.gz') is False # .gz not allowed
|
||||
|
||||
Reference in New Issue
Block a user