Files
daniel156161 3db374b1ab
Build and Push Docker Container / build-and-push (push) Successful in 2m37s
feat(files): cache served files and add HTML text viewer
File cache:
- Add my_modules/filecache with metadata, blob, validator and store layers
- Cache convex.get_file via AsyncCache, TTL clamped to the file expiry
- Skip caching for missing, incomplete or already expired records
- Add a process-local, size-bounded LRU for immutable storage blobs
- Build the cache stack from FILE_CACHE_* env vars in build_file_cache
- Attach the cache to the app in my_modules/app/setup.py
- Expose app-bound helpers in filecache/access.py for routes

Serving:
- Send strong ETag, Last-Modified and Cache-Control on served files
- Answer If-None-Match and If-Modified-Since with 304 from metadata only
- Avoid pulling the blob from Convex on a revalidation hit
- Add a short Cloudflare-CDN-Cache-Control edge window for deleted files
- Record cache_status in the wide event on a 304
- Invalidate cached metadata after file update and delete in both routes

Text viewer:
- Add my_modules/textview to render text/plain and text/markdown as HTML
- Linkify http(s) URLs and trim trailing sentence punctuation
- Resolve links pointing at this instance to the referenced file
- Inline image previews, mark missing and expired targets
- Cap resolution at MAX_LOOKUPS backend lookups per document
- Serve the viewer only for real browser navigations, never for curl or CLI
- Keep raw bytes for ?raw=1, downloads and Accept headers without text/html
- Add templates/side/views/webpage/files/text_view.htm with a raw link

Other:
- Add tests for cache layers, serving, text rendering and the route
- Bump multidict to 7.2.1 in uv.lock
- Update the my_helpers submodule pointer
2026-10-25 18:54:46 +01:00

173 lines
6.6 KiB
Python

