feat(v5.2.0): finalize Infrastructure & Polish (tests isolation, xdist, lint, CI)
FlowDeck CI / lint (push) Successful in 43s
FlowDeck CI / test (push) Successful in 4m2s
FlowDeck CI / lint (pull_request) Successful in 42s
FlowDeck CI / test (pull_request) Successful in 4m3s
FlowDeck CI / docker (push) Successful in 1m2s
FlowDeck CI / docker (pull_request) Successful in 35s

tests/conftest.py: mutate the settings singleton (instead of rebinding) so DB + backup dir are isolated per test -> pytest-xdist safe.
Real backup tests (snapshot/prune/admin API) and OAuth mock tests (Gitea/GitHub/link) replace the previous skips.
init_db() now also creates webhook_subscriptions (full schema without the FastAPI lifespan).
ruff check is clean; .eslintrc.json migrated to eslint.config.mjs (flat config).
CI: lint job (ruff + eslint), parallel tests (-n auto), run on every branch push.
VERSION 5.11.1.
This commit is contained in:
2026-09-11 23:36:53 -04:00
parent 881c3e3e0a
commit ba363eaee9
42 changed files with 566 additions and 206 deletions
+203 -9
View File
@@ -148,16 +148,64 @@ def test_onboarding_workspace_and_project(client):
# ═══════════════ Backups ═══════════════
def test_backup_snapshot_and_pruning(client):
"""Skipped: flaky due to test isolation issues with global config state.
Passes when run in isolation.
"""
pytest.skip("Flaky: backup_dir cleanup interference between tests")
"""backup_db() snapshots the SQLite file and prune keeps the newest N."""
import datetime
def test_backup_admin_api_requires_admin(client):
"""Skipped: flaky due to test isolation issues with global config state.
Passes when run in isolation.
"""
pytest.skip("Flaky: admin API test interference between tests")
from app.services import backup as backup_svc
base = datetime.datetime(2026, 1, 1, 0, 0, 0)
first = backup_svc.backup_db(now=base)
assert first and first.startswith("flowdeck-") and first.endswith(".db")
backups = backup_svc.list_backups()
assert len(backups) == 1
assert backups[0]["filename"] == first
assert backups[0]["size"] > 0
# Distinct timestamps → distinct filenames (the format has 1-second resolution).
for i in range(1, 4):
assert backup_svc.backup_db(now=base + datetime.timedelta(seconds=i))
assert len(backup_svc.list_backups()) == 4
removed = backup_svc.prune_old_backups(keep=2)
assert removed == 2
remaining = backup_svc.list_backups()
assert len(remaining) == 2
# Newest first (filename sort == chronological for this format).
assert remaining[0]["filename"] > remaining[1]["filename"]
# Age + due logic (file mtime is "now", so a fresh backup is not due).
assert backup_svc.last_backup_age_hours() is not None
assert backup_svc.backup_due() is False
def test_backup_disabled_returns_none(client):
from app.config import settings
from app.services import backup as backup_svc
previous = settings.backup_enabled
settings.backup_enabled = False
try:
assert backup_svc.backup_db() is None
finally:
settings.backup_enabled = previous
def test_backup_admin_api(client):
"""The backup admin API is admin-only and snapshots on demand."""
# Unauthenticated → forbidden.
assert client.post("/api/settings/backups/run").status_code == 403
# First registered user becomes admin → allowed.
_register(client)
run = client.post("/api/settings/backups/run")
assert run.status_code == 200, run.text
assert run.json()["status"] == "ok"
assert run.json()["filename"].startswith("flowdeck-")
listing = client.get("/api/settings/backups")
assert listing.status_code == 200
assert len(listing.json()["backups"]) >= 1
# ═══════════════ Projects registry ═══════════════
@@ -266,6 +314,152 @@ def test_projects_sync_with_mock_client(client):
os.environ.setdefault("PROJECT_SYNC_ENABLED", "false")
# ═══════════════ OAuth integration (mocked providers) ═══════════════
class _FakeProvider:
"""Stand-in OAuth provider — no network calls."""
name = "gitea"
icon = "🔗"
def __init__(self, login="octocat", full_name="Octo Cat", email="[email protected]"):
self.login = login
self.full_name = full_name
self.email = email
def is_enabled(self) -> bool:
return True
def get_authorize_url(self, state, redirect_uri=None, force_login=False):
return f"https://gitea.test/login/oauth/authorize?client_id=cid&state={state}"
async def exchange_code(self, code, redirect_uri=None):
return {"access_token": "tok-123", "refresh_token": "ref-123"}
async def get_user(self, access_token):
return {
"login": self.login,
"full_name": self.full_name,
"email": self.email,
"avatar_url": "https://gitea.test/avatar.png",
"provider_id": "42",
}
async def list_repositories(self, access_token):
return []
def _patch_provider(monkeypatch, provider=None):
from app.auth import providers
fake = provider or _FakeProvider()
monkeypatch.setattr(providers, "get_provider", lambda name: fake if name in ("gitea", "github") else None)
return fake
def _extract_state(location: str) -> str:
from urllib.parse import parse_qs, urlparse
return parse_qs(urlparse(location).query)["state"][0]
def test_oauth_login_callback_flow(client, monkeypatch):
"""Full login flow: authorize redirect → callback → user + token + session."""
_patch_provider(monkeypatch)
login = client.get("/auth/login?provider=gitea", follow_redirects=False)
assert login.status_code == 302
assert "gitea.test/login/oauth/authorize" in login.headers["location"]
state = _extract_state(login.headers["location"])
cb = client.get(f"/auth/callback?code=abc&state={state}", follow_redirects=False)
assert cb.status_code == 302
assert "/workspaces" in cb.headers["location"]
assert client.cookies.get("flowdeck_session")
from app.db import get_conn
with get_conn() as conn:
user = conn.execute("SELECT id, auth_method FROM users WHERE login='gitea_octocat'").fetchone()
assert user and user["auth_method"] == "gitea"
token = conn.execute(
"SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='gitea'",
(user["id"],),
).fetchone()
assert token and token["access_token"] == "tok-123"
me = client.get("/auth/user").json()
assert me["authenticated"] is True
assert me["user"]["login"] == "gitea_octocat"
def test_oauth_github_callback_creates_github_user(client, monkeypatch):
_patch_provider(monkeypatch, _FakeProvider(login="hubber", full_name="Hub Ber"))
login = client.get("/auth/login?provider=github", follow_redirects=False)
assert login.status_code == 302
state = _extract_state(login.headers["location"])
cb = client.get(f"/auth/callback?code=xyz&state={state}", follow_redirects=False)
assert cb.status_code == 302
from app.db import get_conn
with get_conn() as conn:
user = conn.execute("SELECT id, auth_method FROM users WHERE login='github_hubber'").fetchone()
assert user and user["auth_method"] == "github"
token = conn.execute(
"SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='github'",
(user["id"],),
).fetchone()
assert token and token["access_token"] == "tok-123"
def test_oauth_link_mode_attaches_to_current_user(client, monkeypatch):
"""mode=link must attach the forge token to the already-logged-in user."""
_register(client)
_patch_provider(monkeypatch)
login = client.get("/auth/login?provider=gitea&mode=link", follow_redirects=False)
state = _extract_state(login.headers["location"])
assert state.endswith(":link")
cb = client.get(f"/auth/callback?code=abc&state={state}", follow_redirects=False)
assert cb.status_code == 302
assert "settings" in cb.headers["location"]
from app.db import get_conn
with get_conn() as conn:
user = conn.execute("SELECT id FROM users WHERE login='[email protected]'").fetchone()
token = conn.execute(
"SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='gitea'",
(user["id"],),
).fetchone()
assert token and token["access_token"] == "tok-123"
# No new OAuth user was created.
count = conn.execute("SELECT COUNT(*) FROM users WHERE login LIKE 'gitea_%'").fetchone()[0]
assert count == 0
def test_oauth_callback_rejects_invalid_state(client, monkeypatch):
_patch_provider(monkeypatch)
client.get("/auth/login?provider=gitea", follow_redirects=False)
bad = client.get("/auth/callback?code=abc&state=tampered", follow_redirects=False)
assert bad.status_code == 400
def test_oauth_provider_authorize_urls():
from app.auth.providers import GiteaProvider, GitHubProvider
gitea = GiteaProvider("https://git.example.com", "cid", "sec", "https://app/auth/callback")
assert gitea.is_enabled()
gitea_url = gitea.get_authorize_url("st", redirect_uri="https://app/auth/callback")
assert gitea_url.startswith("https://git.example.com/login/oauth/authorize?")
assert "client_id=cid" in gitea_url and "state=st" in gitea_url
github = GitHubProvider("cid", "sec", "https://app/auth/callback")
assert github.is_enabled()
assert "github.com/login/oauth/authorize" in github.get_authorize_url("st")
assert not GitHubProvider("", "", "https://app/auth/callback").is_enabled()
# ═══════════════ Multi-user permissions ═══════════════
def test_permission_manager_roles(client):