04edbfdd2c
Replaces a static SPA that shipped an Airtable PAT in the JS bundle.
The new architecture holds all secrets server-side, fronts the app
behind Apache on optical-dev with the shared-vhost split-build pattern,
and is designed for a later Azure AD/MSAL swap-in.
- backend/ FastAPI + uvicorn, local auth (Azure AD stub), Airtable
proxy with TTL cache, Zoho .xlsx/.csv parser, merge
service for utilisation summaries. 28 pytest tests.
- frontend/ React + Vite + TS + Tailwind + Recharts SPA. Login entry
chunk 12.83 KB gzipped; Recharts lazy-loaded. No tokens
or Airtable URLs in the built bundle.
- deploy/ Idempotent deploy.sh (port auto-pick 8200-8299,
.env-persisted) + split-build Apache include template.
- docker-compose.yml pins name: utilisation-dept and binds 127.0.0.1.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
78 lines
2 KiB
Python
78 lines
2 KiB
Python
"""Tests for the TTL async cache + Airtable cache wiring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.services.cache import TTLAsyncCache
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_second_call_hits_cache():
|
|
cache = TTLAsyncCache(ttl=60)
|
|
counter = {"n": 0}
|
|
|
|
async def loader():
|
|
counter["n"] += 1
|
|
return {"v": counter["n"]}
|
|
|
|
a = await cache.get_or_set("k", loader)
|
|
b = await cache.get_or_set("k", loader)
|
|
assert a == b == {"v": 1}
|
|
assert counter["n"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalidate_forces_reload():
|
|
cache = TTLAsyncCache(ttl=60)
|
|
counter = {"n": 0}
|
|
|
|
async def loader():
|
|
counter["n"] += 1
|
|
return counter["n"]
|
|
|
|
assert await cache.get_or_set("k", loader) == 1
|
|
cache.invalidate("k")
|
|
assert await cache.get_or_set("k", loader) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_prevents_thundering_herd():
|
|
cache = TTLAsyncCache(ttl=60)
|
|
counter = {"n": 0}
|
|
|
|
async def slow_loader():
|
|
# Yield once so other coroutines have a chance to enter.
|
|
counter["n"] += 1
|
|
await asyncio.sleep(0.05)
|
|
return counter["n"]
|
|
|
|
results = await asyncio.gather(
|
|
cache.get_or_set("k", slow_loader),
|
|
cache.get_or_set("k", slow_loader),
|
|
cache.get_or_set("k", slow_loader),
|
|
cache.get_or_set("k", slow_loader),
|
|
cache.get_or_set("k", slow_loader),
|
|
)
|
|
# All callers see the same cached value, loader ran only once.
|
|
assert counter["n"] == 1
|
|
assert all(r == 1 for r in results)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_bypasses_cache():
|
|
"""Simulate the 'refresh=true' behaviour the router uses."""
|
|
cache = TTLAsyncCache(ttl=60)
|
|
counter = {"n": 0}
|
|
|
|
async def loader():
|
|
counter["n"] += 1
|
|
return counter["n"]
|
|
|
|
await cache.get_or_set("k", loader)
|
|
# Router invalidates, then re-fetches.
|
|
cache.invalidate("k")
|
|
await cache.get_or_set("k", loader)
|
|
assert counter["n"] == 2
|