room_cal/tests/test_apply_rules.py

import pytest
from datetime import date, timedelta

from room_cal._utils import get_config, iter_dates, local_today, services as _services
from room_cal.closed import ClosedData
import room_cal.rules as _rules
from room_cal.rules.closed.future_block import RULE_NAME, run as future_block_run


# ── future_block (pure) ────────────────────────────────────────────────────────


def _days_out() -> int:
  return get_config().rules.future_block.days_out


@pytest.mark.asyncio
async def test_future_block_rule_registered():
  assert RULE_NAME == 'future_block'
  assert 'future_block' in _rules.list_rules('closed')
  assert 'future_block' in _rules.list_pipelines(type_='closed')


@pytest.mark.asyncio
async def test_future_block_beyond_threshold_all_closed():
  """Every returned date strictly beyond today + days_out should have closed=True.
  Dates already blocked by a reservation or Closed event are omitted."""
  days_out = _days_out()
  start = local_today() + timedelta(days=days_out + 1)
  end = start + timedelta(days=4)
  result = await future_block_run(start, end, {})
  assert set(result).issubset(set(iter_dates(start, end)))
  assert all(bool(rec.services) for rec in result.values())


@pytest.mark.asyncio
async def test_future_block_within_threshold_all_open():
  """Every date on or before today + days_out should have closed=False."""
  days_out = _days_out()
  start = local_today() + timedelta(days=1)
  end = local_today() + timedelta(days=min(days_out - 1, 5))
  result = await future_block_run(start, end, {})
  assert all(not rec.services for rec in result.values())


@pytest.mark.asyncio
async def test_future_block_threshold_boundary():
  """The threshold date itself is not closed; the day after is.
  Dates already blocked on the calendar are omitted from results."""
  days_out = _days_out()
  threshold = local_today() + timedelta(days=days_out)
  result = await future_block_run(threshold - timedelta(days=1), threshold + timedelta(days=1), {})
  if threshold in result:
    assert not result[threshold].services
  if threshold + timedelta(days=1) in result:
    assert bool(result[threshold + timedelta(days=1)].services)


@pytest.mark.asyncio
async def test_future_block_covers_full_range():
  """run() returns at most one record per date, in range, with no duplicates."""
  days_out = _days_out()
  start = local_today() + timedelta(days=days_out - 2)
  end = start + timedelta(days=5)
  result = await future_block_run(start, end, {})
  assert set(result).issubset(set(iter_dates(start, end)))
  assert len(result) == len(set(result))


# ── apply_rules — future_block (live CalDAV) ───────────────────────────────────


@pytest.mark.asyncio
async def test_apply_rules_future_block_creates_closed_events():
  """apply_rules with future_block writes Closed events for dates beyond the threshold."""
  from room_cal.actions.apply_rules import apply_rules
  from room_cal.closed import Closed

  closed_cal = Closed()
  days_out = _days_out()
  start = local_today() + timedelta(days=days_out + 10)
  end = start + timedelta(days=2)

  try:
    result = await apply_rules(start, end, pipeline_name='future_block')

    assert 'future_block' in result
    assert set(result['future_block']).issubset(set(iter_dates(start, end)))

    for d in iter_dates(start, end):
      svcs = await closed_cal.get(d)
      assert svcs is not None and svcs.services, f"Expected Closed event on {d}"

    svc_keys = sorted(s.key for s in _services())
    for d in iter_dates(start, end):
      assert sorted((await closed_cal.get(d)).services) == svc_keys

  finally:
    for d in iter_dates(start, end):
      await closed_cal.set(d, ClosedData(services=[]))


@pytest.mark.asyncio
async def test_apply_rules_future_block_removes_stale_closed_events():
  """Rule-sourced Closed events on dates now inside the threshold are removed on re-run."""
  from room_cal.actions.apply_rules import apply_rules
  from room_cal.closed import Closed

  closed_cal = Closed()
  svc_keys = [s.key for s in _services()]

  near_date = local_today() + timedelta(days=5)
  source = {'domain': 'rules', 'name': 'Rule =>', 'id': 'future_block'}
  await closed_cal.set(near_date, ClosedData(services=svc_keys, source=source))

  try:
    assert await closed_cal.get(near_date) is not None

    result = await apply_rules(near_date, near_date, pipeline_name='future_block')

    assert result['future_block'] == []
    assert await closed_cal.get(near_date) is None, f"Stale Closed event not removed on {near_date}"

  finally:
    await closed_cal.set(near_date, ClosedData(services=[]))


@pytest.mark.asyncio
async def test_apply_rules_future_block_idempotent():
  """Running apply_rules twice on the same beyond-threshold range is idempotent."""
  from room_cal.actions.apply_rules import apply_rules
  from room_cal.closed import Closed

  closed_cal = Closed()
  days_out = _days_out()
  start = local_today() + timedelta(days=days_out + 20)
  end = start + timedelta(days=1)

  try:
    result1 = await apply_rules(start, end, pipeline_name='future_block')
    result2 = await apply_rules(start, end, pipeline_name='future_block')

    assert result1 == result2

    for d in iter_dates(start, end):
      svcs = await closed_cal.get(d)
      assert svcs is not None and svcs.services

  finally:
    for d in iter_dates(start, end):
      await closed_cal.set(d, ClosedData(services=[]))