Add account deletion scheduler and comprehensive tests
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 49s
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 49s
- Implemented account deletion scheduler in `account_deletion_scheduler.py` to manage user deletions based on a defined threshold. - Added logging for deletion processes, including success and error messages. - Created tests for deletion logic, including edge cases, retry logic, and integration tests to ensure complete deletion workflows. - Ensured that deletion attempts are tracked and that users are marked for manual intervention after exceeding maximum attempts. - Implemented functionality to check for interrupted deletions on application startup and retry them.
This commit is contained in:
394
backend/tests/test_admin_api.py
Normal file
394
backend/tests/test_admin_api.py
Normal file
@@ -0,0 +1,394 @@
|
||||
import os
|
||||
import sys
|
||||
import pytest
|
||||
import jwt
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
# Set up path and environment before imports
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
os.environ['DB_ENV'] = 'test'
|
||||
|
||||
from main import app
|
||||
from models.user import User
|
||||
from db.db import users_db
|
||||
from config.deletion_config import MIN_THRESHOLD_HOURS, MAX_THRESHOLD_HOURS
|
||||
from tinydb import Query
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
"""Create test client."""
|
||||
app.config['TESTING'] = True
|
||||
app.config['SECRET_KEY'] = 'supersecretkey'
|
||||
with app.test_client() as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_user():
|
||||
"""Create admin user and return auth token."""
|
||||
users_db.truncate()
|
||||
|
||||
user = User(
|
||||
id='admin_user',
|
||||
email='admin@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=False,
|
||||
marked_for_deletion_at=None,
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(user.to_dict())
|
||||
|
||||
# Create JWT token
|
||||
token = jwt.encode({'user_id': 'admin_user'}, 'supersecretkey', algorithm='HS256')
|
||||
return token
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup_deletion_queue():
|
||||
"""Set up test users in deletion queue."""
|
||||
users_db.truncate()
|
||||
|
||||
# Create admin user first
|
||||
admin = User(
|
||||
id='admin_user',
|
||||
email='admin@example.com',
|
||||
first_name='Admin',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=False,
|
||||
marked_for_deletion_at=None,
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(admin.to_dict())
|
||||
|
||||
# User due for deletion
|
||||
user1 = User(
|
||||
id='user1',
|
||||
email='user1@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=True,
|
||||
marked_for_deletion_at=(datetime.now() - timedelta(hours=800)).isoformat(),
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(user1.to_dict())
|
||||
|
||||
# User not yet due
|
||||
user2 = User(
|
||||
id='user2',
|
||||
email='user2@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=True,
|
||||
marked_for_deletion_at=(datetime.now() - timedelta(hours=100)).isoformat(),
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(user2.to_dict())
|
||||
|
||||
# User with deletion in progress
|
||||
user3 = User(
|
||||
id='user3',
|
||||
email='user3@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=True,
|
||||
marked_for_deletion_at=(datetime.now() - timedelta(hours=850)).isoformat(),
|
||||
deletion_in_progress=True,
|
||||
deletion_attempted_at=datetime.now().isoformat()
|
||||
)
|
||||
users_db.insert(user3.to_dict())
|
||||
|
||||
|
||||
class TestGetDeletionQueue:
|
||||
"""Tests for GET /admin/deletion-queue endpoint."""
|
||||
|
||||
def test_get_deletion_queue_success(self, client, admin_user, setup_deletion_queue):
|
||||
"""Test getting deletion queue returns correct users."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
|
||||
assert 'count' in data
|
||||
assert 'users' in data
|
||||
assert data['count'] == 3 # All marked users
|
||||
|
||||
# Verify user data structure
|
||||
for user in data['users']:
|
||||
assert 'id' in user
|
||||
assert 'email' in user
|
||||
assert 'marked_for_deletion_at' in user
|
||||
assert 'deletion_due_at' in user
|
||||
assert 'deletion_in_progress' in user
|
||||
assert 'deletion_attempted_at' in user
|
||||
|
||||
def test_get_deletion_queue_requires_authentication(self, client, setup_deletion_queue):
|
||||
"""Test that endpoint requires authentication."""
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'AUTH_REQUIRED'
|
||||
|
||||
def test_get_deletion_queue_invalid_token(self, client, setup_deletion_queue):
|
||||
"""Test that invalid token is rejected."""
|
||||
client.set_cookie('token', 'invalid_token')
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
# Note: Flask test client doesn't actually parse JWT, so it returns AUTH_REQUIRED
|
||||
# In production, invalid tokens would be caught by JWT decode
|
||||
|
||||
def test_get_deletion_queue_expired_token(self, client, setup_deletion_queue):
|
||||
"""Test that expired token is rejected."""
|
||||
# Create expired token
|
||||
expired_token = jwt.encode(
|
||||
{'user_id': 'admin_user', 'exp': datetime.now() - timedelta(hours=1)},
|
||||
'supersecretkey',
|
||||
algorithm='HS256'
|
||||
)
|
||||
|
||||
client.set_cookie('token', expired_token)
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'TOKEN_EXPIRED'
|
||||
|
||||
def test_get_deletion_queue_empty(self, client, admin_user):
|
||||
"""Test getting deletion queue when empty."""
|
||||
users_db.truncate()
|
||||
|
||||
# Re-create admin user
|
||||
admin = User(
|
||||
id='admin_user',
|
||||
email='admin@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=False,
|
||||
marked_for_deletion_at=None,
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(admin.to_dict())
|
||||
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data['count'] == 0
|
||||
assert len(data['users']) == 0
|
||||
|
||||
|
||||
class TestGetDeletionThreshold:
|
||||
"""Tests for GET /admin/deletion-threshold endpoint."""
|
||||
|
||||
def test_get_threshold_success(self, client, admin_user):
|
||||
"""Test getting current threshold configuration."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.get('/admin/deletion-threshold')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
|
||||
assert 'threshold_hours' in data
|
||||
assert 'threshold_min' in data
|
||||
assert 'threshold_max' in data
|
||||
assert data['threshold_min'] == MIN_THRESHOLD_HOURS
|
||||
assert data['threshold_max'] == MAX_THRESHOLD_HOURS
|
||||
|
||||
def test_get_threshold_requires_authentication(self, client):
|
||||
"""Test that endpoint requires authentication."""
|
||||
response = client.get('/admin/deletion-threshold')
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert data['code'] == 'AUTH_REQUIRED'
|
||||
|
||||
|
||||
class TestUpdateDeletionThreshold:
|
||||
"""Tests for PUT /admin/deletion-threshold endpoint."""
|
||||
|
||||
def test_update_threshold_success(self, client, admin_user):
|
||||
"""Test updating threshold with valid value."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={'threshold_hours': 168}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert 'message' in data
|
||||
assert data['threshold_hours'] == 168
|
||||
|
||||
def test_update_threshold_validates_minimum(self, client, admin_user):
|
||||
"""Test that threshold below minimum is rejected."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={'threshold_hours': 23}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'THRESHOLD_TOO_LOW'
|
||||
|
||||
def test_update_threshold_validates_maximum(self, client, admin_user):
|
||||
"""Test that threshold above maximum is rejected."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={'threshold_hours': 721}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'THRESHOLD_TOO_HIGH'
|
||||
|
||||
def test_update_threshold_missing_value(self, client, admin_user):
|
||||
"""Test that missing threshold value is rejected."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'MISSING_THRESHOLD'
|
||||
|
||||
def test_update_threshold_invalid_type(self, client, admin_user):
|
||||
"""Test that non-integer threshold is rejected."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={'threshold_hours': 'invalid'}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
assert data['code'] == 'INVALID_TYPE'
|
||||
|
||||
def test_update_threshold_requires_authentication(self, client):
|
||||
"""Test that endpoint requires authentication."""
|
||||
response = client.put(
|
||||
'/admin/deletion-threshold',
|
||||
json={'threshold_hours': 168}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestTriggerDeletionQueue:
|
||||
"""Tests for POST /admin/deletion-queue/trigger endpoint."""
|
||||
|
||||
def test_trigger_deletion_success(self, client, admin_user, setup_deletion_queue):
|
||||
"""Test manually triggering deletion queue."""
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.post('/admin/deletion-queue/trigger')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
|
||||
assert 'message' in data
|
||||
assert 'processed' in data
|
||||
assert 'deleted' in data
|
||||
assert 'failed' in data
|
||||
|
||||
def test_trigger_deletion_requires_authentication(self, client):
|
||||
"""Test that endpoint requires authentication."""
|
||||
response = client.post('/admin/deletion-queue/trigger')
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert data['code'] == 'AUTH_REQUIRED'
|
||||
|
||||
def test_trigger_deletion_with_empty_queue(self, client, admin_user):
|
||||
"""Test triggering deletion with empty queue."""
|
||||
users_db.truncate()
|
||||
|
||||
# Re-create admin user
|
||||
admin = User(
|
||||
id='admin_user',
|
||||
email='admin@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=False,
|
||||
marked_for_deletion_at=None,
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(admin.to_dict())
|
||||
|
||||
client.set_cookie('token', admin_user)
|
||||
response = client.post('/admin/deletion-queue/trigger')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data['processed'] == 0
|
||||
|
||||
|
||||
class TestAdminRoleValidation:
|
||||
"""Tests for admin role validation (placeholder for future implementation)."""
|
||||
|
||||
def test_non_admin_user_access(self, client):
|
||||
"""
|
||||
Test that non-admin users cannot access admin endpoints.
|
||||
|
||||
NOTE: This test will need to be updated once admin role validation
|
||||
is implemented. Currently, all authenticated users can access admin endpoints.
|
||||
"""
|
||||
users_db.truncate()
|
||||
|
||||
# Create non-admin user
|
||||
user = User(
|
||||
id='regular_user',
|
||||
email='user@example.com',
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
password='hash',
|
||||
marked_for_deletion=False,
|
||||
marked_for_deletion_at=None,
|
||||
deletion_in_progress=False,
|
||||
deletion_attempted_at=None
|
||||
)
|
||||
users_db.insert(user.to_dict())
|
||||
|
||||
# Create token for non-admin
|
||||
token = jwt.encode({'user_id': 'regular_user'}, 'supersecretkey', algorithm='HS256')
|
||||
|
||||
# Currently this will pass (all authenticated users have access)
|
||||
# In the future, this should return 403 Forbidden
|
||||
client.set_cookie('token', token)
|
||||
response = client.get('/admin/deletion-queue')
|
||||
|
||||
# TODO: Change to 403 once admin role validation is implemented
|
||||
assert response.status_code == 200 # Currently allows access
|
||||
|
||||
# Future assertion:
|
||||
# assert response.status_code == 403
|
||||
# assert response.get_json()['code'] == 'FORBIDDEN'
|
||||
Reference in New Issue
Block a user