loreal-utilisation-dept/backend/tests/test_airtable_cache.py

"""Tests for the TTL async cache + Airtable cache wiring."""

from __future__ import annotations

import asyncio
from datetime import date

import pytest

from app.services.cache import TTLAsyncCache

# NOTE: don't import app.services.airtable_fetch at module level — it pulls
# in app.config at collection time, which can race with the conftest
# session-autouse env-seeding fixture and leave app.auth.local holding a
# stale `settings` object (empty ADMIN_PASSWORD_BCRYPT) for the rest of
# the suite. Importing inside each test that needs it is safe because by
# then the autouse fixture has run and the env vars are populated.


@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)


def test_bookings_filter_combines_date_and_multivalue_filters():
    """Date overlap + multi-value department + name with an apostrophe
    must all combine into a single AND() with properly-escaped quotes."""
    from app.services.airtable_fetch import _bookings_filter

    formula = _bookings_filter(
        date(2026, 5, 4),
        date(2026, 5, 17),
        department="Creative Team,Operation Team",
        name="O'Brien,Bhakti Doshi",
    )
    assert formula is not None
    # Date overlap clauses
    assert "IS_BEFORE({Start Date}, '2026-05-17')" in formula
    assert "IS_AFTER({End Date}, '2026-05-04')" in formula
    # Department OR
    assert (
        "OR({Department (from Resource Name)}='Creative Team', "
        "{Department (from Resource Name)}='Operation Team')"
    ) in formula
    # Name OR with the apostrophe escaped
    assert "{Resource Name (from Resource)}='O\\'Brien'" in formula
    assert "{Resource Name (from Resource)}='Bhakti Doshi'" in formula
    # Wrapped in AND(...) since we have multiple top-level clauses.
    assert formula.startswith("AND(") and formula.endswith(")")


def test_bookings_filter_no_filters_returns_none():
    from app.services.airtable_fetch import _bookings_filter

    assert _bookings_filter(None, None, None, None) is None


def test_bookings_filter_single_department_no_or_wrapper():
    """A single value shouldn't get wrapped in OR(...)."""
    from app.services.airtable_fetch import _bookings_filter

    formula = _bookings_filter(None, None, "Creative Team", None)
    assert formula == "{Department (from Resource Name)}='Creative Team'"


@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