daniel156161
42064de633
Build and Push Docker Container / build-and-push (push) Successful in 1m3s
- Add a separate installable NanoShare CLI package under cli/. - Implement browser login with local callback and refresh-token config. - Add upload, list, download, sync, and watch CLI commands. - Add private owner-only file download endpoint for CLI downloads. - Add CLI auth endpoints for browser login and token refresh. - Return file_id from ServiceLink uploads for reliable sync state. - Exclude the CLI package from NanoShare container builds. - Include tests for private downloads and sync state updates.
128 lines
3.7 KiB
Python
128 lines
3.7 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import sys
|
|
import types
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from quart import Quart, g
|
|
|
|
class _NoOpLimiter:
|
|
def limit(self, *args, **kwargs):
|
|
return lambda fn: fn
|
|
|
|
async def _fake_token_required(required_scope=None):
|
|
def decorator(func):
|
|
async def wrapper(*args, **kwargs):
|
|
return await func(user={'sub': 'user_123', 'scope': 'files'}, *args, **kwargs)
|
|
wrapper.__name__ = func.__name__
|
|
return wrapper
|
|
return decorator
|
|
|
|
def _load_download_module():
|
|
setup_stub = types.ModuleType('my_modules.app.setup')
|
|
setup_stub.LIMITER = _NoOpLimiter()
|
|
|
|
header_stub = types.ModuleType('my_modules.decoratory.header')
|
|
|
|
def token_required(required_scope=None):
|
|
def decorator(func):
|
|
async def wrapper(*args, **kwargs):
|
|
return await func(user={'sub': 'user_123', 'scope': 'files'}, *args, **kwargs)
|
|
wrapper.__name__ = func.__name__
|
|
return wrapper
|
|
return decorator
|
|
|
|
header_stub.token_required = token_required
|
|
|
|
sys.modules['my_modules.app.setup'] = setup_stub
|
|
sys.modules['my_modules.decoratory.header'] = header_stub
|
|
|
|
module_path = Path(__file__).resolve().parents[1] / 'routes' / 'api' / 'download.py'
|
|
spec = importlib.util.spec_from_file_location('api_download_under_test', module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
class FakeConvex:
|
|
def __init__(self, file_data=None):
|
|
self.file_data = file_data
|
|
self.info_calls = []
|
|
self.storage_calls = []
|
|
|
|
async def get_file_informations(self, file_id, user_id):
|
|
self.info_calls.append((file_id, user_id))
|
|
return self.file_data
|
|
|
|
async def get_from_storage(self, storage_id):
|
|
self.storage_calls.append(storage_id)
|
|
return BytesIO(b'private bytes')
|
|
|
|
def _client(file_data):
|
|
module = _load_download_module()
|
|
app = Quart(__name__)
|
|
app.secret_key = 'test-secret'
|
|
app.convex = FakeConvex(file_data)
|
|
app.register_blueprint(module.api_download_bp)
|
|
|
|
@app.before_request
|
|
async def wide_event_context():
|
|
g.wide_event = {}
|
|
|
|
return app, app.test_client()
|
|
|
|
def test_private_download_serves_owner_file():
|
|
async def run_test():
|
|
app, client = _client({
|
|
'file_id': 'file_123',
|
|
'file_name': 'report.txt',
|
|
'content_type': 'text/plain',
|
|
'db_image_url': 'storage_123',
|
|
'uploaded_at': 1700000000000,
|
|
'user_id': 'user_123',
|
|
})
|
|
|
|
response = await client.get('/api/files/file_123/download')
|
|
|
|
assert response.status_code == 200
|
|
assert await response.get_data() == b'private bytes'
|
|
assert response.headers['Content-Type'].startswith('text/plain')
|
|
assert 'report.txt' in response.headers['Content-Disposition']
|
|
assert response.headers['Cache-Control'] == 'private, no-store'
|
|
assert app.convex.info_calls == [('file_123', 'user_123')]
|
|
assert app.convex.storage_calls == ['storage_123']
|
|
|
|
asyncio.run(run_test())
|
|
|
|
def test_private_download_returns_404_when_file_is_not_owned():
|
|
async def run_test():
|
|
app, client = _client(None)
|
|
|
|
response = await client.get('/api/files/missing/download')
|
|
|
|
assert response.status_code == 404
|
|
assert app.convex.info_calls == [('missing', 'user_123')]
|
|
assert app.convex.storage_calls == []
|
|
|
|
asyncio.run(run_test())
|
|
|
|
def test_private_download_returns_410_for_expired_file():
|
|
async def run_test():
|
|
app, client = _client({
|
|
'file_id': 'file_123',
|
|
'file_name': 'report.txt',
|
|
'content_type': 'text/plain',
|
|
'db_image_url': 'storage_123',
|
|
'uploaded_at': 1700000000000,
|
|
'expired': True,
|
|
})
|
|
|
|
response = await client.get('/api/files/file_123/download')
|
|
|
|
assert response.status_code == 410
|
|
assert app.convex.storage_calls == []
|
|
|
|
asyncio.run(run_test())
|