daniel156161
2982d44e55
Build and Push Docker Container / build-and-push (push) Successful in 3m21s
- Add the servicelink submodule and register POST /rpc for node-to-node file operations. - Require bearer tokens with the mesh scope and apply rate/body-size limits to RPC calls. - Map database connectivity failures to the existing 504 database error flow, with JSON responses for API routes. - Cover the new RPC handlers and database error handling with focused pytest tests. - Bump the NanoShare package version to 1.21.0.
107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from quart import Quart
|
|
|
|
from servicelink import Request
|
|
|
|
# Load routes/api/link.py in isolation. Importing it as `routes.api.link` would
|
|
# run routes/__init__.py -> my_modules.app.setup -> constens (asyncio.run at
|
|
# import), which fails under pytest's output capture. We stub my_modules.app.setup
|
|
# with a no-op limiter; link.py's other deps (my_modules.expiry/file_meta,
|
|
# servicelink, quart) are side-effect free.
|
|
class _NoOpLimiter:
|
|
def limit(self, *args, **kwargs):
|
|
return lambda fn: fn
|
|
|
|
_setup_stub = types.ModuleType('my_modules.app.setup')
|
|
_setup_stub.LIMITER = _NoOpLimiter()
|
|
sys.modules.setdefault('my_modules.app.setup', _setup_stub)
|
|
|
|
_LINK_PATH = Path(__file__).resolve().parents[1] / 'routes' / 'api' / 'link.py'
|
|
_spec = importlib.util.spec_from_file_location('picoshare_link_under_test', _LINK_PATH)
|
|
link = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(link)
|
|
|
|
class FakeConvex:
|
|
def __init__(self):
|
|
self.files = {}
|
|
|
|
async def decode_access_token_payload(self, access_token, required_scope=None, required_scopes=None):
|
|
if access_token == 'good':
|
|
return {'sub': 'user_123', 'scope': 'all'}
|
|
if access_token == 'nomesh':
|
|
return {'sub': 'user_123', 'scope': 'files'}
|
|
return {'error': 'invalid token'}
|
|
|
|
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):
|
|
self.files[file_name] = {
|
|
'file_name': file_name,
|
|
'note': note,
|
|
'content_type': content_type,
|
|
'storage_id': storage_id,
|
|
'user_id': user_id,
|
|
}
|
|
return self.files[file_name]
|
|
|
|
async def get_file(self, file_id):
|
|
return self.files.get(file_id)
|
|
|
|
def _call(envelope, token=None):
|
|
# Call the handler the /rpc route delegates to, bypassing the LIMITER/size-cap
|
|
# wrapper; exercises the real verify (mesh scope) + dispatch + handlers.
|
|
async def run():
|
|
app = Quart(__name__)
|
|
app.convex = FakeConvex()
|
|
async with app.test_request_context('/rpc', method='POST'):
|
|
status, body, _ = await link.handle_envelope(
|
|
link.router,
|
|
json.dumps(envelope).encode('utf-8'),
|
|
authorization=f'Bearer {token}' if token else None,
|
|
verify=link._verify,
|
|
content_type='application/json',
|
|
)
|
|
return status, json.loads(body)
|
|
|
|
return asyncio.run(run())
|
|
|
|
def _request(method, params=None):
|
|
return Request.create(method, params or {}, source='browser-cli', target='picoshare').to_dict()
|
|
|
|
def test_text_upload_roundtrip():
|
|
status, body = _call(_request('files.upload', {'text': 'hello mesh', 'file_name': 'note.txt', 'note': 'scraped'}), token='good')
|
|
assert status == 200
|
|
assert body['ok'] is True
|
|
assert body['result']['file_name'] == 'note.txt'
|
|
assert body['result']['content_type'] == 'text/plain'
|
|
assert body['result']['size'] == len('hello mesh')
|
|
|
|
def test_upload_requires_content():
|
|
status, body = _call(_request('files.upload', {'file_name': 'x'}), token='good')
|
|
assert status == 422
|
|
assert body['error']['code'] == 'invalid_params'
|
|
|
|
def test_get_missing_file_is_not_found():
|
|
status, body = _call(_request('files.get', {'file_id': 'ghost'}), token='good')
|
|
assert status == 404
|
|
assert body['error']['code'] == 'not_found'
|
|
|
|
def test_missing_token_is_unauthorized():
|
|
status, body = _call(_request('files.list'))
|
|
assert status == 401
|
|
assert body['error']['code'] == 'unauthorized'
|
|
|
|
def test_token_without_mesh_scope_is_forbidden():
|
|
status, body = _call(_request('files.list'), token='nomesh')
|
|
assert status == 403
|
|
assert body['error']['code'] == 'forbidden'
|