Refactor database connection to use settings and improve test isolation with temporary DB
This commit is contained in:
@@ -21,7 +21,6 @@ logger = logging.getLogger(__name__)
|
|||||||
# Database connection setup
|
# Database connection setup
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
conn = sqlite3.connect('urls.db')
|
|
||||||
conn = sqlite3.connect(settings.database_url)
|
conn = sqlite3.connect(settings.database_url)
|
||||||
conn.execute("""
|
conn.execute("""
|
||||||
CREATE TABLE IF NOT EXISTS urls (
|
CREATE TABLE IF NOT EXISTS urls (
|
||||||
|
|||||||
@@ -0,0 +1,7 @@
|
|||||||
|
# app/settings.py
|
||||||
|
|
||||||
|
import os
|
||||||
|
|
||||||
|
# Example configuration
|
||||||
|
database_url = os.getenv("DATABASE_URL", "sqlite:///default.db")
|
||||||
|
debug = os.getenv("DEBUG", "True") == "True"
|
||||||
+25
-21
@@ -3,60 +3,64 @@ import sys
|
|||||||
import os
|
import os
|
||||||
|
|
||||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||||
|
from app.main import app
|
||||||
from app.main import settings
|
from app import settings
|
||||||
import tempfile
|
import tempfile
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
client = TestClient(app)
|
@pytest.fixture
|
||||||
|
def test_client():
|
||||||
|
with tempfile.NamedTemporaryFile() as tmp:
|
||||||
|
settings.database_url = tmp.name
|
||||||
|
with TestClient(app) as c:
|
||||||
|
yield c
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture
|
||||||
def use_temp_db():
|
def use_temp_db():
|
||||||
with tempfile.NamedTemporaryFile() as tmp:
|
with tempfile.NamedTemporaryFile() as tmp:
|
||||||
settings.database_url = tmp.name
|
settings.database_url = tmp.name
|
||||||
yield
|
yield
|
||||||
|
|
||||||
def test_home():
|
def test_home(test_client):
|
||||||
response = client.get("/")
|
response = test_client.get("/")
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
assert response.json() == {"message": "URL Shortener API"}
|
assert response.json() == {"message": "URL Shortener API"}
|
||||||
|
|
||||||
def test_shorten_url():
|
def test_shorten_url(test_client):
|
||||||
response = client.post("/shorten", json={"url": "https://google.com"})
|
response = test_client.post("/shorten", json={"url": "https://google.com"})
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert "short_url" in data
|
assert "short_url" in data
|
||||||
short_url = data["short_url"]
|
short_url = data["short_url"]
|
||||||
assert short_url.startswith("http://localhost:8000/")
|
assert short_url.startswith("http://localhost:8000/")
|
||||||
|
|
||||||
def test_redirect():
|
def test_redirect(test_client):
|
||||||
response = client.post("/shorten", json={"url": "https://google.com"})
|
response = test_client.post("/shorten", json={"url": "https://google.com"})
|
||||||
short_url = response.json()["short_url"]
|
short_url = response.json()["short_url"]
|
||||||
code = short_url.split("/")[-1]
|
code = short_url.split("/")[-1]
|
||||||
|
|
||||||
redirect = client.get(f"/{code}", allow_redirects=False)
|
redirect = test_client.get(f"/{code}", follow_redirects=False)
|
||||||
assert redirect.status_code == 307
|
assert redirect.status_code == 307
|
||||||
|
|
||||||
def test_duplicate_url_returns_same_code():
|
def test_duplicate_url_returns_same_code(test_client):
|
||||||
r1 = client.post("/shorten", json={"url": "https://example.com"})
|
r1 = test_client.post("/shorten", json={"url": "https://example.com"})
|
||||||
r2 = client.post("/shorten", json={"url": "https://example.com"})
|
r2 = test_client.post("/shorten", json={"url": "https://example.com"})
|
||||||
|
|
||||||
assert r1.status_code == 200
|
assert r1.status_code == 200
|
||||||
assert r2.status_code == 200
|
assert r2.status_code == 200
|
||||||
assert r1.json()["short_url"] == r2.json()["short_url"]
|
assert r1.json()["short_url"] == r2.json()["short_url"]
|
||||||
|
|
||||||
def test_stats_endpoint():
|
def test_stats_endpoint(test_client):
|
||||||
response = client.post("/shorten", json={"url": "https://stats-test.com"})
|
response = test_client.post("/shorten", json={"url": "https://stats-test.com"})
|
||||||
code = response.json()["short_url"].split("/")[-1]
|
code = response.json()["short_url"].split("/")[-1]
|
||||||
|
|
||||||
# trigger one redirect
|
test_client.get(f"/{code}", follow_redirects=False)
|
||||||
client.get(f"/{code}", allow_redirects=False)
|
|
||||||
|
|
||||||
stats = client.get(f"/stats/{code}")
|
stats = test_client.get(f"/stats/{code}")
|
||||||
assert stats.status_code == 200
|
assert stats.status_code == 200
|
||||||
data = stats.json()
|
data = stats.json()
|
||||||
assert data["clicks"] == 1
|
assert data["clicks"] == 1
|
||||||
|
|
||||||
def test_redirect_404():
|
def test_redirect_404(test_client):
|
||||||
response = client.get("/nonexistent", allow_redirects=False)
|
response = test_client.get("/nonexistent", follow_redirects=False)
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
Reference in New Issue
Block a user