fix: :memory: db handling + tests (7/7 pass, +csrf +auth tests)
- config: handle sqlite:///:memory: properly in db_path - db: skip mkdir for :memory: path - tests: use temp file instead of :memory: (avoids per-connection DB isolation) - tests: added test_csrf_rejected (403) and test_auth_user
This commit is contained in:
@@ -43,6 +43,8 @@ class Settings(BaseSettings):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def db_path(self) -> Path:
|
def db_path(self) -> Path:
|
||||||
|
if self.database_url == "sqlite:///:memory:":
|
||||||
|
return Path(":memory:") # Special SQLite in-memory
|
||||||
if self.database_url.startswith("sqlite:///"):
|
if self.database_url.startswith("sqlite:///"):
|
||||||
return Path(self.database_url.replace("sqlite:///", "/"))
|
return Path(self.database_url.replace("sqlite:///", "/"))
|
||||||
return Path("/data/flowdeck.db")
|
return Path("/data/flowdeck.db")
|
||||||
|
|||||||
@@ -14,7 +14,8 @@ def get_db_path() -> Path:
|
|||||||
def init_db():
|
def init_db():
|
||||||
"""Create tables if they don't exist."""
|
"""Create tables if they don't exist."""
|
||||||
db_path = get_db_path()
|
db_path = get_db_path()
|
||||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
if str(db_path) != ":memory:":
|
||||||
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
with get_conn() as conn:
|
with get_conn() as conn:
|
||||||
conn.execute("PRAGMA journal_mode=WAL")
|
conn.execute("PRAGMA journal_mode=WAL")
|
||||||
|
|||||||
+24
-4
@@ -1,19 +1,27 @@
|
|||||||
"""FlowDeck — Basic tests."""
|
"""FlowDeck — Basic tests."""
|
||||||
|
import os
|
||||||
|
import tempfile
|
||||||
import pytest
|
import pytest
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def client():
|
def client():
|
||||||
import os
|
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
||||||
|
db_path = db_file.name
|
||||||
|
db_file.close()
|
||||||
|
|
||||||
os.environ["GITEA_URL"] = "https://git.dracodev.net"
|
os.environ["GITEA_URL"] = "https://git.dracodev.net"
|
||||||
os.environ["GITEA_TOKEN"] = "test"
|
os.environ["GITEA_TOKEN"] = "test"
|
||||||
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
|
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
|
||||||
|
|
||||||
from app.main import app
|
from app.main import app
|
||||||
from app.db import init_db
|
from app.db import init_db
|
||||||
init_db()
|
init_db()
|
||||||
return TestClient(app)
|
|
||||||
|
yield TestClient(app)
|
||||||
|
|
||||||
|
os.unlink(db_path)
|
||||||
|
|
||||||
|
|
||||||
def test_health(client):
|
def test_health(client):
|
||||||
@@ -44,5 +52,17 @@ def test_projects(client):
|
|||||||
|
|
||||||
def test_board_404(client):
|
def test_board_404(client):
|
||||||
resp = client.get("/board/test/test")
|
resp = client.get("/board/test/test")
|
||||||
# Should render or return error, not crash
|
|
||||||
assert resp.status_code in (200, 500)
|
assert resp.status_code in (200, 500)
|
||||||
|
|
||||||
|
|
||||||
|
def test_csrf_rejected(client):
|
||||||
|
"""POST without CSRF token should return 403."""
|
||||||
|
resp = client.post("/api/move?owner=x&repo=y&issue_id=1&column=Test")
|
||||||
|
assert resp.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
def test_auth_user(client):
|
||||||
|
resp = client.get("/auth/user")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
data = resp.json()
|
||||||
|
assert "authenticated" in data
|
||||||
|
|||||||
Reference in New Issue
Block a user