v1
Some checks failed
CI / test (3.11) (push) Failing after 5m36s
CI / build-image (push) Has been skipped

This commit is contained in:
2025-10-22 16:48:55 +02:00
commit 4cefd4e3ab
53 changed files with 5837 additions and 0 deletions

30
tests/conftest.py Normal file
View File

@@ -0,0 +1,30 @@
import os
import tempfile
import pytest
import importlib
import sys
from pathlib import Path
# Ensure the repository root is on sys.path so tests can import the server package.
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture(autouse=True, scope="function")
def setup_tmp_db(tmp_path, monkeypatch):
"""Set up the database for each test function."""
tmp_db = tmp_path / "forms.db"
# Patch the module attribute directly to avoid package name collisions
monkeypatch.setattr(server_app_module, "DB_PATH", tmp_db, raising=False)
monkeypatch.setattr("server.settings.ADMIN_USERNAME", "admin")
monkeypatch.setattr("server.settings.ADMIN_PASSWORD", "admin")
init_db()
yield

97
tests/test_admin.py Normal file
View File

@@ -0,0 +1,97 @@
"""Tests for admin routes."""
import pytest
from server.app import app
@pytest.fixture
def client():
with app.test_client() as client:
yield client
@pytest.fixture(autouse=True)
def setup_admin_creds(monkeypatch):
monkeypatch.setattr("server.settings.ADMIN_USERNAME", "admin")
monkeypatch.setattr("server.settings.ADMIN_PASSWORD", "admin")
def test_admin_settings_requires_login(client):
"""Test admin settings page requires login."""
resp = client.get("/admin/settings")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_settings_with_login(client):
"""Test admin settings page displays when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Access settings
resp = client.get("/admin/settings")
assert resp.status_code == 200
assert b"Application Settings" in resp.data
assert b"Database" in resp.data
assert b"SMTP" in resp.data
assert b"Logout" in resp.data
def test_admin_dashboard_requires_login(client):
"""Test admin dashboard requires login."""
resp = client.get("/admin/")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_dashboard_with_login(client):
"""Test admin dashboard displays when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Access dashboard
resp = client.get("/admin/")
assert resp.status_code == 200
assert b"Admin Dashboard" in resp.data
assert b"Newsletter Subscribers" in resp.data
assert b"Logout" in resp.data
def test_admin_newsletter_subscribers_requires_login(client):
"""Test newsletter subscribers page requires login."""
resp = client.get("/admin/newsletter")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_newsletter_subscribers_with_login(client):
"""Test newsletter subscribers page displays when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Access newsletter subscribers
resp = client.get("/admin/newsletter")
assert resp.status_code == 200
assert b"Newsletter Subscribers" in resp.data
assert b"Logout" in resp.data
def test_admin_newsletter_create_requires_login(client):
"""Test newsletter create page requires login."""
resp = client.get("/admin/newsletter/create")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_newsletter_create_with_login(client):
"""Test newsletter create page displays when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Access newsletter create
resp = client.get("/admin/newsletter/create")
assert resp.status_code == 200
assert b"Create Newsletter" in resp.data
assert b"Subject Line" in resp.data
assert b"Content" in resp.data
assert b"Logout" in resp.data

View File

