From f1d8bae6a126ce2b1ecd1b3965a4e5c97d9eea8c Mon Sep 17 00:00:00 2001 From: Markus Graf Date: Sat, 27 Dec 2025 22:34:21 +0100 Subject: [PATCH] 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 --- app/__init__.py | 58 +++ app/email_service.py | 85 ++++ app/models.py | 41 ++ app/routes.py | 473 ++++++++++++++++++ app/utils.py | 30 ++ app/validators.py | 48 ++ .../005-refactor-app-modular-structure.md | 258 ++++++++++ run.py | 9 + tests/conftest.py | 11 +- tests/test_application_view.py | 8 +- tests/test_config.py | 48 +- tests/test_email.py | 64 +-- tests/test_hr_notifications.py | 25 +- tests/test_integration.py | 14 +- tests/test_routes.py | 2 +- tests/test_storage.py | 193 +++---- tests/test_uploads.py | 2 +- tests/test_validation.py | 114 +++-- 18 files changed, 1254 insertions(+), 229 deletions(-) create mode 100644 app/__init__.py create mode 100644 app/email_service.py create mode 100644 app/models.py create mode 100644 app/routes.py create mode 100644 app/utils.py create mode 100644 app/validators.py create mode 100644 prompts/completed/005-refactor-app-modular-structure.md create mode 100644 run.py diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..ee99cd3 --- /dev/null +++ b/app/__init__.py @@ -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 diff --git a/app/email_service.py b/app/email_service.py new file mode 100644 index 0000000..c698109 --- /dev/null +++ b/app/email_service.py @@ -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 diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..5991ef0 --- /dev/null +++ b/app/models.py @@ -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) diff --git a/app/routes.py b/app/routes.py new file mode 100644 index 0000000..3612ec2 --- /dev/null +++ b/app/routes.py @@ -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//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//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//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//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//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//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//remove-file/', 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//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//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//') + 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//download/') + 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/') + 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)) diff --git a/app/utils.py b/app/utils.py new file mode 100644 index 0000000..98aa2ac --- /dev/null +++ b/app/utils.py @@ -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 diff --git a/app/validators.py b/app/validators.py new file mode 100644 index 0000000..528828b --- /dev/null +++ b/app/validators.py @@ -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'] diff --git a/prompts/completed/005-refactor-app-modular-structure.md b/prompts/completed/005-refactor-app-modular-structure.md new file mode 100644 index 0000000..e10f459 --- /dev/null +++ b/prompts/completed/005-refactor-app-modular-structure.md @@ -0,0 +1,258 @@ + +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. + + + +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 + + + +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 + + + +**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) + + + +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 + + + +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. + + + +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) + + + +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// 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 + + + +- 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 + +Completed: Sa 27 Dez 2025 22:29:29 CET diff --git a/run.py b/run.py new file mode 100644 index 0000000..befa411 --- /dev/null +++ b/run.py @@ -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) diff --git a/tests/conftest.py b/tests/conftest.py index 3cd8ad7..df2274f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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, diff --git a/tests/test_application_view.py b/tests/test_application_view.py index f4559fe..40a9a27 100644 --- a/tests/test_application_view.py +++ b/tests/test_application_view.py @@ -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) diff --git a/tests/test_config.py b/tests/test_config.py index f3535e6..c5e5b61 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -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,33 +59,37 @@ class TestPathHelpers: def test_get_application_path(self, app): """Test get_application_path returns correct path.""" - session_id = 'test-123' - expected_path = os.path.join(app.config['APPLICATIONS_FOLDER'], session_id) - assert get_application_path(session_id) == expected_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.""" - session_id = 'test-456' - expected_path = os.path.join( - app.config['APPLICATIONS_FOLDER'], - session_id, - 'data.yaml' - ) - assert get_data_file_path(session_id) == expected_path + with app.app_context(): + session_id = 'test-456' + expected_path = os.path.join( + app.config['APPLICATIONS_FOLDER'], + session_id, + 'data.yaml' + ) + assert get_data_file_path(session_id) == expected_path def test_get_attachments_path(self, app): """Test get_attachments_path returns correct attachments directory path.""" - session_id = 'test-789' - expected_path = os.path.join( - app.config['APPLICATIONS_FOLDER'], - session_id, - 'attachments' - ) - assert get_attachments_path(session_id) == expected_path + with app.app_context(): + session_id = 'test-789' + expected_path = os.path.join( + app.config['APPLICATIONS_FOLDER'], + session_id, + 'attachments' + ) + assert get_attachments_path(session_id) == expected_path def test_path_helpers_with_special_characters(self, app): """Test path helpers handle session IDs with special characters.""" - session_id = 'test-abc-123-def' - app_path = get_application_path(session_id) - assert session_id in app_path - assert os.path.isabs(app_path) or app_path.startswith('.') + with app.app_context(): + session_id = 'test-abc-123-def' + app_path = get_application_path(session_id) + assert session_id in app_path + assert os.path.isabs(app_path) or app_path.startswith('.') diff --git a/tests/test_email.py b/tests/test_email.py index 8d3b981..21e69da 100644 --- a/tests/test_email.py +++ b/tests/test_email.py @@ -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,47 +12,51 @@ class TestEmailSending: def test_send_resume_email_success(self, app): """Test successfully sending a resume email.""" - with patch('app.mail.send') as mock_send: - result = send_resume_email('test@example.com', 'test-session-123', 'Software Developer') + 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 - assert mock_send.called + assert result is True + assert mock_send.called def test_send_resume_email_failure(self, app): """Test handling email sending failure.""" - with patch('app.mail.send') as mock_send: - mock_send.side_effect = Exception('SMTP error') + 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') + result = send_resume_email('test@example.com', 'test-session-123', 'Test Job') - assert result is False + assert result is False def test_resume_email_content(self, app): """Test that resume email contains correct content.""" - with patch('app.mail.send') as mock_send: - send_resume_email('applicant@example.com', 'abc-123', 'Marketing Manager') + 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() - assert mock_send.called - call_args = mock_send.call_args - message = call_args[0][0] if call_args[0] else None - - if message: - assert isinstance(message, Message) - assert 'applicant@example.com' in message.recipients - assert 'Bewerbung' in message.subject or 'Marketing Manager' in message.subject - - 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: - send_resume_email('test@example.com', 'session-xyz', 'Test Position') - - if mock_send.called: + # Get the Message object that was passed to send() + assert mock_send.called call_args = mock_send.call_args message = call_args[0][0] if call_args[0] else None - if message and hasattr(message, 'body'): - assert 'session-xyz' in message.body or '/resume/' in message.body + if message: + assert isinstance(message, Message) + assert 'applicant@example.com' in message.recipients + assert 'Bewerbung' in message.subject or 'Marketing Manager' in message.subject + + def test_resume_link_in_email(self, app): + """Test that resume link is included in email body.""" + 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: + call_args = mock_send.call_args + message = call_args[0][0] if call_args[0] else None + + if message and hasattr(message, 'body'): + assert 'session-xyz' in message.body or '/resume/' in message.body class TestEmailIntegration: @@ -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' diff --git a/tests/test_hr_notifications.py b/tests/test_hr_notifications.py index 54ce061..112b241 100644 --- a/tests/test_hr_notifications.py +++ b/tests/test_hr_notifications.py @@ -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' diff --git a/tests/test_integration.py b/tests/test_integration.py index 1bf34f6..7170613 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -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, { diff --git a/tests/test_routes.py b/tests/test_routes.py index 7b708bc..37f5573 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -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: diff --git a/tests/test_storage.py b/tests/test_storage.py index 9d6937c..c0e47fc 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -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,84 +19,89 @@ class TestYAMLStorage: def test_save_application_data(self, app, sample_application_data): """Test saving application data to YAML file.""" - session_id = sample_application_data['session_id'] - save_application_data(session_id, sample_application_data) + with app.app_context(): + session_id = sample_application_data['session_id'] + save_application_data(session_id, sample_application_data) - # Verify file was created - data_file = get_data_file_path(session_id) - assert os.path.exists(data_file) + # Verify file was created + data_file = get_data_file_path(session_id) + assert os.path.exists(data_file) - # Verify content - with open(data_file, 'r', encoding='utf-8') as f: - loaded_data = yaml.safe_load(f) + # Verify content + with open(data_file, 'r', encoding='utf-8') as f: + loaded_data = yaml.safe_load(f) - assert loaded_data['session_id'] == session_id - assert loaded_data['email'] == sample_application_data['email'] - assert loaded_data['job_name'] == sample_application_data['job_name'] + assert loaded_data['session_id'] == session_id + assert loaded_data['email'] == sample_application_data['email'] + assert loaded_data['job_name'] == sample_application_data['job_name'] def test_load_application_data(self, app, sample_application_data): """Test loading application data from YAML file.""" - session_id = sample_application_data['session_id'] + with app.app_context(): + session_id = sample_application_data['session_id'] - # Save data first - save_application_data(session_id, sample_application_data) + # Save data first + save_application_data(session_id, sample_application_data) - # Load and verify - loaded_data = load_application_data(session_id) - assert loaded_data is not None - assert loaded_data['session_id'] == session_id - assert loaded_data['email'] == sample_application_data['email'] - assert loaded_data['personal_info']['name'] == 'Müller' + # Load and verify + loaded_data = load_application_data(session_id) + assert loaded_data is not None + assert loaded_data['session_id'] == session_id + assert loaded_data['email'] == sample_application_data['email'] + assert loaded_data['personal_info']['name'] == 'Müller' def test_load_nonexistent_application(self, app): """Test loading data for non-existent application.""" - loaded_data = load_application_data('nonexistent-session') - assert loaded_data is None + 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.""" - session_id = 'test-unicode-123' - data = { - 'session_id': session_id, - 'email': 'test@example.com', - 'job_name': 'Test Job', - 'current_page': 2, - 'personal_info': { - 'name': 'Müller', - 'firstname': 'Jürgen', - 'city': 'München', - 'notes': 'Grüße aus Zürich' + with app.app_context(): + session_id = 'test-unicode-123' + data = { + 'session_id': session_id, + 'email': 'test@example.com', + 'job_name': 'Test Job', + 'current_page': 2, + 'personal_info': { + 'name': 'Müller', + 'firstname': 'Jürgen', + 'city': 'München', + 'notes': 'Grüße aus Zürich' + } } - } - save_application_data(session_id, data) - loaded_data = load_application_data(session_id) + save_application_data(session_id, data) + loaded_data = load_application_data(session_id) - assert loaded_data['personal_info']['name'] == 'Müller' - assert loaded_data['personal_info']['firstname'] == 'Jürgen' - assert loaded_data['personal_info']['city'] == 'München' - assert loaded_data['personal_info']['notes'] == 'Grüße aus Zürich' + assert loaded_data['personal_info']['name'] == 'Müller' + assert loaded_data['personal_info']['firstname'] == 'Jürgen' + assert loaded_data['personal_info']['city'] == 'München' + assert loaded_data['personal_info']['notes'] == 'Grüße aus Zürich' def test_update_existing_application(self, app, sample_application_data): """Test updating existing application data.""" - session_id = sample_application_data['session_id'] + with app.app_context(): + session_id = sample_application_data['session_id'] - # Save initial data - save_application_data(session_id, sample_application_data) + # Save initial data + save_application_data(session_id, sample_application_data) - # Update data - sample_application_data['current_page'] = 3 - sample_application_data['motivation_answers'] = { - 'current_job': 'Developer', - 'motivation': 'Great company' - } - save_application_data(session_id, sample_application_data) + # Update data + sample_application_data['current_page'] = 3 + sample_application_data['motivation_answers'] = { + 'current_job': 'Developer', + 'motivation': 'Great company' + } + save_application_data(session_id, sample_application_data) - # Verify update - loaded_data = load_application_data(session_id) - assert loaded_data['current_page'] == 3 - assert 'motivation_answers' in loaded_data - assert loaded_data['motivation_answers']['current_job'] == 'Developer' + # Verify update + loaded_data = load_application_data(session_id) + assert loaded_data['current_page'] == 3 + assert 'motivation_answers' in loaded_data + assert loaded_data['motivation_answers']['current_job'] == 'Developer' class TestFolderStructure: @@ -104,30 +109,33 @@ class TestFolderStructure: def test_application_folder_created(self, app, sample_application_data): """Test that application folder is created when saving data.""" - session_id = sample_application_data['session_id'] - save_application_data(session_id, sample_application_data) + with app.app_context(): + session_id = sample_application_data['session_id'] + save_application_data(session_id, sample_application_data) - app_path = get_application_path(session_id) - assert os.path.exists(app_path) - assert os.path.isdir(app_path) + app_path = get_application_path(session_id) + assert os.path.exists(app_path) + assert os.path.isdir(app_path) def test_data_yaml_created(self, app, sample_application_data): """Test that data.yaml file is created.""" - session_id = sample_application_data['session_id'] - save_application_data(session_id, sample_application_data) + with app.app_context(): + session_id = sample_application_data['session_id'] + save_application_data(session_id, sample_application_data) - data_file = get_data_file_path(session_id) - assert os.path.exists(data_file) - assert os.path.isfile(data_file) - assert data_file.endswith('data.yaml') + data_file = get_data_file_path(session_id) + assert os.path.exists(data_file) + assert os.path.isfile(data_file) + assert data_file.endswith('data.yaml') def test_attachments_folder_path(self, app): """Test that attachments folder path is correct.""" - session_id = 'test-attachments-123' - attachments_path = get_attachments_path(session_id) + with app.app_context(): + session_id = 'test-attachments-123' + attachments_path = get_attachments_path(session_id) - assert 'attachments' in attachments_path - assert session_id in attachments_path + assert 'attachments' in attachments_path + assert session_id in attachments_path class TestDataPersistence: @@ -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,26 +197,27 @@ class TestDataPersistence: def test_empty_optional_fields_preserved(self, app): """Test that empty optional fields are preserved correctly.""" - session_id = 'test-empty-fields' - data = { - 'session_id': session_id, - 'email': 'test@example.com', - 'job_name': 'Test', - 'current_page': 3, - 'personal_info': { - 'civil_status': '' # Optional, empty - }, - 'motivation_answers': { - 'current_job': '', # Optional, empty - 'motivation': 'Some text', - 'qualifications': '', - 'salary': '' + with app.app_context(): + session_id = 'test-empty-fields' + data = { + 'session_id': session_id, + 'email': 'test@example.com', + 'job_name': 'Test', + 'current_page': 3, + 'personal_info': { + 'civil_status': '' # Optional, empty + }, + 'motivation_answers': { + 'current_job': '', # Optional, empty + 'motivation': 'Some text', + 'qualifications': '', + 'salary': '' + } } - } - save_application_data(session_id, data) - loaded_data = load_application_data(session_id) + save_application_data(session_id, data) + loaded_data = load_application_data(session_id) - assert loaded_data['personal_info']['civil_status'] == '' - assert loaded_data['motivation_answers']['current_job'] == '' - assert loaded_data['motivation_answers']['motivation'] == 'Some text' + assert loaded_data['personal_info']['civil_status'] == '' + assert loaded_data['motivation_answers']['current_job'] == '' + assert loaded_data['motivation_answers']['motivation'] == 'Some text' diff --git a/tests/test_uploads.py b/tests/test_uploads.py index 536f03d..0c9d04f 100644 --- a/tests/test_uploads.py +++ b/tests/test_uploads.py @@ -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: diff --git a/tests/test_validation.py b/tests/test_validation.py index a6968cf..8e2aaf4 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -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,86 +63,96 @@ 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.""" - assert validate_year('1940') is True # Minimum - assert validate_year('1990') is True # Middle - assert validate_year('2010') is True # Maximum + 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.""" - 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 + 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.""" - assert validate_year('') is False - assert validate_year('90') is False # 2 digits - assert validate_year('990') is False # 3 digits - assert validate_year('19900') is False # 5 digits - assert validate_year('abcd') is False # Non-numeric - assert validate_year('199a') is False # Mixed + 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 + assert validate_year('19900') is False # 5 digits + assert validate_year('abcd') is False # Non-numeric + assert validate_year('199a') is False # Mixed 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.""" - 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 + 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.""" - 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 + 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.""" - assert validate_zip('0') is True # Zero is valid - assert validate_zip('00000') is True # Leading zeros + with app.app_context(): + assert validate_zip('0') is True # Zero is valid + assert validate_zip('00000') is True # Leading zeros 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.""" - assert allowed_file('document.pdf') is True - assert allowed_file('resume.doc') is True - assert allowed_file('cover.docx') is True - assert allowed_file('notes.txt') is True - assert allowed_file('photo.jpg') is True - assert allowed_file('image.jpeg') is True - assert allowed_file('picture.png') is True + 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 + assert allowed_file('notes.txt') is True + assert allowed_file('photo.jpg') is True + 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.""" - assert allowed_file('script.exe') is False - assert allowed_file('data.zip') is False - assert allowed_file('code.py') is False - assert allowed_file('file.unknown') is False + 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 + assert allowed_file('file.unknown') is False def test_file_without_extension(self): """Test that files without extensions are rejected.""" 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.""" - 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 + 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.""" - 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 + 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