simple-nanoshare/tests/test_nanoshare_cli.py

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 is_ignored
from nanoshare_client.table import format_table

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')

  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 = {}

  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'