simple-nanoshare/tests/test_serve_file_cache.py

import asyncio
import importlib.util
import sys
import time
import types
from io import BytesIO
from pathlib import Path
from unittest.mock import AsyncMock

import pytest
from quart import Blueprint, Quart

from my_modules.AsyncCache import AsyncCache
from my_modules.filecache.store import build_file_cache

ROOT = Path(__file__).resolve().parents[1]
IMAGE = b'\x89PNG\r\n\x1a\n' + b'pixels' * 100

def base_file(**overrides):
  data = {
    'file_name': 'barend.png',
    'content_type': 'image/png',
    'db_image_url': 'http://convex.internal/api/storage/abc',
    'uploaded_at': 1700000000000,
    'expires_at': '',
    'user_id': 'owner',
  }
  data.update(overrides)
  return data

def load_share_module(monkeypatch):
  stubs = {
    'my_modules.decoratory.header': {'login_required': lambda fn: fn},
    'my_modules.functions': {'get_ip': lambda: '192.0.2.1'},
    'my_modules.app.setup': {
      'LIMITER': types.SimpleNamespace(limit=lambda *args, **kwargs: lambda fn: fn),
    },
    'my_modules.app.logger': {'logger': None},
    'my_modules.expiry': {'parse_expires': lambda value: value},
    'quart_common.web.wide_event': {'add_wide_event_context': lambda **kwargs: None},
  }
  for name, values in stubs.items():
    module = types.ModuleType(name)
    module.__dict__.update(values)
    monkeypatch.setitem(sys.modules, name, module)

  path = ROOT / 'routes/side/main.py'
  spec = importlib.util.spec_from_file_location('serve_file_cache_under_test', path)
  module = importlib.util.module_from_spec(spec)
  spec.loader.exec_module(module)
  return module

def build_app(monkeypatch, file_data=None, with_cache=True):
  module = load_share_module(monkeypatch)
  app = Quart(
    'serve_file_cache_test',
    template_folder=str(ROOT / 'templates/side'),
    static_folder=str(ROOT / 'templates/static'),
  )
  app.secret_key = 'test-secret'
  record = base_file() if file_data is None else file_data

  async def get_file(file_id):
    return record if file_id == 'abc' else None

  async def get_from_storage(storage_id):
    return BytesIO(IMAGE)

  app.convex = types.SimpleNamespace(
    get_file=AsyncMock(side_effect=get_file),
    get_from_storage=AsyncMock(side_effect=get_from_storage),
    add_file_access=AsyncMock(return_value=None),
  )
  if with_cache:
    app.file_cache = build_file_cache(AsyncCache(backend='memory'))
  app.register_blueprint(module.side_main_bp)
  register_template_endpoints(app)
  return app

def register_template_endpoints(app):
  """Minimal endpoints so base.htm and navigation.htm can resolve url_for."""
  for name, rule, endpoint in (
    ('basic', '/favicon.ico', 'favicon'),
    ('auth_login', '/login', 'login'),
    ('link_composer', '/links', 'link_composer_page'),
  ):
    blueprint = Blueprint(name, __name__)
    blueprint.add_url_rule(rule, endpoint, lambda: '')
    app.register_blueprint(blueprint)

def get(app, path='/-abc', headers=None):
  async def run():
    return await app.test_client().get(path, headers=headers or {})
  return asyncio.run(run())

# --- conditional requests -----------------------------------------------------

def test_first_response_carries_validators(monkeypatch):
  app = build_app(monkeypatch)
  response = get(app)

  assert response.status_code == 200
  assert response.headers['ETag'].startswith('"')
  assert response.headers['Cache-Control'] == 'public, max-age=86400'
  assert 'Last-Modified' in response.headers

# --- Cloudflare edge TTL ------------------------------------------------------

def test_edge_header_is_shorter_than_the_browser_header(monkeypatch):
  app = build_app(monkeypatch)
  response = get(app)

  assert response.headers['Cache-Control'] == 'public, max-age=86400'
  assert response.headers['Cloudflare-CDN-Cache-Control'] == 'public, max-age=1800'

def test_plain_cdn_header_is_not_sent(monkeypatch):
  # Only the Cloudflare-specific header, so it is not proxied to the browser.
  app = build_app(monkeypatch)
  assert 'CDN-Cache-Control' not in get(app).headers

def test_edge_header_is_on_the_304_too(monkeypatch):
  app = build_app(monkeypatch)
  etag = get(app).headers['ETag']
  response = get(app, headers={'If-None-Match': etag})

  assert response.status_code == 304
  assert response.headers['Cloudflare-CDN-Cache-Control'] == 'public, max-age=1800'

def test_edge_ttl_is_configurable(monkeypatch):
  monkeypatch.setenv('FILE_CACHE_EDGE_MAX_AGE', '42')
  app = build_app(monkeypatch)
  assert get(app).headers['Cloudflare-CDN-Cache-Control'] == 'public, max-age=42'

def test_edge_caching_can_be_switched_off(monkeypatch):
  monkeypatch.setenv('FILE_CACHE_EDGE_MAX_AGE', '0')
  app = build_app(monkeypatch)
  response = get(app)

  assert response.headers['Cloudflare-CDN-Cache-Control'] == 'no-store'
  # The browser must keep caching even when the edge does not.
  assert response.headers['Cache-Control'] == 'public, max-age=86400'