@@ -0,0 +1,174 @@
import sqlite3
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_get_contact_submissions_requires_auth(client):
"""Test that getting contact submissions requires authentication."""
resp = client.get("/api/contact")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_get_contact_submissions_with_auth(client):
"""Test getting contact submissions when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create some test submissions
client.post("/api/contact", data={"name": "Test User 1",
"email": "test1@example.com", "message": "Message 1", "consent": "on"})
client.post("/api/contact", data={"name": "Test User 2",
"email": "test2@example.com", "message": "Message 2", "consent": "on"})
resp = client.get("/api/contact")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "submissions" in data
assert len(data["submissions"]) == 2
# Check pagination info
assert "pagination" in data
assert data["pagination"]["total"] == 2
assert data["pagination"]["page"] == 1
assert data["pagination"]["per_page"] == 50
def test_admin_get_contact_submissions_requires_auth(client):
"""Test that getting contact submissions via admin API requires authentication."""
resp = client.get("/admin/api/contact")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_get_contact_submissions_with_auth(client):
"""Test getting contact submissions via admin API when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create some test submissions
client.post("/api/contact", data={"name": "Test User 1",
"email": "test1@example.com", "message": "Message 1", "consent": "on"})
client.post("/api/contact", data={"name": "Test User 2",
"email": "test2@example.com", "message": "Message 2", "consent": "on"})
resp = client.get("/admin/api/contact")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "submissions" in data
assert len(data["submissions"]) == 2
# Check pagination info
assert "pagination" in data
assert data["pagination"]["total"] == 2
assert data["pagination"]["page"] == 1
assert data["pagination"]["per_page"] == 50
def test_delete_contact_submission_requires_auth(client):
"""Test that deleting contact submissions requires authentication."""
resp = client.delete("/api/contact/1")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_delete_contact_submission_with_auth(client):
"""Test deleting contact submissions when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create a test submission
resp = client.post("/api/contact", data={"name": "Test User",
"email": "test@example.com", "message": "Message", "consent": "on"})
submission_id = resp.get_json()["id"]
# Delete the submission
resp = client.delete(f"/api/contact/{submission_id}")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "deleted successfully" in data["message"]
# Verify it's gone
resp = client.get("/api/contact")
data = resp.get_json()
assert len(data["submissions"]) == 0
def test_admin_submissions_page_requires_auth(client):
"""Test that admin submissions page requires authentication."""
resp = client.get("/admin/submissions")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_submissions_page_with_auth(client):
"""Test admin submissions page loads when authenticated."""
# Login and access submissions page
client.post("/auth/login", data={"username": "admin", "password": "admin"})
resp = client.get("/admin/submissions")
assert resp.status_code == 200
assert b"Contact Form Submissions" in resp.data
assert b"Loading submissions" in resp.data
def test_admin_delete_contact_submission_requires_auth(client):
"""Test that deleting contact submissions via admin API requires authentication."""
resp = client.delete("/admin/api/contact/1")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_admin_delete_contact_submission_with_auth(client):
"""Test deleting contact submissions via admin API when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create a test submission
client.post("/api/contact", data={"name": "Test User",
"email": "test@example.com", "message": "Message", "consent": "on"})
# Get the submission to find its ID
resp = client.get("/admin/api/contact")
data = resp.get_json()
submission_id = data["submissions"][0]["id"]
# Delete the submission
resp = client.delete(f"/admin/api/contact/{submission_id}")
assert resp.status_code == 200
delete_data = resp.get_json()
assert delete_data["status"] == "ok"
# Verify it's deleted
resp = client.get("/admin/api/contact")
data = resp.get_json()
assert len(data["submissions"]) == 0
def test_admin_delete_nonexistent_contact_submission(client):
"""Test deleting a non-existent contact submission."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Try to delete a non-existent submission
resp = client.delete("/admin/api/contact/999")
assert resp.status_code == 404
data = resp.get_json()
assert data["status"] == "error"
assert "not found" in data["message"]

View File

@@ -0,0 +1,115 @@
"""Tests for admin newsletter API endpoints."""
import pytest
from server.app import app
@pytest.fixture
def client():
with app.test_client() as client:
yield client
@pytest.fixture(autouse=True)
def setup_admin_creds(monkeypatch):
monkeypatch.setattr("server.settings.ADMIN_USERNAME", "admin")
monkeypatch.setattr("server.settings.ADMIN_PASSWORD", "admin")
def test_create_newsletter_requires_login(client):
"""Test creating newsletter requires login."""
resp = client.post("/admin/api/newsletters", json={
"subject": "Test Subject",
"content": "Test content"
})
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_create_newsletter_with_login(client):
"""Test creating newsletter when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create newsletter
resp = client.post("/admin/api/newsletters", json={
"subject": "Test Subject",
"content": "Test content",
"sender_name": "Test Sender"
})
assert resp.status_code == 201
data = resp.get_json()
assert data["status"] == "ok"
assert "newsletter_id" in data
def test_create_newsletter_missing_fields(client):
"""Test creating newsletter with missing required fields."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Try without subject
resp = client.post("/admin/api/newsletters", json={
"content": "Test content"
})
assert resp.status_code == 400
data = resp.get_json()
assert data["status"] == "error"
assert "required" in data["message"]
# Try without content
resp = client.post("/admin/api/newsletters", json={
"subject": "Test Subject"
})
assert resp.status_code == 400
data = resp.get_json()
assert data["status"] == "error"
assert "required" in data["message"]
def test_get_newsletters_requires_login(client):
"""Test getting newsletters requires login."""
resp = client.get("/admin/api/newsletters")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_get_newsletters_with_login(client):
"""Test getting newsletters when logged in."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Create a newsletter first
client.post("/admin/api/newsletters", json={
"subject": "Test Subject",
"content": "Test content"
})
# Get newsletters
resp = client.get("/admin/api/newsletters")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "newsletters" in data
assert "pagination" in data
assert len(data["newsletters"]) >= 1
def test_send_newsletter_requires_login(client):
"""Test sending newsletter requires login."""
resp = client.post("/admin/api/newsletters/1/send")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_send_newsletter_not_found(client):
"""Test sending non-existent newsletter."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Try to send non-existent newsletter
resp = client.post("/admin/api/newsletters/999/send")
assert resp.status_code == 404
data = resp.get_json()
assert data["status"] == "error"
assert "not found" in data["message"]

View File

@@ -0,0 +1,79 @@
import sqlite3
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_get_app_settings_api_requires_auth(client):
"""Test that getting app settings requires authentication."""
resp = client.get("/admin/api/settings")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_get_app_settings_api_with_auth(client):
"""Test getting app settings via API when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
resp = client.get("/admin/api/settings")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "settings" in data
assert isinstance(data["settings"], dict)
def test_update_app_setting_api_requires_auth(client):
"""Test that updating app settings requires authentication."""
resp = client.put("/admin/api/settings/test_key",
json={"value": "test_value"})
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
def test_update_app_setting_api_with_auth(client):
"""Test updating app settings via API when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Update a setting
resp = client.put("/admin/api/settings/test_key",
json={"value": "test_value"})
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "updated successfully" in data["message"]
# Verify it was saved
resp = client.get("/admin/api/settings")
data = resp.get_json()
assert data["settings"]["test_key"] == "test_value"
def test_delete_app_setting_api_with_auth(client):
"""Test deleting app settings via API when authenticated."""
# Login first
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Add a setting first
client.put("/admin/api/settings/delete_test", json={"value": "to_delete"})
# Delete the setting
resp = client.delete("/admin/api/settings/delete_test")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "ok"
assert "deleted successfully" in data["message"]

27
tests/test_api.py Normal file
View File

@@ -0,0 +1,27 @@
import sqlite3
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture(autouse=True)
def setup_tmp_db(tmp_path, monkeypatch):
tmp_db = tmp_path / "forms.db"
# Patch the module attribute directly to avoid package name collisions
monkeypatch.setattr(server_app_module, "DB_PATH", tmp_db, raising=False)
monkeypatch.setattr("server.settings.ADMIN_USERNAME", "admin")
monkeypatch.setattr("server.settings.ADMIN_PASSWORD", "admin")
init_db()
yield
@pytest.fixture
def client():
with app.test_client() as client:
yield client

69
tests/test_auth.py Normal file
View File

@@ -0,0 +1,69 @@
"""Tests for authentication functionality."""
import pytest
from server.app import app
@pytest.fixture
def client():
with app.test_client() as client:
yield client
@pytest.fixture(autouse=True)
def setup_admin_creds(monkeypatch):
monkeypatch.setattr("server.settings.ADMIN_USERNAME", "admin")
monkeypatch.setattr("server.settings.ADMIN_PASSWORD", "admin")
def test_login_page_get(client):
"""Test login page renders."""
resp = client.get("/auth/login")
assert resp.status_code == 200
assert b"Admin Login" in resp.data
def test_login_success(client):
"""Test successful login."""
resp = client.post(
"/auth/login", data={"username": "admin", "password": "admin"})
assert resp.status_code == 302 # Redirect to admin dashboard
assert resp.headers["Location"] == "/admin/"
# Check session
with client.session_transaction() as sess:
assert sess["logged_in"] is True
def test_login_failure(client):
"""Test failed login."""
resp = client.post(
"/auth/login", data={"username": "wrong", "password": "wrong"})
assert resp.status_code == 200
assert b"Invalid credentials" in resp.data
# Check session not set
with client.session_transaction() as sess:
assert "logged_in" not in sess
def test_logout(client):
"""Test logout."""
# First login
client.post("/auth/login", data={"username": "admin", "password": "admin"})
# Then logout
resp = client.get("/auth/logout")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"
# Check session cleared
with client.session_transaction() as sess:
assert "logged_in" not in sess
def test_protected_route_without_login(client):
"""Test accessing protected route without login redirects to login."""
resp = client.get("/admin/settings")
assert resp.status_code == 302
assert resp.headers["Location"] == "/auth/login"

39
tests/test_contact_api.py Normal file
View File

@@ -0,0 +1,39 @@
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def post(client, data):
return client.post("/api/contact", data=data)
def test_valid_submission_creates_record_and_returns_201(client):
resp = post(
client,
{"name": "Test User", "email": "test@example.com",
"message": "Hello", "consent": "on"},
)
assert resp.status_code in (201, 202)
body = resp.get_json()
assert body["status"] == "ok"
assert isinstance(body.get("id"), int)
def test_missing_required_fields_returns_400(client):
resp = post(client, {"name": "", "email": "", "message": ""})
assert resp.status_code == 400
body = resp.get_json()
assert body["status"] == "error"
assert "errors" in body

View File

@@ -0,0 +1,94 @@
from __future__ import annotations
from email.message import EmailMessage
from typing import Any, cast
import pytest
from server.services import contact as contact_service # noqa: E402 pylint: disable=wrong-import-position
@pytest.fixture
def patched_settings(monkeypatch):
original = contact_service.settings.SMTP_SETTINGS.copy()
patched = original.copy()
monkeypatch.setattr(contact_service.settings, "SMTP_SETTINGS", patched)
return patched
def test_send_notification_returns_false_when_unconfigured(monkeypatch, patched_settings):
patched_settings.update({"host": None, "recipients": []})
# Ensure we do not accidentally open a socket if called
monkeypatch.setattr(contact_service.smtplib, "SMTP", pytest.fail)
submission = contact_service.ContactSubmission(
name="Test",
email="test@example.com",
company=None,
message="Hello",
timeline=None,
)
assert contact_service.send_notification(submission) is False
def test_send_notification_sends_email(monkeypatch, patched_settings):
patched_settings.update(
{
"host": "smtp.example.com",
"port": 2525,
"sender": "sender@example.com",
"username": "user",
"password": "secret",
"use_tls": True,
"recipients": ["owner@example.com"],
}
)
smtp_calls: dict[str, Any] = {}
class DummySMTP:
def __init__(self, host, port, timeout=None):
smtp_calls["init"] = (host, port, timeout)
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def starttls(self):
smtp_calls["starttls"] = True
def login(self, username, password):
smtp_calls["login"] = (username, password)
def send_message(self, message):
smtp_calls["message"] = message
monkeypatch.setattr(contact_service.smtplib, "SMTP", DummySMTP)
submission = contact_service.ContactSubmission(
name="Alice",
email="alice@example.com",
company="Example Co",
message="Hello there",
timeline="Soon",
)
assert contact_service.send_notification(submission) is True
assert smtp_calls["init"] == (
patched_settings["host"],
patched_settings["port"],
15,
)
assert smtp_calls["starttls"] is True
assert smtp_calls["login"] == (
patched_settings["username"], patched_settings["password"])
message = cast(EmailMessage, smtp_calls["message"])
assert message["Subject"] == "Neue Kontaktanfrage von Alice"
assert message["To"] == "owner@example.com"
assert "Hello there" in message.get_content()

90
tests/test_database.py Normal file
View File

@@ -0,0 +1,90 @@
import sqlite3
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_is_postgres_enabled():
"""Test postgres detection logic."""
from server.database import is_postgres_enabled, set_postgres_override
# Test override functionality
set_postgres_override(True)
assert is_postgres_enabled()
set_postgres_override(False)
assert not is_postgres_enabled()
set_postgres_override(None) # Reset to default
def test_db_cursor_context_manager():
"""Test database cursor context manager."""
from server.database import db_cursor
with db_cursor() as (conn, cur):
assert conn is not None
assert cur is not None
# Test that we can execute a query
cur.execute("SELECT 1")
result = cur.fetchone()
assert result[0] == 1
def test_get_app_settings_empty():
"""Test getting app settings when none exist."""
from server.database import get_app_settings
settings = get_app_settings()
assert isinstance(settings, dict)
assert len(settings) == 0
def test_update_app_setting():
"""Test updating app settings."""
from server.database import update_app_setting, get_app_settings
# Update a setting
update_app_setting("test_key", "test_value")
# Verify it was saved
settings = get_app_settings()
assert settings["test_key"] == "test_value"
def test_delete_app_setting():
"""Test deleting app settings."""
from server.database import update_app_setting, delete_app_setting, get_app_settings
# Add a setting
update_app_setting("delete_test", "to_delete")
# Delete it
delete_app_setting("delete_test")
# Verify it's gone
settings = get_app_settings()
assert "delete_test" not in settings
def test_get_contacts_pagination():
"""Test contact pagination."""
from server.database import get_contacts
# Get first page
submissions, total = get_contacts(page=1, per_page=10)
assert isinstance(submissions, list)
assert isinstance(total, int)
assert total >= 0

View File

@@ -0,0 +1,79 @@
"""SMTP integration tests relying on real infrastructure."""
from __future__ import annotations
import os
import smtplib
from email.message import EmailMessage
import pytest
from server.services import contact as contact_service
RUN_INTEGRATION = os.getenv("RUN_SMTP_INTEGRATION_TEST")
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(
not RUN_INTEGRATION,
reason="Set RUN_SMTP_INTEGRATION_TEST=1 to enable SMTP integration tests.",
),
]
def _require_smtp_settings():
settings = contact_service.settings.SMTP_SETTINGS
if not settings["host"] or not settings["recipients"] or not settings["username"]:
pytest.skip("SMTP settings not fully configured via environment")
return settings
def _build_submission() -> contact_service.ContactSubmission:
settings = contact_service.settings.SMTP_SETTINGS
return contact_service.ContactSubmission(
name="Integration Test",
email=settings["sender"] or settings["username"] or "integration@example.com",
company="Integration",
message="Integration test notification",
timeline=None,
)
'''
Test sending a notification via SMTP using real settings.
This requires a properly configured SMTP server and valid credentials.
Commenting out to avoid accidental execution during local runs.
@pytest.mark.skip(reason="Requires real SMTP server configuration")
'''
'''
def test_send_notification_real_smtp():
settings = _require_smtp_settings()
submission = _build_submission()
assert contact_service.send_notification(submission) is True
def test_direct_smtp_connection():
settings = _require_smtp_settings()
use_ssl = settings["port"] == 465
client_cls = smtplib.SMTP_SSL if use_ssl else smtplib.SMTP
with client_cls(settings["host"], settings["port"], timeout=10) as client:
client.ehlo()
if settings["use_tls"] and not use_ssl:
client.starttls()
client.ehlo()
client.login(settings["username"], settings["password"] or "")
message = EmailMessage()
message["Subject"] = "SMTP integration check"
message["From"] = settings["sender"] or settings["username"]
message["To"] = settings["recipients"][0]
message.set_content(
"This is a test email for SMTP integration checks.")
client.send_message(message)
'''

56
tests/test_metrics.py Normal file
View File

@@ -0,0 +1,56 @@
import importlib
import time
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture(autouse=True)
def setup_tmp_db(tmp_path, monkeypatch):
tmp_db = tmp_path / "forms.db"
monkeypatch.setattr(server_app_module, "DB_PATH", tmp_db, raising=False)
init_db()
yield
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_metrics_endpoint_reports_uptime_and_total(client):
# Ensure a simple GET to /metrics succeeds and returns recent uptime
resp = client.get("/metrics")
assert resp.status_code == 200
# If prometheus_client isn't installed, metrics returns JSON
if resp.content_type.startswith("application/json"):
body = resp.get_json()
assert "uptime_seconds" in body
assert "total_submissions" in body
else:
# If prometheus_client is present, the response is the Prometheus text format
text = resp.get_data(as_text=True)
assert "# HELP" in text or "http_request_duration_seconds" in text
def test_request_metrics_increment_on_request(client):
# Make sure we have a baseline
before = client.get("/metrics").get_data(as_text=True)
# Trigger a contact submission attempt (invalid payload will still count the request)
client.post("/api/contact", data={})
# Wait a tiny bit for histogram observation
time.sleep(0.01)
after = client.get("/metrics").get_data(as_text=True)
# If prometheus_client isn't present, the JSON will include total_submissions
if client.get("/metrics").content_type.startswith("application/json"):
before_json = client.get("/metrics").get_json()
after_json = client.get("/metrics").get_json()
assert after_json["uptime_seconds"] >= before_json["uptime_seconds"]
else:
# Ensure some metrics text exists and that it changed (best-effort)
assert after != before

View File

@@ -0,0 +1,118 @@
import sqlite3
import importlib
import pytest
server_app_module = importlib.import_module("server.app")
# Expose app and init_db from the imported module
app = server_app_module.app
init_db = server_app_module.init_db
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_newsletter_subscription_creates_record(client):
resp = client.post("/api/newsletter", json={"email": "test@example.com"})
assert resp.status_code == 201
body = resp.get_json()
assert body["status"] == "ok"
# Note: The API doesn't return an ID in the response
def test_newsletter_duplicate_subscription_returns_conflict(client):
# First subscription
client.post("/api/newsletter", json={"email": "test@example.com"})
# Duplicate subscription
resp = client.post("/api/newsletter", json={"email": "test@example.com"})
assert resp.status_code == 409
body = resp.get_json()
assert body["status"] == "error"
assert "already subscribed" in body["message"].lower()
def test_newsletter_unsubscribe(client):
# Subscribe first
client.post("/api/newsletter", json={"email": "test@example.com"})
# Unsubscribe
resp = client.delete("/api/newsletter", json={"email": "test@example.com"})
assert resp.status_code == 200
body = resp.get_json()
assert body["status"] == "ok"
assert "unsubscribed" in body["message"].lower()
def test_newsletter_unsubscribe_not_subscribed(client):
resp = client.delete("/api/newsletter", json={"email": "test@example.com"})
assert resp.status_code == 404
body = resp.get_json()
assert body["status"] == "error"
assert "not subscribed" in body["message"].lower()
def test_newsletter_update_email(client):
# Subscribe first
client.post("/api/newsletter", json={"email": "old@example.com"})
# Update email
resp = client.put(
"/api/newsletter", json={"old_email": "old@example.com", "new_email": "new@example.com"})
assert resp.status_code == 200
body = resp.get_json()
assert body["status"] == "ok"
assert "updated" in body["message"].lower()
def test_newsletter_update_email_not_found(client):
resp = client.put(
"/api/newsletter", json={"old_email": "nonexistent@example.com", "new_email": "new@example.com"})
assert resp.status_code == 404
body = resp.get_json()
assert body["status"] == "error"
assert "not found" in body["message"].lower()
def test_newsletter_manage_page_get(client):
resp = client.get("/api/newsletter/manage")
assert resp.status_code == 200
assert b"Newsletter Subscription Management" in resp.data
def test_newsletter_manage_subscribe(client):
resp = client.post("/api/newsletter/manage",
data={"email": "manage@example.com", "action": "subscribe"})
assert resp.status_code == 200
assert b"Successfully subscribed" in resp.data
def test_newsletter_manage_unsubscribe(client):
# Subscribe first
client.post("/api/newsletter/manage",
data={"email": "manage@example.com", "action": "subscribe"})
# Unsubscribe
resp = client.post("/api/newsletter/manage",
data={"email": "manage@example.com", "action": "unsubscribe"})
assert resp.status_code == 200
assert b"Successfully unsubscribed" in resp.data
def test_newsletter_manage_update(client):
# Subscribe first
client.post("/api/newsletter/manage",
data={"email": "old@example.com", "action": "subscribe"})
# Update
resp = client.post("/api/newsletter/manage", data={
"old_email": "old@example.com", "new_email": "updated@example.com", "action": "update"})
assert resp.status_code == 200
# Check that some success message is displayed
assert b"success" in resp.data.lower() or b"updated" in resp.data.lower()
def test_newsletter_manage_invalid_email(client):
resp = client.post("/api/newsletter/manage",
data={"email": "invalid-email", "action": "subscribe"})
assert resp.status_code == 200
assert b"Please enter a valid email address" in resp.data