Vasanko/cista-storage
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lint: apply manual ruff cleanup (non-preview files)

Browse Source
This commit is contained in:
Leo Vasanko
2026-04-26 06:16:42 +00:00
parent 9321961a11
commit 3767fb0cec
29 changed files with 169 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cista/__init__.py
+1 -3
View File
@@ -1,3 +1 @@
from cista._version import __version__
__version__ # Public API
from cista._version import __version__ as __version__
cista/__main__.py
+2 -6
View File
@@ -34,10 +34,7 @@ def create_startup_box(
location = f"{folder} @ {listen}"
lines = [title, location]
# Auth line: Paskia <url> or Password, with optional Public suffix
if paskia_url:
auth_line = f"Auth: Paskia {paskia_url}"
else:
auth_line = "Auth: Password"
auth_line = f"Auth: Paskia {paskia_url}" if paskia_url else "Auth: Password"
if public:
auth_line += ", Public"
lines.append(auth_line)
@@ -49,8 +46,7 @@ def create_startup_box(
# Build the box
box = [f"╭{'─' * inner_width}╮"]
for line in lines:
box.append(f"│ {line:<{inner_width - 1}}│")
box.extend(f"│ {line:<{inner_width - 1}}│" for line in lines)
box.append(f"╰{'─' * inner_width}╯")
return "\n".join(box) + "\n"
cista/api.py
+3 -3
View File
@@ -1,10 +1,10 @@
import asyncio
from pathlib import PurePosixPath
from secrets import token_bytes
import msgspec
from sanic import Blueprint, json
from sanic.exceptions import BadRequest
from sanic.log import logger
from cista import __version__, auth, config, sso, watching
from cista.auth import (
@@ -38,8 +38,8 @@ async def watch(req, ws):
# SSO auth: call validation to get user info (don't enforce auth in public mode)
try:
await sso.validate_sso_request(req)
except Exception:
pass # Ignore auth errors, user_info stays None
except Exception as e:
logger.debug("watch SSO validation failed: %s", e)
if sso_user := getattr(req.ctx, "sso_user", None):
ctx = sso_user.get("ctx", {})
perms = ctx.get("permissions", [])
cista/app.py
+4 -4
View File
@@ -16,10 +16,9 @@ from setproctitle import setproctitle
from stream_zip import ZIP_AUTO, stream_zip
from zstandard import ZstdCompressor
from cista import auth, config, preview, session, sso, watching
from cista.preview import shutdown_preview_workers, start_preview_workers
from cista import auth, config, fileserver, preview, session, sso, watching
from cista.api import bp
from cista import fileserver
from cista.preview import shutdown_preview_workers, start_preview_workers
from cista.sanic_logging import (
configure_access_logging,
configure_main_logging,
@@ -295,7 +294,8 @@ async def zip_download(req, keys, zipfile, ext):
while size > 0 and (chunk := f.read(min(size, 1 << 20))):
size -= len(chunk)
yield chunk
assert size == 0
if size != 0:
raise OSError(f"stream ended early while zipping {name}")
pending_put = None # Current queue.put future, can be cancelled
cista/auth.py
+16 -22
View File
@@ -1,7 +1,7 @@
import base64
import binascii
import hmac
import hashlib
import hmac
import re
import secrets
import struct
@@ -275,7 +275,6 @@ def _log_webdav_user_agent_once(request, user_agent: str):
return
_seen_webdav_uas.add(key)
# Temporary stdout print so operators can quickly capture real client UAs.
print(f"WebDAV User-Agent observed: {key} path={request.path}")
def _build_ua_auth_headers(request, *, include_hint=False) -> dict[str, str]:
@@ -285,8 +284,7 @@ def _build_ua_auth_headers(request, *, include_hint=False) -> dict[str, str]:
challenge = f'Basic realm="{_AUTH_REALM}", Negotiate'
else:
challenge = f'Basic realm="{_AUTH_REALM}"'
headers = {"WWW-Authenticate": challenge}
return headers
return {"WWW-Authenticate": challenge}
def _cleanup_ntlm_challenges():
@@ -399,14 +397,13 @@ def _spnego_wrap_ntlm_challenge(ntlm_type2: bytes) -> bytes:
neg_state_accept_incomplete = _der_tlv(0xA0, _der_tlv(0x0A, b"\x01"))
supported_mech = _der_tlv(0xA1, ntlm_oid)
response_token = _der_tlv(0xA2, _der_tlv(0x04, ntlm_type2))
neg_token_resp = _der_tlv(
return _der_tlv(
0xA1,
_der_tlv(
0x30,
neg_state_accept_incomplete + supported_mech + response_token,
),
)
return neg_token_resp
def _ntlm_parse_type3(data: bytes) -> dict | None:
@@ -417,7 +414,7 @@ def _ntlm_parse_type3(data: bytes) -> dict | None:
return None
def read_buf(offset: int) -> bytes:
length, max_len, buf_offset = struct.unpack("<HHI", data[offset : offset + 8])
length, _max_len, buf_offset = struct.unpack("<HHI", data[offset : offset + 8])
if length == 0:
return b""
if buf_offset + length > len(data):
@@ -460,7 +457,7 @@ def _ntlmv2_verify(
blob = nt_response[16:]
# NT hash = MD4(UTF-16LE(password))
nt_hash = MD4.new(token_secret.encode("utf-16le")).digest()
nt_hash = MD4.new(token_secret.encode("utf-16le")).digest() # noqa: S303
raw_username = username or ""
raw_domain = domain or ""
@@ -629,7 +626,7 @@ def _basic_auth_login(request):
async def _token_auth_login(request, privileged=False):
async def _token_auth_login(request, *, privileged=False):
"""Authenticate via Basic token:<secret> in SSO mode.
Returns True if authenticated, False if no token matched.
@@ -686,7 +683,7 @@ async def _token_auth_login(request, privileged=False):
return False
async def _ntlm_auth_login(request, privileged=False):
async def _ntlm_auth_login(request, *, privileged=False):
"""Handle NTLM authentication for token-based login.
Supports NTLMv2 responses where the token secret is used as the password.
@@ -706,9 +703,9 @@ async def _ntlm_auth_login(request, privileged=False):
try:
data = base64.b64decode(encoded)
except Exception:
except Exception as e:
logger.warning("NTLM decode failed: client=%s", client_key)
raise Unauthorized("Invalid NTLM message", www_auth_scheme, quiet=True)
raise Unauthorized("Invalid NTLM message", www_auth_scheme, quiet=True) from e
# Windows commonly sends SPNEGO-wrapped Negotiate tokens that embed NTLMSSP.
# Extract the NTLMSSP blob when present so downstream parsing sees raw Type 1/3.
@@ -797,7 +794,6 @@ async def _ntlm_auth_login(request, privileged=False):
secret_candidates.append(("token-key", token.key))
matched_by = None
matched_challenge = None
for secret_kind, secret_value in secret_candidates:
for challenge in challenges:
if _ntlmv2_verify(
@@ -808,7 +804,6 @@ async def _ntlm_auth_login(request, privileged=False):
nt_response,
):
matched_by = secret_kind
matched_challenge = challenge
break
if matched_by:
break
@@ -919,14 +914,14 @@ async def verify(request, *, privileged=False):
perm = "cista:admin" if privileged else "cista:login"
await sso.validate_sso_request(request, perm=perm)
return
except Unauthorized:
except Unauthorized as e:
auth_flow.append(f"tried={','.join(tried)} result=failed")
_set_auth_failure_log(request, auth_flow)
raise Unauthorized(
"Invalid credentials",
headers=_build_ua_auth_headers(request),
quiet=True,
)
) from e
tried.append("sso")
perm = "cista:admin" if privileged else "cista:login"
await sso.validate_sso_request(request, perm=perm)
@@ -969,7 +964,7 @@ async def verify(request, *, privileged=False):
user = await _ntlm_auth_login(request, privileged=privileged)
except Unauthorized as e:
auth_hdr = (e.headers or {}).get("WWW-Authenticate", "")
if (auth_hdr.startswith("NTLM ") or auth_hdr.startswith("Negotiate ")) and "realm=" not in auth_hdr:
if (auth_hdr.startswith(("NTLM ", "Negotiate "))) and "realm=" not in auth_hdr:
raise
ntlm_failed = True
user = None
@@ -1067,8 +1062,9 @@ async def login_page(request):
doc.style(_LOGIN_PAGE_CSS)
with doc.div(class_="login-card"):
doc.h1("Authentication Required")
with doc.div(class_="content"):
with doc.form(method="POST", id="loginForm", autocomplete="on"):
with doc.div(class_="content"), doc.form(
method="POST", id="loginForm", autocomplete="on"
):
doc.label("Username:", for_="username")
doc.input(
type="text",
@@ -1289,9 +1285,7 @@ def _token_belongs_to_user(token, username, sso_user_id):
"""Check if a token belongs to the given user."""
if username is not None and token.username == username:
return True
if sso_user_id is not None and token.sso_user_id == sso_user_id:
return True
return False
return bool(sso_user_id is not None and token.sso_user_id == sso_user_id)
# Token management handlers (shared between /auth and /api blueprints)
cista/config.py
+5 -4
View File
@@ -3,12 +3,13 @@ from __future__ import annotations
import os
import secrets
import sys
from collections.abc import Callable
from contextlib import suppress
from functools import wraps
from hashlib import sha256
from pathlib import Path, PurePath
from time import sleep, time
from typing import Callable, Concatenate, Literal, ParamSpec
from typing import Concatenate, Literal, ParamSpec
import msgspec
import msgspec.toml
@@ -49,7 +50,7 @@ class Token(msgspec.Struct, omit_defaults=True):
username: str = "" # set in built-in mode
sso_user_id: str = "" # set in SSO mode
name: str = ""
created: int = 0 # noqa: N815
created: int = 0
# Global variables - initialized during application startup
@@ -72,7 +73,7 @@ def init_confdir() -> None:
conffile = home / "db.toml"
def derived_secret(*params, len=8) -> bytes:
def derived_secret(*params, size=8) -> bytes:
"""Used to derive secret keys from the main secret"""
# Each part is made the same length by hashing first
combined = b"".join(
@@ -80,7 +81,7 @@ def derived_secret(*params, len=8) -> bytes:
for p in [config.secret, *params]
)
# Output a bytes of the desired length
return sha256(combined).digest()[:len]
return sha256(combined).digest()[:size]
def enc_hook(obj):
cista/droppy.py
+1 -1
View File
@@ -17,7 +17,7 @@ def _droppy_listeners(cf):
for listener in cf["listeners"]:
try:
if listener["protocol"] == "https":
# TODO: Add support for TLS
# TLS listeners are currently ignored here.
continue
socket = listener.get("socket")
if socket:
cista/fileio.py
+11 -7
View File
@@ -1,5 +1,6 @@
import os
import threading
from pathlib import Path
from cista import config
from cista.util import filename
@@ -31,20 +32,23 @@ class File:
if not self.writable:
# Create/open file
self.open_rw()
assert self.fd is not None
if self.fd is None:
raise RuntimeError("file descriptor is not available for write")
if file_size is not None:
assert pos + len(buffer) <= file_size
if pos + len(buffer) > file_size:
raise ValueError("write exceeds declared file size")
os.ftruncate(self.fd, file_size)
if buffer:
os.lseek(self.fd, pos, os.SEEK_SET)
os.write(self.fd, buffer)
def __getitem__(self, slice):
def __getitem__(self, slc):
if self.fd is None:
self.open_ro()
assert self.fd is not None
os.lseek(self.fd, slice.start, os.SEEK_SET)
size = slice.stop - slice.start
if self.fd is None:
raise RuntimeError("file descriptor is not available for read")
os.lseek(self.fd, slc.start, os.SEEK_SET)
size = slc.stop - slc.start
data = os.read(self.fd, size)
if len(data) < size:
raise EOFError("Error reading requested range")
@@ -71,7 +75,7 @@ class FileServer:
@staticmethod
def _stat_size(path):
try:
return os.stat(path).st_size
return Path(path).stat().st_size
except FileNotFoundError:
return None
cista/fileserver.py
+13 -13
View File
@@ -1,12 +1,14 @@
import asyncio
import contextlib
import mimetypes
import os
import re
import shutil
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from datetime import UTC, datetime
from pathlib import Path, PurePosixPath
from urllib.parse import quote as url_quote, unquote, urlparse
from urllib.parse import quote as url_quote
from urllib.parse import unquote, urlparse
from wsgiref.handlers import format_date_time
from sanic import Blueprint, HTTPResponse, empty, json
@@ -155,7 +157,7 @@ async def copy_or_move(request, name=""):
raise NotFound("Files not found", context={"missing": missing})
# Validate target shape/type before mutating anything.
for op_name, op_keys in (("cp", cp_keys), ("mv", mv_keys)):
for _op_name, op_keys in (("cp", cp_keys), ("mv", mv_keys)):
if len(op_keys) > 1 and not dst_is_dir:
raise BadRequest("Destination must be an existing directory for multiple keys")
if not op_keys:
@@ -175,7 +177,7 @@ async def copy_or_move(request, name=""):
changed: set[PurePosixPath] = set()
completed: list[dict[str, str]] = []
class _FileOpFailed(Exception):
class _FileOpError(Exception):
def __init__(self, op_name: str, key: str, error: Exception):
self.op_name = op_name
self.key = key
@@ -236,11 +238,11 @@ async def copy_or_move(request, name=""):
changed.add(dst_item_rel.parent)
completed.append({"op": op_name, "key": key})
except Exception as e:
raise _FileOpFailed(op_name, key, e) from e
raise _FileOpError(op_name, key, e) from e
try:
await asyncio.to_thread(_apply)
except _FileOpFailed as e:
except _FileOpError as e:
raise BadRequest(
"File operation failed after partial progress",
context={
@@ -310,7 +312,7 @@ async def dav_copy(request, name=""):
if not dest_header:
raise BadRequest("Missing Destination header")
overwrite = request.headers.get("overwrite", "T").strip().upper() != "F"
src_rel, src_abs = _safe_relpath(name)
_src_rel, src_abs = _safe_relpath(name)
dst_rel, dst_abs = _parse_webdav_destination(dest_header)
request.ctx._log_extra = f"→ {dst_rel}"
if not src_abs.exists():
@@ -537,7 +539,7 @@ def _parse_webdav_destination(dest_header: str) -> tuple[PurePosixPath, Path]:
return _safe_relpath(rel_str)
def _rel_to_href(rel: PurePosixPath, is_dir: bool) -> str:
def _rel_to_href(rel: PurePosixPath, *, is_dir: bool) -> str:
"""Build a DAV href from a storage-relative path."""
parts = rel.parts
if not parts:
@@ -560,10 +562,8 @@ def _collect_propfind_entries(rel: PurePosixPath, path: Path, depth: str) -> lis
if depth == "1" and path.is_dir():
for child in sorted(path.iterdir(), key=lambda p: (p.is_file(), p.name)):
child_rel = rel / child.name if rel.parts else PurePosixPath(child.name)
try:
with contextlib.suppress(OSError):
entries.append(_propfind_entry(child_rel, child))
except OSError:
pass
return entries
@@ -571,14 +571,14 @@ def _propfind_entry(rel: PurePosixPath, path: Path) -> dict:
st = path.stat()
is_dir = path.is_dir()
return {
"href": _rel_to_href(rel, is_dir),
"href": _rel_to_href(rel, is_dir=is_dir),
"name": rel.parts[-1] if rel.parts else "",
"is_dir": is_dir,
"size": st.st_size,
"etag": f'"{st.st_mtime:.0f}-{st.st_size}"',
"content_type": mimetypes.guess_type(path.name)[0] or "application/octet-stream",
"last_modified": format_date_time(st.st_mtime),
"created": datetime.fromtimestamp(st.st_ctime, tz=timezone.utc).strftime(
"created": datetime.fromtimestamp(st.st_ctime, tz=UTC).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
}
cista/protocol.py
-2
View File
@@ -4,8 +4,6 @@ from typing import Any
import msgspec
from cista import config
class ErrorMsg(msgspec.Struct):
error: dict[str, Any]
cista/sanic_logging.py
+1 -1
View File
@@ -232,7 +232,7 @@ def configure_access_logging() -> None:
_LEVEL_EMOJI = {
logging.DEBUG: "🔍",
logging.INFO: "ℹ️",
logging.INFO: "i",
logging.WARNING: "⚠️",
logging.ERROR: "🛑",
logging.CRITICAL: "🛑",
cista/serve.py
+4 -4
View File
@@ -12,7 +12,7 @@ def run(*, dev=False):
"""Run Sanic main process that spawns worker processes to serve HTTP requests."""
from .app import app
url, opts = parse_listen(config.config.listen)
_url, opts = parse_listen(config.config.listen)
# Silence Sanic's warning about running in production rather than debug
os.environ["SANIC_IGNORE_PRODUCTION_WARNING"] = "1"
confdir = config.conffile.parent
@@ -21,14 +21,14 @@ def run(*, dev=False):
server80.app.prepare(port=80, motd=False)
domain = opts["host"]
check_cert(confdir / domain, domain)
opts["ssl"] = str(confdir / domain) # type: ignore
opts["ssl"] = str(confdir / domain) # type: ignore[assignment]
app.prepare(
**opts,
motd=False,
dev=dev,
auto_reload=dev,
access_log=False,
) # type: ignore
) # type: ignore[call-arg]
if dev:
Sanic.serve()
else:
@@ -38,7 +38,7 @@ def run(*, dev=False):
def check_cert(certdir, domain):
if (certdir / "privkey.pem").exist() and (certdir / "fullchain.pem").exists():
return
# TODO: Use certbot to fetch a cert
# Certificate provisioning is external; files must exist before startup.
raise ValueError(
f"TLS certificate files privkey.pem and fullchain.pem needed in {certdir}",
)
cista/session.py
+3 -3
View File
@@ -31,9 +31,9 @@ def create(res, username, *, secure: bool = True, **kwargs):
def update(res, s, *, secure: bool = True, **kwargs):
s.update(kwargs)
s = jwt.encode(s, session_secret())
max_age = max(1, s["exp"] - int(time())) # type: ignore
res.cookies.add_cookie("s", s, httponly=True, max_age=max_age, secure=secure)
max_age = max(1, s["exp"] - int(time()))
token = jwt.encode(s, session_secret())
res.cookies.add_cookie("s", token, httponly=True, max_age=max_age, secure=secure)
def delete(res):
cista/sso.py
+7 -9
View File
@@ -126,13 +126,12 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
context=error_data,
quiet=True,
)
elif response.status_code == 403:
if response.status_code == 403:
raise Forbidden(
error_data.get("detail", "Access denied"),
context=error_data,
quiet=True,
)
else:
detail = error_data.get("detail", "")
logger.warning(
f"SSO validation {url} returned {response.status_code}: {detail}"
@@ -149,7 +148,7 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
"Authentication service unavailable",
status_code=502,
quiet=True,
)
) from e
async def check_permissions(user_id: str, perm: str) -> dict:
@@ -194,7 +193,6 @@ async def check_permissions(user_id: str, perm: str) -> dict:
error_data.get("detail", "Access denied"),
quiet=True,
)
else:
raise Forbidden(
error_data.get("detail", "Permission check failed"),
quiet=True,
@@ -206,7 +204,7 @@ async def check_permissions(user_id: str, perm: str) -> dict:
"Authentication service unavailable",
status_code=502,
quiet=True,
)
) from e
async def proxy_auth_request(request):
@@ -324,15 +322,15 @@ async def proxy_auth_websocket(request, ws):
try:
async for message in ws:
await backend_ws.send(message)
except Exception:
pass
except Exception as e:
logger.debug("WebSocket forward_to_backend ended: %s", e)
async def forward_to_client():
try:
async for message in backend_ws:
await ws.send(message)
except Exception:
pass
except Exception as e:
logger.debug("WebSocket forward_to_client ended: %s", e)
await asyncio.gather(
forward_to_backend(),
cista/util/apphelpers.py
+1 -1
View File
@@ -30,7 +30,7 @@ async def handle_sanic_exception(request, e):
context = e.context or {}
code = e.status_code
headers = getattr(e, "headers", None)
if not message or not request.app.debug and code == 500:
if not message or (not request.app.debug and code == 500):
message = "Internal Server Error"
message = f"⚠️ {message}" if code < 500 else f"🛑 {message}"
if code == 500:
cista/util/asynclink.py
+1 -1
View File
@@ -40,7 +40,7 @@ class AsyncLink:
async def stop(self):
"""Stop worker and clean up."""
while not self.queue.empty():
command, future = self.queue.get_nowait()
_command, future = self.queue.get_nowait()
if not future.done():
future.set_exception(Exception("AsyncLink stopped"))
self.queue.task_done()
cista/util/lrucache.py
+7 -7
View File
@@ -1,5 +1,5 @@
from collections.abc import Callable
from time import monotonic
from typing import Callable
class LRUCache:
@@ -7,22 +7,22 @@ class LRUCache:
LRUCache is a least-recently-used (LRU) cache with expiry time.
Attributes:
open (callable): Function to open a new handle.
opener (callable): Function to open a new handle.
capacity (int): Max number of items in the cache.
maxage (float): Max age for items in cache in seconds.
cache (list): Internal list storing the cache items.
"""
def __init__(self, open: Callable, *, capacity: int, maxage: float):
def __init__(self, opener: Callable, *, capacity: int, maxage: float):
"""
Initialize LRUCache.
Args:
open (callable): Function to open a new handle.
opener (callable): Function to open a new handle.
capacity (int): Maximum capacity of the cache.
maxage (float): Max age for items in cache in seconds.
"""
self.open = open
self.opener = opener
self.capacity = capacity
self.maxage = maxage
self.cache = [] # Each item is a tuple: (key, handle, timestamp), recent items first
@@ -47,7 +47,7 @@ class LRUCache:
self.cache.pop(i)
break
else:
f = self.open(key)
f = self.opener(key)
# Add/restore to end of cache
self.cache.insert(0, (key, f, monotonic()))
self.expire_items()
@@ -58,7 +58,7 @@ class LRUCache:
Expire items that are either too old or exceed cache capacity.
"""
ts = monotonic() - self.maxage
while len(self.cache) > self.capacity or self.cache and self.cache[-1][2] < ts:
while len(self.cache) > self.capacity or (self.cache and self.cache[-1][2] < ts):
self.cache.pop()[1].close()
def close(self):
cista/util/pwgen.py
+3 -53
View File
File diff suppressed because one or more lines are too long
cista/watching.py
+16 -13
View File
@@ -141,8 +141,8 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
state = State()
rootpath: Path = None # type: ignore
quit = threading.Event()
rootpath: Path | None = None
stop_event = threading.Event()
# Thread-safe queue for signaling path updates from websockets
_update_queue: queue.Queue[PurePosixPath] = queue.Queue()
@@ -150,9 +150,8 @@ _update_queue: queue.Queue[PurePosixPath] = queue.Queue()
def notify_change(*paths: PurePosixPath | str):
"""Signal that paths have changed. Called from control/upload websockets."""
for path in paths:
if isinstance(path, str):
path = PurePosixPath(path)
for raw_path in paths:
path = PurePosixPath(raw_path) if isinstance(raw_path, str) else raw_path
# Convert absolute paths to relative (strip leading /)
if path.is_absolute():
path = (
@@ -192,10 +191,10 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
if isfile:
return [entry]
# Walk all entries of the directory
ret: list[FileEntry] = [...] # type: ignore
ret: list[FileEntry] = []
li = []
for f in path.iterdir():
if quit.is_set():
if stop_event.is_set():
raise SystemExit("quit")
if f.name.startswith("."):
continue # No dotfiles
@@ -508,7 +507,7 @@ class PathIndex:
if lo < len(children):
return children[lo]
elif children:
if children:
# Insert after last child's subtree
last_idx = children[-1]
last_entry = self.root[last_idx]
@@ -656,7 +655,7 @@ def watcher(loop):
)
)
while not quit.is_set():
while not stop_event.is_set():
if use_inotify:
import inotify.adapters
@@ -674,7 +673,11 @@ def watcher(loop):
first_event_time: float | None = None
last_event_time: float | None = None
def add_dirty(path: PurePosixPath, source: str) -> bool:
def add_dirty(
path: PurePosixPath,
source: str,
dirty_paths=dirty_paths,
) -> bool:
"""Add path to dirty set. Returns True if added, False if redundant."""
nonlocal first_event_time, last_event_time
# Check if already covered by an existing dirty path
@@ -708,7 +711,7 @@ def watcher(loop):
last_event_time = now
return True
while not quit.is_set():
while not stop_event.is_set():
now = time.monotonic()
# Full refresh every 300s
@@ -779,7 +782,7 @@ def watcher(loop):
# Collect inotify events if available (short timeout for responsiveness)
if inotify_tree:
for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05):
if quit.is_set():
if stop_event.is_set():
return
if not (modified_flags & set(event[1])):
continue
@@ -823,5 +826,5 @@ def start(app):
def stop(app):
quit.set()
stop_event.set()
app.ctx.watcher.join()
scripts/devserver.py
+9 -4
View File
@@ -23,7 +23,12 @@ from pathlib import Path
# Import devutil from scripts/fastapi-vue (not a package, so we adjust sys.path)
sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
from devutil import ProcessGroup, logger, ready, setup_vite # type: ignore
from devutil import ( # type: ignore[import-not-found]
ProcessGroup,
logger,
ready,
setup_vite,
)
from cista import config
from cista.serve import parse_listen
@@ -40,13 +45,13 @@ def setup_sanic_backend(
"""
config.load_config()
listen = listen or config.config.listen or f":{DEFAULT_BACKEND_PORT}"
url, opts = parse_listen(listen)
_url, opts = parse_listen(listen)
port = opts.get("port", DEFAULT_BACKEND_PORT)
host = opts.get("host", "localhost") or "localhost"
# Use the current interpreter/module path so devserver always runs
# workspace source code instead of a potentially stale installed script.
cmd = [sys.executable, "-m", "cista", "--dev", "-l", listen] + extra_args
cmd = [sys.executable, "-m", "cista", "--dev", "-l", listen, *extra_args]
return f"http://{host}:{port}", cmd
@@ -59,7 +64,7 @@ async def run_devserver(
logger.warning("Frontend source not found at %s", front)
raise SystemExit(1)
frontend_url, npm_install, vite = setup_vite(frontend or "")
_frontend_url, npm_install, vite = setup_vite(frontend or "")
backend_url, sanic_cmd = setup_sanic_backend(backend, extra_args)
# Tell vite where to proxy API requests
scripts/fastapi-vue/build-frontend.py
+3 -1
View File
@@ -3,7 +3,9 @@
import sys
from pathlib import Path
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore
from hatchling.builders.hooks.plugin.interface import (
BuildHookInterface, # type: ignore[import-not-found]
)
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
scripts/fastapi-vue/buildutil.py
+9 -6
View File
@@ -30,8 +30,11 @@ def _check_node_version(node_path: str) -> None:
Raises RuntimeError if version is too old or cannot be determined.
"""
try:
result = subprocess.run(
[node_path, "--version"], capture_output=True, text=True, check=True
result = subprocess.run( # noqa: S603
[node_path, "--version"],
capture_output=True,
text=True,
check=True,
)
version_str = result.stdout.strip()
# Parse version like "v20.10.0" or "v18.17.1"
@@ -176,16 +179,16 @@ def build(folder: str = "frontend") -> None:
install_cmd, build_cmd = find_build_tool()
except RuntimeError as e:
logger.warning(e)
raise SystemExit(1)
raise SystemExit(1) from e
def run(cmd):
display_cmd = [Path(cmd[0]).name, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder)
subprocess.run(cmd, check=True, cwd=folder) # noqa: S603
try:
run(install_cmd)
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1)
except subprocess.CalledProcessError as e:
raise SystemExit(1) from e
scripts/fastapi-vue/devutil.py
+5 -8
View File
@@ -1,6 +1,7 @@
"""Utilities meant for devserver script, used only in source repository with dev deps."""
import asyncio
import contextlib
from pathlib import Path
import httpx
@@ -58,10 +59,8 @@ class ProcessGroup:
# Terminate remaining processes
for p in self._procs:
if p.returncode is None:
try:
with contextlib.suppress(ProcessLookupError):
p.terminate()
except ProcessLookupError:
pass
# Wait for all to finish (with overall timeout)
still_running = [p for p in self._procs if p.returncode is None]
@@ -74,10 +73,8 @@ class ProcessGroup:
except TimeoutError:
for p in self._procs:
if p.returncode is None:
try:
with contextlib.suppress(ProcessLookupError):
p.kill()
except ProcessLookupError:
pass
await p.wait()
@@ -95,10 +92,10 @@ async def ready(url: str, path: str = "") -> None:
await client.get(full_url, timeout=1.0)
logger.info("✓ Backend ready!")
return
except httpx.RequestError:
except httpx.RequestError as e:
if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time")
raise SystemExit(1)
raise SystemExit(1) from e
await asyncio.sleep(0.1)
tests/test_files_auth.py
+1 -2
View File
@@ -1,7 +1,6 @@
import base64
import hashlib
import hmac
import re
import struct
from pathlib import Path
from time import time
@@ -94,7 +93,7 @@ def _session_cookie_header(username: str) -> dict[str, str]:
return {"Cookie": f"s={token}"}
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
user = config.User()
auth.set_password(user, "secret")
tests/test_files_path_security.py
+3 -3
View File
@@ -10,7 +10,7 @@ from cista import config, watching
from cista.fileserver import bp as fileserver_bp
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
config.config = config.Config(path=tmp_path, listen=":0", public=True)
watching.state.root = []
@@ -33,7 +33,7 @@ async def client(setup_storage: Path):
@pytest.mark.asyncio
async def test_get_percent2F_decoded_as_path_separator(client, setup_storage: Path):
async def test_get_percent2f_decoded_as_path_separator(client, setup_storage: Path):
"""%2F in the URL path is decoded to '/' and treated as a path separator."""
(setup_storage / "sub").mkdir()
(setup_storage / "sub" / "file.txt").write_text("hello", encoding="utf-8")
@@ -45,7 +45,7 @@ async def test_get_percent2F_decoded_as_path_separator(client, setup_storage: Pa
@pytest.mark.asyncio
async def test_mkcol_percent2F_creates_nested_directory(client, setup_storage: Path):
async def test_mkcol_percent2f_creates_nested_directory(client, setup_storage: Path):
"""%2F in MKCOL path is decoded as a separator, creating nested dirs."""
_, res = await client.request("MKCOL", "/files/parent%2Fchild")
tests/test_files_rest_api.py
+1 -1
View File
@@ -10,7 +10,7 @@ from cista.fileserver import bp as fileserver_bp
from cista.protocol import FileEntry
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
config.config = config.Config(path=tmp_path, listen=":0", public=True)
watching.state.root = []
tests/test_files_static_streaming.py
+1 -1
View File
@@ -9,7 +9,7 @@ from cista import config, watching
from cista.fileserver import bp as fileserver_bp
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
config.config = config.Config(path=tmp_path, listen=":0", public=True)
watching.state.root = []
tests/test_files_webdav.py
+1 -2
View File
@@ -9,13 +9,12 @@ from sanic import Sanic
from cista import config, watching
from cista.fileserver import bp as fileserver_bp
from cista.protocol import FileEntry
_DAV_NS = "DAV:"
_METHODS = ("MKCOL", "MOVE", "COPY", "PROPFIND")
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
config.config = config.Config(path=tmp_path, listen=":0", public=True)
watching.state.root = []
tests/test_tokens.py
+5 -6
View File
@@ -1,8 +1,6 @@
from pathlib import Path
from time import time
from uuid import uuid4
import os
from pathlib import Path
from uuid import uuid4
import pytest
import pytest_asyncio
@@ -13,9 +11,10 @@ from cista.auth import bp as auth_bp
def _persist_config():
import msgspec
from pathlib import PurePath
import msgspec
def enc_hook(obj):
if isinstance(obj, PurePath):
return obj.as_posix()
@@ -25,7 +24,7 @@ def _persist_config():
config.conffile.write_bytes(msgspec.toml.encode(raw))
@pytest.fixture()
@pytest.fixture
def setup_storage(tmp_path: Path):
os.environ["CISTA_HOME"] = str(tmp_path)
config.init_confdir()