def test_expired_file_is_never_edge_cached(monkeypatch):
  app = build_app(monkeypatch, file_data={'expired': True, 'user_id': 'owner'})
  response = get(app)

  assert response.status_code == 410
  assert response.headers['Cache-Control'] == 'no-store'

def test_text_view_is_never_edge_cached(monkeypatch):
  app = build_app(monkeypatch, file_data=base_file(
    file_name='note.txt', content_type='text/plain',
  ))
  response = get(app, headers={'Accept': 'text/html,application/xhtml+xml'})

  assert response.headers['Cache-Control'] == 'private, no-store'

def test_revalidation_returns_304_without_touching_storage(monkeypatch):
  app = build_app(monkeypatch)
  etag = get(app).headers['ETag']
  app.convex.get_from_storage.reset_mock()

  response = get(app, headers={'If-None-Match': etag})

  assert response.status_code == 304
  assert response.headers['ETag'] == etag
  app.convex.get_from_storage.assert_not_awaited()

def test_revalidation_by_date_returns_304(monkeypatch):
  app = build_app(monkeypatch)
  last_modified = get(app).headers['Last-Modified']
  app.convex.get_from_storage.reset_mock()

  response = get(app, headers={'If-Modified-Since': last_modified})

  assert response.status_code == 304
  app.convex.get_from_storage.assert_not_awaited()

def test_stale_etag_still_gets_the_body(monkeypatch):
  app = build_app(monkeypatch)
  response = get(app, headers={'If-None-Match': '"stale-value"'})

  assert response.status_code == 200
  assert asyncio.run(response.get_data()) == IMAGE

def test_download_variant_has_its_own_etag(monkeypatch):
  app = build_app(monkeypatch)
  inline = get(app).headers['ETag']
  attachment = get(app, '/-abc?download=1').headers['ETag']

  assert inline != attachment

def test_download_etag_does_not_satisfy_the_inline_request(monkeypatch):
  app = build_app(monkeypatch)
  attachment = get(app, '/-abc?download=1').headers['ETag']

  assert get(app, '/-abc', {'If-None-Match': attachment}).status_code == 200

# --- cache reuse --------------------------------------------------------------

def test_cold_clients_reuse_metadata_and_blob(monkeypatch):
  app = build_app(monkeypatch)
  for _ in range(5):
    assert get(app).status_code == 200

  assert app.convex.get_file.await_count == 1
  assert app.convex.get_from_storage.await_count == 1

def test_without_a_cache_every_request_hits_convex(monkeypatch):
  app = build_app(monkeypatch, with_cache=False)
  for _ in range(3):
    assert get(app).status_code == 200

  assert app.convex.get_file.await_count == 3
  assert app.convex.get_from_storage.await_count == 3

def test_cached_body_is_identical_on_repeat(monkeypatch):
  app = build_app(monkeypatch)
  first = asyncio.run(get(app).get_data())
  second = asyncio.run(get(app).get_data())

  assert first == second == IMAGE

def test_oversized_blobs_are_not_cached(monkeypatch):
  monkeypatch.setenv('FILE_CACHE_BLOB_ITEM_BYTES', '4')
  app = build_app(monkeypatch)
  get(app)
  get(app)

  assert app.convex.get_from_storage.await_count == 2
  # Metadata is small, so it stays cached even when the blob does not.
  assert app.convex.get_file.await_count == 1

# --- expiry and invalidation --------------------------------------------------

def test_expired_file_is_not_served_from_cache(monkeypatch):
  app = build_app(monkeypatch, file_data={'expired': True, 'user_id': 'owner'})
  for _ in range(3):
    assert get(app).status_code == 410

  # The expired marker must never be cached.
  assert app.convex.get_file.await_count == 3

def test_cache_control_is_clamped_for_a_file_that_expires_soon(monkeypatch):
  soon = (time.time() + 120) * 1000
  app = build_app(monkeypatch, file_data=base_file(expires_at=soon))
  value = get(app).headers['Cache-Control']

  assert 100 <= int(value.rsplit('=', 1)[1]) <= 120

def test_invalidation_makes_the_next_request_refetch_metadata(monkeypatch):
  app = build_app(monkeypatch)
  get(app)

  async def invalidate():
    async with app.app_context():
      from my_modules.filecache.access import drop_file_cache
      await drop_file_cache('abc')

  asyncio.run(invalidate())
  get(app)

  assert app.convex.get_file.await_count == 2

def test_missing_file_is_never_cached(monkeypatch):
  app = build_app(monkeypatch)
  for _ in range(3):
    assert get(app, '/-unknown').status_code == 404

  assert app.convex.get_file.await_count == 3

# --- access logging -----------------------------------------------------------

@pytest.mark.parametrize('headers,expected_status', [
  ({}, 200),
  ({'If-None-Match': 'PLACEHOLDER'}, 304),
])
def test_access_is_logged_for_both_fresh_and_revalidated_requests(monkeypatch, headers, expected_status):
  app = build_app(monkeypatch)
  if 'If-None-Match' in headers:
    headers['If-None-Match'] = get(app).headers['ETag']
    app.convex.add_file_access.reset_mock()

  response = get(app, headers=headers)

  assert response.status_code == expected_status
  app.convex.add_file_access.assert_awaited_once()