import asyncio
import pytest
from my_modules.textview import (
build_segments,
decode_text,
parse_share_link,
tokenize,
wants_text_view,
)
EXAMPLE = (
'Ref Barend: https://i.example.dev/-Q6jmHoRYjFi\n'
'Ref Hanzoku: https://i.example.dev/-eluTduc-npP in female shape like '
'https://i.example.dev/-kzmIlsf7uUe please\n\n'
'Hoodie/Skateboard : https://i.example.dev/-to4wPonAadV\n\n'
'Barend has just returned from a skating trip.'
)
# --- tokenize -----------------------------------------------------------------
def test_tokenize_finds_every_url_in_the_example():
urls = [value for kind, value in tokenize(EXAMPLE) if kind == 'url']
assert urls == [
'https://i.example.dev/-Q6jmHoRYjFi',
'https://i.example.dev/-eluTduc-npP',
'https://i.example.dev/-kzmIlsf7uUe',
'https://i.example.dev/-to4wPonAadV',
]
def test_tokenize_keeps_text_lossless():
assert ''.join(value for _, value in tokenize(EXAMPLE)) == EXAMPLE
@pytest.mark.parametrize('raw,expected', [
('see https://example.com/-abc.', 'https://example.com/-abc'),
('see https://example.com/-abc, next', 'https://example.com/-abc'),
('(https://example.com/-abc)', 'https://example.com/-abc'),
('https://example.com/wiki/Foo_(bar)', 'https://example.com/wiki/Foo_(bar)'),
])
def test_tokenize_trims_sentence_punctuation(raw, expected):
assert [value for kind, value in tokenize(raw) if kind == 'url'] == [expected]
@pytest.mark.parametrize('raw', [
'javascript:alert(1)',
'data:text/html,<script>alert(1)</script>',
'ftp://example.com/file',
])
def test_tokenize_ignores_non_http_schemes(raw):
assert [kind for kind, _ in tokenize(raw)] == ['text']
# --- parse_share_link ---------------------------------------------------------
@pytest.mark.parametrize('url,expected', [
('https://i.example.dev/-abc123', 'abc123'),
('http://i.example.dev/-abc123', 'abc123'),
('https://I.Example.Dev/-abc123', 'abc123'),
('https://i.example.dev/-abc123?download=1', 'abc123'),
('https://other.example.dev/-abc123', None),
('https://i.example.dev/files/-abc123', None),
('https://i.example.dev/-abc/def', None),
('https://i.example.dev/-', None),
('https://i.example.dev/', None),
])
def test_parse_share_link(url, expected):
assert parse_share_link(url, host='i.example.dev') == expected
def test_parse_share_link_honours_root_path():
assert parse_share_link('https://h/sub/-abc', host='h', root_path='/sub') == 'abc'
assert parse_share_link('https://h/-abc', host='h', root_path='/sub') is None
def test_parse_share_link_rejects_host_lookalike():
assert parse_share_link('https://i.example.dev.evil.tld/-abc', host='i.example.dev') is None
# --- build_segments -----------------------------------------------------------
FILES = {
'Q6jmHoRYjFi': {'file_name': 'barend.png', 'content_type': 'image/png'},
'eluTduc-npP': {'file_name': 'hanzoku.png', 'content_type': 'image/png'},
'kzmIlsf7uUe': {'file_name': 'gone.png', 'content_type': 'image/png', 'expired': True},
'to4wPonAadV': {'file_name': 'notes.txt', 'content_type': 'text/plain'},
}
def make_lookup(calls: list):
async def lookup(file_id):
calls.append(file_id)
return FILES.get(file_id)
return lookup
def run(coro):
return asyncio.run(coro)
def test_build_segments_resolves_own_links():
calls = []
segments = run(build_segments(
EXAMPLE, host='i.example.dev', root_path='', lookup=make_lookup(calls)
))
by_url = {s['url']: s for s in segments if s['kind'] in {'link', 'file'}}
assert by_url['https://i.example.dev/-Q6jmHoRYjFi'] == {
'kind': 'file', 'url': 'https://i.example.dev/-Q6jmHoRYjFi', 'state': 'ok',
'file_id': 'Q6jmHoRYjFi', 'label': 'barend.png', 'media': 'image',
}
assert by_url['https://i.example.dev/-to4wPonAadV']['media'] == 'other'
assert by_url['https://i.example.dev/-kzmIlsf7uUe']['kind'] == 'link'
assert by_url['https://i.example.dev/-kzmIlsf7uUe']['state'] == 'expired'
assert sorted(calls) == sorted(FILES)
def test_build_segments_keeps_foreign_links_unresolved():
calls = []
segments = run(build_segments(
'see https://other.example.dev/-Q6jmHoRYjFi',
host='i.example.dev', root_path='', lookup=make_lookup(calls),
))
assert segments[-1] == {'kind': 'link', 'url': 'https://other.example.dev/-Q6jmHoRYjFi'}
assert calls == []
def test_build_segments_marks_missing_files():
segments = run(build_segments(
'https://i.example.dev/-nope', host='i.example.dev', root_path='', lookup=make_lookup([])
))
assert segments[0]['kind'] == 'link'
assert segments[0]['state'] == 'missing'
def test_build_segments_deduplicates_lookups():
calls = []
text = 'https://i.example.dev/-Q6jmHoRYjFi and https://i.example.dev/-Q6jmHoRYjFi'
run(build_segments(text, host='i.example.dev', root_path='', lookup=make_lookup(calls)))
assert calls == ['Q6jmHoRYjFi']
def test_build_segments_caps_lookups():
calls = []
text = ' '.join(f'https://i.example.dev/-id{index}' for index in range(30))
segments = run(build_segments(
text, host='i.example.dev', root_path='', lookup=make_lookup(calls), max_lookups=5
))
assert len(calls) == 5
assert all(s['kind'] == 'link' for s in segments if s['kind'] != 'text')
def test_build_segments_without_lookup_only_linkifies():
segments = run(build_segments(EXAMPLE, host='i.example.dev', root_path='', lookup=None))
assert all(segment['kind'] in {'text', 'link'} for segment in segments)
# --- wants_text_view ----------------------------------------------------------
BROWSER_ACCEPT = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
@pytest.mark.parametrize('content_type,accept,raw,download,expected', [
('text/plain', BROWSER_ACCEPT, None, False, True),
('text/plain; charset=utf-8', BROWSER_ACCEPT, None, False, True),
('text/markdown', BROWSER_ACCEPT, None, False, True),
('text/plain', '*/*', None, False, False),
('text/plain', '', None, False, False),
('text/plain', BROWSER_ACCEPT, '1', False, False),
('text/plain', BROWSER_ACCEPT, 'true', False, False),
('text/plain', BROWSER_ACCEPT, None, True, False),
('image/png', BROWSER_ACCEPT, None, False, False),
('application/json', BROWSER_ACCEPT, None, False, False),
])
def test_wants_text_view(content_type, accept, raw, download, expected):
assert wants_text_view(
content_type=content_type, accept_header=accept, raw_param=raw, force_download=download
) is expected
# --- decode_text --------------------------------------------------------------
def test_decode_text_keeps_the_whole_document():
raw = b'x' * (512 * 1024)
assert len(decode_text(raw)) == len(raw)
def test_decode_text_survives_invalid_utf8():
assert decode_text(b'ok \xff\xfe').startswith('ok ')