daniel156161
9db36c2d5b
Build and Push Docker Container / build-and-push (push) Successful in 1m2s
- Accept files-scoped bearer tokens on servicelink RPC calls. - Keep mesh shared-secret auth for trusted internal callers. - Validate CLI auth scopes and reject unsupported values early. - Stop CLI browser login waiting for the full timeout after callback. - Add tests for scope normalization, RPC access, and login callback timing.
107 lines
3.1 KiB
Python
107 lines
3.1 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
|
|
|
|
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_size': '1 B',
|
|
})
|
|
return {'file_id': file_id, 'file_name': params['file_name']}
|
|
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 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_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'
|