"""Memory leak detection tests.""" from __future__ import annotations import gc import tracemalloc from typing import TYPE_CHECKING import pytest from app import create_app if TYPE_CHECKING: from flask import Flask from flask.testing import FlaskClient @pytest.fixture def app() -> Flask: """Create application configured for testing.""" app = create_app("testing") app.config["POW_DIFFICULTY"] = 0 app.config["RATE_LIMIT_ENABLED"] = False return app @pytest.fixture def client(app: Flask) -> FlaskClient: """Create test client.""" return app.test_client() class TestMemoryLeaks: """Test for memory leaks in common operations.""" def test_paste_create_delete_no_leak(self, client: FlaskClient) -> None: """Creating and deleting pastes should not leak memory.""" gc.collect() tracemalloc.start() # Warm-up phase (allocations may happen on first run) for _ in range(5): resp = client.post("/", data=b"warmup content") paste_id = resp.get_json()["id"] client.delete(f"/{paste_id}") gc.collect() snapshot1 = tracemalloc.take_snapshot() baseline = sum(stat.size for stat in snapshot1.statistics("lineno")) # Main test: create and delete many pastes iterations = 100 for i in range(iterations): resp = client.post("/", data=f"test content {i}".encode()) assert resp.status_code == 201 paste_id = resp.get_json()["id"] # Don't delete to accumulate if there's a leak # (pastes are in-memory during test) gc.collect() snapshot2 = tracemalloc.take_snapshot() after_creates = sum(stat.size for stat in snapshot2.statistics("lineno")) # Clean up tracemalloc.stop() # Allow reasonable growth (100 pastes * ~1KB each + overhead) # but fail if growth is excessive (> 50MB suggests a leak) growth_mb = (after_creates - baseline) / (1024 * 1024) assert growth_mb < 50, f"Memory grew by {growth_mb:.2f}MB after {iterations} creates" def test_paste_access_no_leak(self, client: FlaskClient) -> None: """Repeatedly accessing a paste should not leak memory.""" # Create a test paste resp = client.post("/", data=b"access test content") paste_id = resp.get_json()["id"] gc.collect() tracemalloc.start() # Warm-up for _ in range(10): client.get(f"/{paste_id}/raw") gc.collect() snapshot1 = tracemalloc.take_snapshot() baseline = sum(stat.size for stat in snapshot1.statistics("lineno")) # Main test: access many times iterations = 500 for _ in range(iterations): resp = client.get(f"/{paste_id}/raw") assert resp.status_code == 200 gc.collect() snapshot2 = tracemalloc.take_snapshot() after_accesses = sum(stat.size for stat in snapshot2.statistics("lineno")) tracemalloc.stop() # Access should have minimal memory growth (< 10MB) growth_mb = (after_accesses - baseline) / (1024 * 1024) assert growth_mb < 10, f"Memory grew by {growth_mb:.2f}MB after {iterations} accesses" def test_challenge_generation_no_leak(self, app: Flask) -> None: """PoW challenge generation should not leak memory.""" app.config["POW_DIFFICULTY"] = 8 client = app.test_client() gc.collect() tracemalloc.start() # Warm-up for _ in range(10): client.get("/challenge") gc.collect() snapshot1 = tracemalloc.take_snapshot() baseline = sum(stat.size for stat in snapshot1.statistics("lineno")) # Main test: generate many challenges iterations = 200 for _ in range(iterations): resp = client.get("/challenge") assert resp.status_code == 200 gc.collect() snapshot2 = tracemalloc.take_snapshot() after_challenges = sum(stat.size for stat in snapshot2.statistics("lineno")) tracemalloc.stop() # Challenge generation should have minimal growth (< 5MB) growth_mb = (after_challenges - baseline) / (1024 * 1024) assert growth_mb < 5, f"Memory grew by {growth_mb:.2f}MB after {iterations} challenges" class TestMemoryCleanup: """Test that cleanup operations release memory.""" def test_expired_paste_cleanup_releases_memory(self, app: Flask) -> None: """Expired paste cleanup should release memory.""" app.config["PASTE_EXPIRY_ANON"] = 1 # 1 second expiry client = app.test_client() # Create pastes that will expire paste_ids = [] for i in range(50): resp = client.post("/", data=f"expiring content {i}".encode()) paste_ids.append(resp.get_json()["id"]) gc.collect() tracemalloc.start() snapshot1 = tracemalloc.take_snapshot() before = sum(stat.size for stat in snapshot1.statistics("lineno")) # Wait for expiry and trigger cleanup import time time.sleep(2) # Access index to trigger cleanup client.get("/") gc.collect() snapshot2 = tracemalloc.take_snapshot() after = sum(stat.size for stat in snapshot2.statistics("lineno")) tracemalloc.stop() # Memory should not have grown significantly (cleanup should work) growth_mb = (after - before) / (1024 * 1024) assert growth_mb < 5, f"Memory grew by {growth_mb:.2f}MB after cleanup"