daniel156161
d3c914285c
Build and Push Docker Container / build-and-push (push) Successful in 2m3s
- Add a private SSE endpoint that emits file change events for authenticated clients. - Replace the watch sleep loop with watchdog local file events and remote SSE triggers. - Debounce local event bursts and keep watch running after transient sync failures. - Add watchdog as a standalone CLI dependency and cover watch behavior in tests. - Bump NanoShare to 1.25.0 and the standalone CLI to 0.4.0.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
import json
|
|
import sys
|
|
import threading
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
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 load_ignore_patterns, is_ignored
|
|
from nanoshare_client.table import format_table
|
|
from nanoshare_client.watch import _is_relevant_path
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.registry = {'picoshare': 'https://example.com'}
|
|
self.uploads = []
|
|
self.deleted = []
|
|
self.remote_files = []
|
|
|
|
def call(self, node, method, params):
|
|
if method == 'files.list':
|
|
return {'files': list(self.remote_files)}
|
|
if method == 'files.upload':
|
|
file_id = f'file_{len(self.uploads) + 1}'
|
|
self.uploads.append((node, params, file_id))
|
|
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'], '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', '')
|
|
if 'note' in params:
|
|
item['note'] = params['note']
|
|
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']]
|
|
return {'deleted': True}
|
|
raise AssertionError(method)
|
|
|
|
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)
|
|
|
|
(tmp_path / 'hello.txt').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'
|
|
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_updates_remote_path_when_tracked_file_moves(tmp_path, monkeypatch):
|
|
fake = FakeClient()
|
|
fake.remote_files = [{'file_id': 'file_1', 'file_name': 'old.txt', 'file_path': '', 'file_size': '5 Bytes', 'expires_at': 123456}]
|
|
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
|
|
|
|
state_dir = tmp_path / '.nanoshare-sync'
|
|
state_dir.mkdir()
|
|
(state_dir / 'state.json').write_text(json.dumps({
|
|
'version': 1,
|
|
'files': {
|
|
'old.txt': {
|
|
'file_id': 'file_1',
|
|
'sha256': '2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824',
|
|
'local_mtime': 1,
|
|
'file_name': 'old.txt',
|
|
'file_path': '',
|
|
}
|
|
}
|
|
}))
|
|
nested = tmp_path / 'docs' / 'new.txt'
|
|
nested.parent.mkdir(parents=True)
|
|
nested.write_text('hello')
|
|
|
|
code = cli.main([
|
|
'sync',
|
|
'--url', 'picoshare=https://example.com',
|
|
'--token', 'token',
|
|
'--delete',
|
|
str(tmp_path),
|
|
])
|
|
|
|
assert code == 0
|
|
assert fake.uploads == []
|
|
assert fake.deleted == []
|
|
assert fake.remote_files[0]['file_name'] == 'new.txt'
|
|
assert fake.remote_files[0]['file_path'] == 'docs'
|
|
assert fake.remote_files[0]['expires_at'] == 123456
|
|
assert 'note' not in fake.remote_files[0]
|
|
state = json.loads((state_dir / 'state.json').read_text())
|
|
assert 'old.txt' not in state['files']
|
|
assert state['files']['docs/new.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')
|
|
(tmp_path / '.comments').mkdir()
|
|
(tmp_path / '.comments' / 'ignored.xml').write_text('<comment />')
|
|
(tmp_path / '.env').write_text('SECRET=ignored')
|
|
(tmp_path / '.venv').mkdir()
|
|
(tmp_path / '.venv' / 'ignored.py').write_text('ignored')
|
|
(tmp_path / 'Thumbs.db').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_default_ignore_patterns_skip_common_generated_files(tmp_path):
|
|
patterns = load_ignore_patterns(tmp_path)
|
|
|
|
assert is_ignored('.comments/comment.xml', patterns)
|
|
assert is_ignored('.env', patterns)
|
|
assert is_ignored('.env.local', patterns)
|
|
assert is_ignored('.venv/bin/python', patterns)
|
|
assert is_ignored('venv/bin/python', patterns)
|
|
assert is_ignored('__pycache__/mod.pyc', patterns)
|
|
assert is_ignored('nested/Thumbs.db', patterns)
|
|
assert not is_ignored('docs/note.txt', patterns)
|
|
|
|
def test_watch_loop_retries_after_sync_error(monkeypatch, tmp_path):
|
|
from nanoshare_client import watch as watch_module
|
|
|
|
calls = []
|
|
|
|
class FakeFlag:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def clear(self):
|
|
pass
|
|
|
|
def wait(self, timeout):
|
|
return False
|
|
|
|
def fake_sync(args):
|
|
calls.append(args.folder)
|
|
if len(calls) == 1:
|
|
return 1
|
|
raise KeyboardInterrupt
|
|
|
|
monkeypatch.setattr(watch_module, '_ChangeFlag', FakeFlag)
|
|
monkeypatch.setattr(watch_module, '_watchdog_observer', lambda *args: None)
|
|
args = type('Args', (), {'folder': str(tmp_path), 'ignore': [], 'interval': 0.1})()
|
|
|
|
code = watch_module.watch_loop(args, fake_sync)
|
|
|
|
assert code == 0
|
|
assert calls == [str(tmp_path), str(tmp_path)]
|
|
|
|
def test_watch_relevance_ignores_sync_state_and_default_ignores(tmp_path):
|
|
patterns = load_ignore_patterns(tmp_path)
|
|
|
|
assert _is_relevant_path(tmp_path, str(tmp_path / 'note.txt'), patterns)
|
|
assert not _is_relevant_path(tmp_path, str(tmp_path / '.nanoshare-sync' / 'state.json'), patterns)
|
|
assert not _is_relevant_path(tmp_path, str(tmp_path / '.comments' / 'comment.xml'), patterns)
|
|
assert not _is_relevant_path(tmp_path, str(tmp_path / '.env'), 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 = {}
|
|
|
|
def wait_for_code():
|
|
result['code'] = server.wait_for_code(30)
|
|
|
|
thread = threading.Thread(target=wait_for_code)
|
|
thread.start()
|
|
with urllib.request.urlopen(f'{server.url}?code=login-code&state=expected-state', timeout=5) as response:
|
|
assert response.status == 200
|
|
assert b'login complete' in response.read()
|
|
thread.join(5)
|
|
|
|
assert not thread.is_alive()
|
|
assert result['code'] == 'login-code'
|
|
|
|
def test_sync_deletes_old_remote_when_local_file_changes(tmp_path, monkeypatch):
|
|
fake = FakeClient()
|
|
fake.remote_files = [{'file_id': 'old_file', 'file_name': 'hello.txt', 'file_size': '5 B'}]
|
|
monkeypatch.setattr(cli, 'make_client', lambda args: fake)
|
|
|
|
path = tmp_path / 'hello.txt'
|
|
path.write_text('changed')
|
|
state_dir = tmp_path / '.nanoshare-sync'
|
|
state_dir.mkdir()
|
|
(state_dir / 'state.json').write_text(json.dumps({
|
|
'version': 1,
|
|
'files': {
|
|
'hello.txt': {
|
|
'file_id': 'old_file',
|
|
'sha256': 'old_hash',
|
|
'local_mtime': 1,
|
|
'file_name': 'hello.txt',
|
|
}
|
|
}
|
|
}))
|
|
|
|
code = cli.main([
|
|
'sync',
|
|
'--url', 'picoshare=https://example.com',
|
|
'--token', 'token',
|
|
str(tmp_path),
|
|
])
|
|
|
|
assert code == 0
|
|
assert fake.deleted == ['old_file']
|
|
state = json.loads((state_dir / 'state.json').read_text())
|
|
assert state['files']['hello.txt']['file_id'] == 'file_1'
|