174 lines
5.5 KiB
Python
174 lines
5.5 KiB
Python
"""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"
|