Compare commits

..

2 Commits

Author SHA1 Message Date
daniel156161 993337be9f feat(cli): sync folders with remote paths
Build and Push Docker Container / build-and-push (push) Successful in 1m36s
- Store synced folder locations in file_path instead of embedding them in file names.
- Add ignore rules from .nanoshareignore and repeatable sync --ignore flags.
- Adopt existing remote files only after SHA-256 verification.
- Move adopted remotes with metadata updates instead of uploading duplicates.
- Bump NanoShare to 1.23.0 and the standalone CLI to 0.2.0.
2026-07-27 20:53:00 +02:00
daniel156161 77c76b16c4 feat(cli): improve file discovery and completion
- Render remote files as an aligned table with JSON and TSV alternatives.
- Add Bash and Zsh completion for commands, options, paths, and remote files.
- Allow downloads by exact filename while rejecting ambiguous duplicate names.
- Cover table formatting, completion scripts, and selector resolution with tests.
2026-07-27 17:30:43 +02:00
14 changed files with 576 additions and 28 deletions
+53 -6
View File
@@ -9,8 +9,11 @@ from pathlib import Path
from .app import make_client
from .auth import login as auth_login
from .client import download, list_remote, upload
from .completion import script as completion_script
from .config import DEFAULT_CONFIG
from .remote_path import join_remote_path
from .sync import sync_once
from .table import format_table
DEFAULT_NODE = 'picoshare'
@@ -42,21 +45,50 @@ def _cmd_list(args) -> int:
return 1
finally:
client.close()
for item in files:
print(f"{item.get('file_id')}\t{item.get('file_name')}\t{item.get('file_size', '')}")
if args.format == 'json':
print(json.dumps(files, indent=2, ensure_ascii=False))
elif args.format == 'tsv':
for item in files:
print(f"{item.get('file_id', '')}\t{item.get('file_path', '')}\t{item.get('file_name', '')}\t{item.get('file_size', '')}")
else:
rows = [
[item.get('file_id'), item.get('file_path', ''), item.get('file_name'), item.get('file_size', '')]
for item in files
]
print(format_table(['ID', 'Path', 'Name', 'Size'], rows))
return 0
def _resolve_remote_file(files: list[dict], selector: str) -> dict:
id_matches = [item for item in files if item.get('file_id') == selector]
if id_matches:
return id_matches[0]
full_path_matches = [item for item in files if join_remote_path(str(item.get('file_name') or ''), str(item.get('file_path') or ''), str(item.get('file_id') or '')) == selector]
if full_path_matches:
if len(full_path_matches) > 1:
raise RuntimeError(f'multiple remote files at {selector!r}; use the file ID')
return full_path_matches[0]
name_matches = [item for item in files if item.get('file_name') == selector]
if not name_matches:
raise FileNotFoundError(f'no remote file matches: {selector}')
if len(name_matches) > 1:
raise RuntimeError(f'multiple remote files named {selector!r}; use the file ID or full path')
return name_matches[0]
def _cmd_download(args) -> int:
dest = Path(args.output).expanduser() if args.output else Path(args.file_id)
client = make_client(args)
try:
download(client, args.node, args.file_id, dest)
remote = _resolve_remote_file(list_remote(client, args.node), args.file)
file_id = str(remote['file_id'])
file_name = str(remote.get('file_name') or file_id)
remote_name = join_remote_path(file_name, str(remote.get('file_path') or ''), file_id)
dest = Path(args.output).expanduser() if args.output else Path(file_name)
download(client, args.node, file_id, dest)
except Exception as exc:
print(f'error: {exc}', file=sys.stderr)
return 1
finally:
client.close()
print(f'downloaded {args.file_id} -> {dest}')
print(f'downloaded {remote_name} ({file_id}) -> {dest}')
return 0
def _cmd_sync(args) -> int:
@@ -74,6 +106,14 @@ def _cmd_watch(args) -> int:
return code
time.sleep(args.interval)
def _cmd_completion(args) -> int:
try:
print(completion_script(args.shell))
except ValueError as exc:
print(f'error: {exc}', file=sys.stderr)
return 1
return 0
def _add_common(parser: argparse.ArgumentParser) -> None:
parser.add_argument('--config', default=str(DEFAULT_CONFIG), help='Path to config.toml.')
parser.add_argument('--url', action='append', metavar='NODE=URL', help='Override/add a node URL.')
@@ -111,11 +151,12 @@ def main(argv: list[str] | None = None) -> int:
list_cmd = sub.add_parser('list', help='List remote files.')
_add_common(list_cmd)
list_cmd.add_argument('--format', choices=['table', 'json', 'tsv'], default='table')
list_cmd.set_defaults(func=_cmd_list)
download_cmd = sub.add_parser('download', help='Download one file via private API endpoint.')
_add_common(download_cmd)
download_cmd.add_argument('file_id')
download_cmd.add_argument('file', help='Remote file ID or exact file name.')
download_cmd.add_argument('-o', '--output')
download_cmd.set_defaults(func=_cmd_download)
@@ -125,6 +166,7 @@ def main(argv: list[str] | None = None) -> int:
sync_cmd.add_argument('--note', default='synced from nanoshare cli')
sync_cmd.add_argument('--expires', default='')
sync_cmd.add_argument('--delete', action='store_true', help='Delete remote files that were deleted locally.')
sync_cmd.add_argument('--ignore', action='append', default=[], help='Ignore glob pattern for sync; repeatable.')
sync_cmd.set_defaults(func=_cmd_sync)
watch_cmd = sub.add_parser('watch', help='Run sync repeatedly.')
@@ -133,9 +175,14 @@ def main(argv: list[str] | None = None) -> int:
watch_cmd.add_argument('--note', default='synced from nanoshare cli')
watch_cmd.add_argument('--expires', default='')
watch_cmd.add_argument('--delete', action='store_true')
watch_cmd.add_argument('--ignore', action='append', default=[], help='Ignore glob pattern for sync; repeatable.')
watch_cmd.add_argument('--interval', type=float, default=10.0)
watch_cmd.set_defaults(func=_cmd_watch)
completion_cmd = sub.add_parser('completion', help='Print shell completion script.')
completion_cmd.add_argument('shell', choices=['zsh', 'bash'])
completion_cmd.set_defaults(func=_cmd_completion)
args = parser.parse_args(argv)
return args.func(args)
+14 -1
View File
@@ -10,7 +10,7 @@ import httpx
def content_type(path: Path) -> str:
return mimetypes.guess_type(path.name)[0] or 'application/octet-stream'
def upload(client, node: str, path: Path, remote_name: str, note: str, expires: str) -> dict:
def upload(client, node: str, path: Path, remote_name: str, note: str, expires: str, remote_path: str = '') -> dict:
content_b64 = base64.b64encode(path.read_bytes()).decode('ascii')
params = {
'file_name': remote_name,
@@ -18,11 +18,24 @@ def upload(client, node: str, path: Path, remote_name: str, note: str, expires:
'content_type': content_type(path),
'note': note,
}
if remote_path:
params['file_path'] = remote_path
if expires:
params['expires'] = expires
result = client.call(node, 'files.upload', params)
return result if isinstance(result, dict) else {'result': result}
def update_remote(client, node: str, file_id: str, file_name: str, file_path: str, note: str, expires: str) -> None:
params = {
'file_id': file_id,
'file_name': file_name,
'file_path': file_path,
'note': note,
}
if expires:
params['expires'] = expires
client.call(node, 'files.update', params)
def delete_remote(client, node: str, file_id: str) -> None:
client.call(node, 'files.delete', {'file_id': file_id})
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
_ZSH = r'''# nanoshare zsh completion -- add to ~/.zshrc: eval "$(nanoshare completion zsh)"
_nanoshare_files() {
local id path name size label
while IFS=$'\t' read -r id path name size; do
label="${path:+$path/}$name"
[[ -n $id ]] && printf '%s:%s\n' "$id" "$label"
done < <(nanoshare list --format tsv 2>/dev/null)
}
_nanoshare() {
local -a commands common_opts login_opts sync_opts watch_opts download_opts
commands=(login upload list download sync watch completion)
common_opts=(--config --url --token --refresh-token --token-url --source --timeout --node)
login_opts=(--config --node --source --scope --callback-host --callback-port --login-timeout --timeout --no-browser)
sync_opts=($common_opts --note --expires --delete --ignore)
watch_opts=($sync_opts --interval)
download_opts=($common_opts --output -o)
if (( CURRENT == 2 )); then
compadd -- $commands
return
fi
case ${words[2]} in
login)
compadd -- $login_opts
;;
upload)
compadd -- $common_opts --name --note --expires
_files
;;
list)
compadd -- $common_opts --format
;;
download)
if (( CURRENT == 3 )); then
local -a remote_files
remote_files=("${(@f)$(_nanoshare_files)}")
_describe 'remote file' remote_files
else
compadd -- $download_opts
_files
fi
;;
sync|watch)
compadd -- ${(P)${:-${words[2]}_opts}}
_files -/
;;
completion)
(( CURRENT == 3 )) && compadd -- zsh bash
;;
esac
}
compdef _nanoshare nanoshare
'''
_BASH = r'''# nanoshare bash completion -- add to ~/.bashrc: eval "$(nanoshare completion bash)"
_nanoshare_files() {
nanoshare list --format tsv 2>/dev/null
}
_nanoshare() {
local cur="${COMP_WORDS[COMP_CWORD]}" cmd="${COMP_WORDS[1]}"
local commands="login upload list download sync watch completion"
local common_opts="--config --url --token --refresh-token --token-url --source --timeout --node"
local login_opts="--config --node --source --scope --callback-host --callback-port --login-timeout --timeout --no-browser"
local sync_opts="$common_opts --note --expires --delete --ignore"
local watch_opts="$sync_opts --interval"
local download_opts="$common_opts --output -o"
if [ "$COMP_CWORD" -eq 1 ]; then
COMPREPLY=( $(compgen -W "$commands" -- "$cur") )
return
fi
case "$cmd" in
login)
COMPREPLY=( $(compgen -W "$login_opts" -- "$cur") ) ;;
upload)
COMPREPLY=( $(compgen -W "$common_opts --name --note --expires" -- "$cur") ) ;;
list)
COMPREPLY=( $(compgen -W "$common_opts --format" -- "$cur") ) ;;
download)
if [ "$COMP_CWORD" -eq 2 ]; then
COMPREPLY=( $(compgen -W "$(_nanoshare_files | cut -f1)" -- "$cur") )
else
COMPREPLY=( $(compgen -W "$download_opts" -- "$cur") )
fi ;;
sync)
COMPREPLY=( $(compgen -W "$sync_opts" -- "$cur") ) ;;
watch)
COMPREPLY=( $(compgen -W "$watch_opts" -- "$cur") ) ;;
completion)
[ "$COMP_CWORD" -eq 2 ] && COMPREPLY=( $(compgen -W "zsh bash" -- "$cur") ) ;;
esac
}
complete -F _nanoshare nanoshare
'''
def script(shell: str) -> str:
if shell == 'zsh':
return _ZSH
if shell == 'bash':
return _BASH
raise ValueError(f'unsupported shell: {shell}')
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
from fnmatch import fnmatch
from pathlib import Path
IGNORE_FILE_NAME = '.nanoshareignore'
def load_ignore_patterns(root: Path, extra_patterns: list[str] | None = None) -> list[str]:
patterns = [IGNORE_FILE_NAME]
ignore_file = root / IGNORE_FILE_NAME
if ignore_file.is_file():
for line in ignore_file.read_text().splitlines():
pattern = line.strip()
if pattern and not pattern.startswith('#'):
patterns.append(pattern)
patterns.extend(pattern for pattern in extra_patterns or [] if pattern)
return patterns
def is_ignored(rel_path: str, patterns: list[str]) -> bool:
path = rel_path.strip('/')
parts = path.split('/')
for pattern in patterns:
normalized = pattern.strip().replace('\\', '/').strip('/')
if not normalized:
continue
if pattern.endswith('/'):
if path == normalized or path.startswith(f'{normalized}/'):
return True
continue
if '/' in normalized:
if fnmatch(path, normalized):
return True
elif any(fnmatch(part, normalized) for part in parts):
return True
return False
+18
View File
@@ -0,0 +1,18 @@
from __future__ import annotations
from pathlib import PurePosixPath
def split_remote_path(value: str, file_id: str = '') -> tuple[str, str]:
raw = (value or file_id).replace('\\', '/')
parts = [
part for part in PurePosixPath(raw).parts
if part not in ('', '.', '..', '/')
]
if not parts:
return file_id, ''
return parts[-1], '/'.join(parts[:-1])
def join_remote_path(file_name: str, file_path: str | None = None, file_id: str = '') -> str:
name, embedded_path = split_remote_path(file_name, file_id)
clean_path = split_remote_path(f'{file_path or ""}/placeholder')[1] if file_path else embedded_path
return f'{clean_path}/{name}' if clean_path else name
+96 -10
View File
@@ -5,10 +5,13 @@ import json
import os
import shutil
import sys
import tempfile
import time
from pathlib import Path
from .client import delete_remote, download, list_remote, upload
from .client import delete_remote, download, list_remote, update_remote, upload
from .ignore import is_ignored, load_ignore_patterns
from .remote_path import join_remote_path, split_remote_path
SYNC_DIR_NAME = '.nanoshare-sync'
STATE_FILE_NAME = 'state.json'
@@ -44,21 +47,23 @@ def save_state(root: Path, state: dict) -> None:
tmp.write_text(json.dumps(state, indent=2, sort_keys=True))
os.replace(tmp, path)
def iter_local_files(root: Path):
def iter_local_files(root: Path, ignore_patterns: list[str] | None = None):
sync_dir = root / SYNC_DIR_NAME
patterns = ignore_patterns or []
for path in sorted(root.rglob('*')):
if not path.is_file():
continue
if path == state_path(root) or sync_dir in path.parents:
continue
if is_ignored(relative(root, path), patterns):
continue
yield path
def relative(root: Path, path: Path) -> str:
return path.relative_to(root).as_posix()
def safe_remote_name(name: str, file_id: str) -> str:
candidate = Path(name or file_id).name
return candidate or file_id
return join_remote_path(name, file_id=file_id)
def unique_path(path: Path) -> Path:
if not path.exists():
@@ -80,19 +85,74 @@ def remote_by_name(remote_files: list[dict]) -> dict[str, dict]:
result = {}
for item in remote_files:
if isinstance(item, dict) and item.get('file_name') and item.get('file_id'):
result.setdefault(item['file_name'], item)
result.setdefault(join_remote_path(item['file_name'], item.get('file_path') or ''), item)
return result
def parse_size(value: object) -> int | None:
if isinstance(value, int):
return value
text = str(value or '').strip()
if not text:
return None
parts = text.split()
try:
number = float(parts[0])
except (ValueError, IndexError):
return None
unit = parts[1].lower() if len(parts) > 1 else 'bytes'
factors = {
'byte': 1,
'bytes': 1,
'b': 1,
'kb': 1024,
'mb': 1024 ** 2,
'gb': 1024 ** 3,
'tb': 1024 ** 4,
}
factor = factors.get(unit)
return int(number * factor) if factor else None
def find_adoptable_remote(client, node: str, path: Path, local_hash: str, remote_files: list[dict], known_ids: set[str]) -> dict | None:
size = path.stat().st_size
name = path.name
candidates = []
for remote in remote_files:
if not isinstance(remote, dict):
continue
file_id = remote.get('file_id')
if not file_id or file_id in known_ids:
continue
if remote.get('file_name') != name:
continue
remote_size = parse_size(remote.get('size_bytes') or remote.get('file_size'))
if remote_size is not None and remote_size != size:
continue
candidates.append(remote)
hash_matches = []
with tempfile.TemporaryDirectory(prefix='nanoshare-adopt-') as temp_dir:
for remote in candidates:
temp_path = Path(temp_dir) / str(remote['file_id'])
try:
download(client, node, str(remote['file_id']), temp_path)
except Exception as exc:
print(f"warning: could not verify remote {remote['file_id']} for adoption: {exc}", file=sys.stderr)
continue
if sha256(temp_path) == local_hash:
hash_matches.append(remote)
return hash_matches[0] if len(hash_matches) == 1 else None
def sync_once(args, client) -> int:
root = Path(args.folder).expanduser().resolve()
root.mkdir(parents=True, exist_ok=True)
state = load_state(root)
known = state.setdefault('files', {})
try:
ignore_patterns = load_ignore_patterns(root, getattr(args, 'ignore', None))
remote_files = list_remote(client, args.node)
remote_ids = remote_by_id(remote_files)
remote_names = remote_by_name(remote_files)
local_paths = {relative(root, path): path for path in iter_local_files(root)}
local_paths = {relative(root, path): path for path in iter_local_files(root, ignore_patterns)}
for rel, path in local_paths.items():
current_hash = sha256(path)
@@ -103,13 +163,34 @@ def sync_once(args, client) -> int:
if old_hash == current_hash and remote_id in remote_ids:
continue
remote_file_name, remote_file_path = split_remote_path(rel)
if not entry:
adopted = find_adoptable_remote(client, args.node, path, current_hash, remote_files, {value.get('file_id') for value in known.values() if isinstance(value, dict)})
if adopted:
adopted_id = adopted['file_id']
current_remote_name = join_remote_path(adopted.get('file_name') or '', adopted.get('file_path') or '', adopted_id)
if current_remote_name != rel:
update_remote(client, args.node, adopted_id, remote_file_name, remote_file_path, args.note or '', args.expires or '')
print(f'move remote: {current_remote_name} -> {rel}')
known[rel] = {
'file_id': adopted_id,
'sha256': current_hash,
'local_mtime': path.stat().st_mtime,
'file_name': remote_file_name,
'file_path': remote_file_path,
}
remote_files = list_remote(client, args.node)
remote_ids = remote_by_id(remote_files)
remote_names = remote_by_name(remote_files)
continue
if remote_id and remote_id not in remote_ids and old_hash != current_hash:
conflict_path = unique_path(path)
shutil.copy2(path, conflict_path)
print(f'conflict: kept changed local copy at {conflict_path}')
print(f'upload: {rel}')
result = upload(client, args.node, path, rel, args.note or '', args.expires or '')
result = upload(client, args.node, path, remote_file_name, args.note or '', args.expires or '', remote_file_path)
new_file_id = result.get('file_id') or result.get('id')
if not new_file_id:
matching = remote_names.get(rel)
@@ -134,7 +215,8 @@ def sync_once(args, client) -> int:
'file_id': new_file_id,
'sha256': current_hash,
'local_mtime': path.stat().st_mtime,
'file_name': rel,
'file_name': remote_file_name,
'file_path': remote_file_path,
}
known_ids = {entry.get('file_id') for entry in known.values() if isinstance(entry, dict)}
@@ -142,15 +224,19 @@ def sync_once(args, client) -> int:
file_id = remote.get('file_id')
if not file_id or file_id in known_ids:
continue
name = safe_remote_name(remote.get('file_name') or '', file_id)
name = join_remote_path(remote.get('file_name') or '', remote.get('file_path') or '', file_id)
if is_ignored(name, ignore_patterns):
continue
path = unique_path(root / name)
print(f'download: {file_id} -> {relative(root, path)}')
download(client, args.node, file_id, path)
remote_file_name, remote_file_path = split_remote_path(name)
known[relative(root, path)] = {
'file_id': file_id,
'sha256': sha256(path),
'local_mtime': path.stat().st_mtime,
'file_name': relative(root, path),
'file_name': remote_file_name,
'file_path': remote_file_path,
}
if args.delete:
+26
View File
@@ -0,0 +1,26 @@
from __future__ import annotations
from collections.abc import Iterable
def _display_width(value: str) -> int:
return len(value)
def _cell(value: object) -> str:
return '' if value is None else str(value)
def format_table(headers: list[str], rows: Iterable[Iterable[object]]) -> str:
string_rows = [[_cell(value) for value in row] for row in rows]
widths = [_display_width(header) for header in headers]
for row in string_rows:
for index, value in enumerate(row):
if index < len(widths):
widths[index] = max(widths[index], _display_width(value))
def render_row(values: list[str]) -> str:
padded = [value.ljust(widths[index]) for index, value in enumerate(values)]
return ' '.join(padded).rstrip()
separator = ' '.join('-' * width for width in widths).rstrip()
lines = [render_row(headers), separator]
lines.extend(render_row(row) for row in string_rows)
return '\n'.join(lines)
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "nanoshare-cli"
version = "0.1.0"
version = "0.2.0"
description = "NanoShare desktop CLI and folder sync client"
readme = "README.md"
requires-python = ">=3.11"
+5 -3
View File
@@ -30,15 +30,16 @@ class ConvexDB(ConvexDbBase):
return [ {
"file_id": x['file_id'],
"file_name": x['file_name'],
"file_path": x.get('file_path', ''),
"file_size": x['file_size'],
"note": x['note'],
"expires_at": int(x['expires_at']) if x.get('expires_at', None) else '',
"uploaded_at": int(x['uploaded_at']),
} for x in data ]
async def add_file(self, file_name:str, file_size:str, note:str, content_type:str, expires_at:datetime|None, storage_id:str, user_id:str):
async def add_file(self, file_name:str, file_size:str, note:str, content_type:str, expires_at:datetime|None, storage_id:str, user_id:str, file_path:str=''):
args = {
'file_name': file_name, 'file_size': file_size, 'content_type': content_type,
'file_name': file_name, 'file_path': file_path, 'file_size': file_size, 'content_type': content_type,
'note': note,
'file_storage_id': storage_id, 'user_id': user_id
}
@@ -51,10 +52,11 @@ class ConvexDB(ConvexDbBase):
)
return data
async def update_file(self, file_id:str, file_name:str, note:str, expires_at:datetime|None, user_id:str):
async def update_file(self, file_id:str, file_name:str, note:str, expires_at:datetime|None, user_id:str, file_path:str=''):
args = {
'file_id': file_id,
'file_name': file_name,
'file_path': file_path,
'note': note,
'user_id': user_id
}
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "nanoshare"
version = "1.22.0"
version = "1.23.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
+16 -3
View File
@@ -12,6 +12,8 @@ from __future__ import annotations
import base64
import os
from pathlib import PurePosixPath
from quart import current_app
from my_modules.app.setup import LIMITER
@@ -24,6 +26,14 @@ RPC_SCOPES = ('files', 'mesh')
router = Router('picoshare')
def _safe_file_path(value: str) -> str:
raw = (value or '').replace('\\', '/')
parts = [
part for part in PurePosixPath(raw).parts
if part not in ('', '.', '..', '/')
]
return '/'.join(parts)
def _user_id(ctx):
if ctx.principal is None:
raise Unauthorized('authentication required')
@@ -47,10 +57,12 @@ async def files_upload(params, ctx):
else:
raise InvalidParams('provide text or content_b64')
file_name = params.get('file_name') or iso_stamp_filename('mesh', default_ext)
file_name = PurePosixPath(str(params.get('file_name') or iso_stamp_filename('mesh', default_ext)).replace('\\', '/')).name
file_path = _safe_file_path(str(params.get('file_path') or ''))
storage_id = await current_app.convex.send_to_storage(data=data, content_type=content_type)
file_record = await current_app.convex.add_file(
file_name=file_name,
file_path=file_path,
file_size=format_size(len(data)),
note=params.get('note', ''),
content_type=content_type,
@@ -59,7 +71,7 @@ async def files_upload(params, ctx):
user_id=user_id,
)
file_id = file_record.get('file_id') if isinstance(file_record, dict) else None
return {'file_id': file_id, 'file_name': file_name, 'size': len(data), 'content_type': content_type}
return {'file_id': file_id, 'file_name': file_name, 'file_path': file_path, 'size': len(data), 'content_type': content_type}
@router.method('files.list')
async def files_list(params, ctx):
@@ -85,12 +97,13 @@ async def files_info(params, ctx):
@router.method('files.update')
async def files_update(params, ctx):
file_id = params.get('file_id')
file_name = params.get('file_name')
file_name = PurePosixPath(str(params.get('file_name') or '').replace('\\', '/')).name
if not file_id or not file_name:
raise InvalidParams('file_id and file_name are required')
await current_app.convex.update_file(
file_id=file_id,
file_name=file_name,
file_path=_safe_file_path(str(params.get('file_path') or '')),
note=params.get('note', ''),
expires_at=ensure_utc(parse_expires(params.get('expires', ''))),
user_id=_user_id(ctx),
+200 -1
View File
@@ -8,6 +8,9 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'cli'))
from nanoshare_client import cli
from nanoshare_client.auth import _CallbackServer
from nanoshare_client.completion import script as completion_script
from nanoshare_client.ignore import is_ignored
from nanoshare_client.table import format_table
class FakeClient:
def __init__(self):
@@ -25,9 +28,17 @@ class FakeClient:
self.remote_files.append({
'file_id': file_id,
'file_name': params['file_name'],
'file_path': params.get('file_path', ''),
'file_size': '1 B',
})
return {'file_id': file_id, 'file_name': params['file_name']}
return {'file_id': file_id, 'file_name': params['file_name'], 'file_path': params.get('file_path', '')}
if method == 'files.update':
for item in self.remote_files:
if item['file_id'] == params['file_id']:
item['file_name'] = params['file_name']
item['file_path'] = params.get('file_path', '')
return {'updated': True}
return {'updated': False}
if method == 'files.delete':
self.deleted.append(params['file_id'])
self.remote_files = [item for item in self.remote_files if item['file_id'] != params['file_id']]
@@ -37,6 +48,10 @@ class FakeClient:
def close(self):
pass
def fake_download(client, node, file_id, path):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(f'downloaded {file_id}')
def test_sync_uploads_new_local_file_and_writes_state(tmp_path, monkeypatch):
fake = FakeClient()
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
@@ -55,6 +70,190 @@ def test_sync_uploads_new_local_file_and_writes_state(tmp_path, monkeypatch):
state = json.loads((tmp_path / '.nanoshare-sync' / 'state.json').read_text())
assert state['files']['hello.txt']['file_id'] == 'file_1'
def test_sync_uploads_nested_local_files(tmp_path, monkeypatch):
fake = FakeClient()
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
nested = tmp_path / 'docs' / 'notes' / 'hello.txt'
nested.parent.mkdir(parents=True)
nested.write_text('hello')
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
str(tmp_path),
])
assert code == 0
assert fake.uploads[0][1]['file_name'] == 'hello.txt'
assert fake.uploads[0][1]['file_path'] == 'docs/notes'
state = json.loads((tmp_path / '.nanoshare-sync' / 'state.json').read_text())
assert state['files']['docs/notes/hello.txt']['file_id'] == 'file_1'
def test_sync_adopts_and_moves_existing_remote_file(tmp_path, monkeypatch):
fake = FakeClient()
fake.remote_files = [{'file_id': 'file_1', 'file_name': 'hello.txt', 'file_path': '', 'file_size': '5 Bytes'}]
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
def matching_download(client, node, file_id, path):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text('hello')
monkeypatch.setattr('nanoshare_client.sync.download', matching_download)
nested = tmp_path / 'docs' / 'hello.txt'
nested.parent.mkdir(parents=True)
nested.write_text('hello')
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
str(tmp_path),
])
assert code == 0
assert fake.uploads == []
assert fake.remote_files[0]['file_name'] == 'hello.txt'
assert fake.remote_files[0]['file_path'] == 'docs'
state = json.loads((tmp_path / '.nanoshare-sync' / 'state.json').read_text())
assert state['files']['docs/hello.txt']['file_id'] == 'file_1'
def test_sync_does_not_adopt_same_name_with_different_hash(tmp_path, monkeypatch):
fake = FakeClient()
fake.remote_files = [{'file_id': 'file_1', 'file_name': 'hello.txt', 'file_path': '', 'file_size': '5 Bytes'}]
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
monkeypatch.setattr('nanoshare_client.sync.download', fake_download)
local = tmp_path / 'hello.txt'
local.write_text('other')
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
str(tmp_path),
])
assert code == 0
assert len(fake.uploads) == 1
assert fake.uploads[0][1]['file_name'] == 'hello.txt'
def test_sync_downloads_remote_subfolders(tmp_path, monkeypatch):
fake = FakeClient()
fake.remote_files = [{'file_id': 'file_1', 'file_name': 'hello.txt', 'file_path': 'docs/notes', 'file_size': '5 B'}]
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
monkeypatch.setattr('nanoshare_client.sync.download', fake_download)
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
str(tmp_path),
])
assert code == 0
assert (tmp_path / 'docs' / 'notes' / 'hello.txt').read_text() == 'downloaded file_1'
state = json.loads((tmp_path / '.nanoshare-sync' / 'state.json').read_text())
assert state['files']['docs/notes/hello.txt']['file_id'] == 'file_1'
def test_sync_ignores_local_files_from_nanoshareignore(tmp_path, monkeypatch):
fake = FakeClient()
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
(tmp_path / '.nanoshareignore').write_text('cache/\n*.tmp\n')
(tmp_path / 'keep.txt').write_text('keep')
(tmp_path / 'cache').mkdir()
(tmp_path / 'cache' / 'ignored.txt').write_text('ignored')
(tmp_path / 'scratch.tmp').write_text('ignored')
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
str(tmp_path),
])
assert code == 0
assert [upload[1]['file_name'] for upload in fake.uploads] == ['keep.txt']
def test_sync_ignores_remote_files_from_cli_pattern(tmp_path, monkeypatch):
fake = FakeClient()
fake.remote_files = [
{'file_id': 'file_1', 'file_name': 'keep.txt', 'file_path': '', 'file_size': '5 B'},
{'file_id': 'file_2', 'file_name': 'ignored.txt', 'file_path': 'cache', 'file_size': '5 B'},
]
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
monkeypatch.setattr('nanoshare_client.sync.download', fake_download)
code = cli.main([
'sync',
'--url', 'picoshare=https://example.com',
'--token', 'token',
'--ignore', 'cache/',
str(tmp_path),
])
assert code == 0
assert (tmp_path / 'keep.txt').is_file()
assert not (tmp_path / 'cache' / 'ignored.txt').exists()
def test_ignore_patterns_match_files_and_directories():
patterns = ['cache/', '*.tmp', 'docs/*.draft.md']
assert is_ignored('cache/file.txt', patterns)
assert is_ignored('nested/scratch.tmp', patterns)
assert is_ignored('docs/page.draft.md', patterns)
assert not is_ignored('docs/page.md', patterns)
def test_completion_scripts_include_commands():
zsh = completion_script('zsh')
bash = completion_script('bash')
assert 'compdef _nanoshare nanoshare' in zsh
assert 'complete -F _nanoshare nanoshare' in bash
assert 'login upload list download sync watch completion' in zsh
assert 'login upload list download sync watch completion' in bash
def test_resolve_remote_file_accepts_id_exact_name_or_path():
files = [
{'file_id': 'file_1', 'file_name': 'report.pdf', 'file_path': 'docs'},
{'file_id': 'file_2', 'file_name': 'photo.png'},
]
assert cli._resolve_remote_file(files, 'file_1')['file_name'] == 'report.pdf'
assert cli._resolve_remote_file(files, 'photo.png')['file_id'] == 'file_2'
assert cli._resolve_remote_file(files, 'docs/report.pdf')['file_id'] == 'file_1'
def test_resolve_remote_file_rejects_duplicate_names():
files = [
{'file_id': 'file_1', 'file_name': 'report.pdf'},
{'file_id': 'file_2', 'file_name': 'report.pdf'},
]
try:
cli._resolve_remote_file(files, 'report.pdf')
except RuntimeError as exc:
assert 'multiple remote files' in str(exc)
else:
raise AssertionError('expected duplicate names to raise')
def test_format_table_aligns_columns():
output = format_table(
['ID', 'Path', 'Name', 'Size'],
[
['abc', '', 'short.txt', '1 KB'],
['longer-id', 'docs', 'a much longer file name.png', '22 MB'],
],
)
assert output.splitlines() == [
'ID Path Name Size',
'--------- ---- --------------------------- -----',
'abc short.txt 1 KB',
'longer-id docs a much longer file name.png 22 MB',
]
def test_login_callback_returns_without_waiting_for_timeout():
server = _CallbackServer('127.0.0.1', 0, 'expected-state')
result = {}
+2 -1
View File
@@ -43,9 +43,10 @@ class FakeConvex:
async def send_to_storage(self, data, content_type):
return 'storage_1'
async def add_file(self, file_name, file_size, note, content_type, expires_at, storage_id, user_id):
async def add_file(self, file_name, file_size, note, content_type, expires_at, storage_id, user_id, file_path=''):
self.files[file_name] = {
'file_name': file_name,
'file_path': file_path,
'note': note,
'content_type': content_type,
'storage_id': storage_id,
Generated
+1 -1
View File
@@ -720,7 +720,7 @@ wheels = [
[[package]]
name = "nanoshare"
version = "1.22.0"
version = "1.23.0"
source = { virtual = "." }
dependencies = [
{ name = "aiohttp" },