room_cal/tests/test_utils.py

import pytest
from datetime import date, datetime, timedelta
from pathlib import Path
from room_cal._utils import safe_json_loads, iter_dates, event_dates, filter_events, read_env
from room_cal.store._backend import EventRecord


# --- safe_json_loads ---

def test_safe_json_loads_none():
  assert safe_json_loads(None) is None


def test_safe_json_loads_empty_string():
  assert safe_json_loads('') is None


def test_safe_json_loads_invalid():
  assert safe_json_loads('not json') is None


def test_safe_json_loads_valid_dict():
  assert safe_json_loads('{"k": 1}') == {'k': 1}


def test_safe_json_loads_valid_list():
  assert safe_json_loads('[1, 2]') == [1, 2]


def test_safe_json_loads_nested():
  assert safe_json_loads('{"a": {"b": true}}') == {'a': {'b': True}}


# --- iter_dates ---

def test_iter_dates_single_day():
  d = date(2025, 1, 15)
  assert list(iter_dates(d, d)) == [d]


def test_iter_dates_range():
  start = date(2025, 3, 1)
  end = date(2025, 3, 5)
  result = list(iter_dates(start, end))
  assert result == [date(2025, 3, d) for d in range(1, 6)]


def test_iter_dates_end_before_start():
  start = date(2025, 5, 10)
  end = date(2025, 5, 9)
  assert list(iter_dates(start, end)) == []


def test_iter_dates_consecutive():
  start = date(2025, 1, 31)
  end = date(2025, 2, 2)
  result = list(iter_dates(start, end))
  assert result == [date(2025, 1, 31), date(2025, 2, 1), date(2025, 2, 2)]


# --- event_dates ---

def _allday(start: date, end: date) -> EventRecord:
  """All-day event; end is exclusive (CalDAV convention)."""
  return EventRecord(
    uid='test-uid',
    start=datetime(start.year, start.month, start.day),
    end=datetime(end.year, end.month, end.day),
    all_day=True,
    title='',
    description='',
    x_data=None,
  )


def _timed(start: datetime, end: datetime) -> EventRecord:
  return EventRecord(
    uid='test-uid',
    start=start,
    end=end,
    all_day=False,
    title='',
    description='',
    x_data=None,
  )


def test_event_dates_all_day_single():
  d = date(2025, 6, 10)
  ev = _allday(d, d + timedelta(days=1))
  assert list(event_dates(ev, d, d)) == [d]


def test_event_dates_all_day_span():
  start = date(2025, 6, 10)
  end_ex = date(2025, 6, 14)  # exclusive DTEND
  ev = _allday(start, end_ex)
  window_start = date(2025, 6, 9)
  window_end = date(2025, 6, 15)
  result = list(event_dates(ev, window_start, window_end))
  assert result == [date(2025, 6, d) for d in range(10, 14)]


def test_event_dates_all_day_clamped_by_window():
  start = date(2025, 6, 10)
  end_ex = date(2025, 6, 14)
  ev = _allday(start, end_ex)
  window_start = date(2025, 6, 11)
  window_end = date(2025, 6, 12)
  result = list(event_dates(ev, window_start, window_end))
  assert result == [date(2025, 6, 11), date(2025, 6, 12)]


def test_event_dates_all_day_no_overlap():
  d = date(2025, 6, 10)
  ev = _allday(d, d + timedelta(days=1))
  result = list(event_dates(ev, date(2025, 6, 11), date(2025, 6, 15)))
  assert result == []


def test_event_dates_timed_event():
  dt = datetime(2025, 6, 10, 14, 0)
  ev = _timed(dt, dt + timedelta(hours=2))
  result = list(event_dates(ev, date(2025, 6, 10), date(2025, 6, 10)))
  assert result == [date(2025, 6, 10)]


# --- filter_events ---

def _ev(title: str, *, all_day=True, x_data='{}') -> EventRecord:
  d = datetime(2025, 1, 1)
  return EventRecord(
    uid='test-uid',
    start=d,
    end=d + timedelta(days=1),
    all_day=all_day,
    title=title,
    description='',
    x_data=x_data,
  )


def test_filter_events_prefix_match():
  events = [_ev('Price: $100'), _ev('Other: thing'), _ev('Price: $200')]
  result = filter_events(events, 'Price: ')
  assert len(result) == 2
  assert all(e.title.startswith('Price: ') for e in result)


def test_filter_events_no_match():
  events = [_ev('Closed: All'), _ev('Discount: 10%')]
  assert filter_events(events, 'Price: ') == []


def test_filter_events_all_day_true():
  events = [_ev('Price: $100', all_day=True), _ev('Price: $200', all_day=False)]
  result = filter_events(events, 'Price: ', all_day=True)
  assert len(result) == 1
  assert result[0].all_day is True


def test_filter_events_all_day_false():
  events = [_ev('Price: $100', all_day=True), _ev('Price: $200', all_day=False)]
  result = filter_events(events, 'Price: ', all_day=False)
  assert len(result) == 1
  assert result[0].all_day is False


def test_filter_events_all_day_none_returns_both():
  events = [_ev('Price: $100', all_day=True), _ev('Price: $200', all_day=False)]
  assert len(filter_events(events, 'Price: ', all_day=None)) == 2


def test_filter_events_require_x_data_excludes_none():
  events = [_ev('Price: $100', x_data='{}'), _ev('Price: $200', x_data=None)]
  result = filter_events(events, 'Price: ', require_x_data=True)
  assert len(result) == 1


def test_filter_events_require_x_data_false_includes_none():
  events = [_ev('Price: $100', x_data='{}'), _ev('Price: $200', x_data=None)]
  result = filter_events(events, 'Price: ', require_x_data=False)
  assert len(result) == 2


# --- read_env ---

def test_read_env_basic(tmp_path):
  env_file = tmp_path / 'test.env'
  env_file.write_text('FOO=bar\nBAZ=qux\n')
  result = read_env(env_file)
  assert result == {'FOO': 'bar', 'BAZ': 'qux'}


def test_read_env_strips_quotes(tmp_path):
  env_file = tmp_path / 'test.env'
  env_file.write_text("A='hello'\nB=\"world\"\n")
  result = read_env(env_file)
  assert result == {'A': 'hello', 'B': 'world'}


def test_read_env_skips_comments_and_blanks(tmp_path):
  env_file = tmp_path / 'test.env'
  env_file.write_text('# comment\n\nFOO=bar\n')
  result = read_env(env_file)
  assert result == {'FOO': 'bar'}


def test_read_env_missing_file(tmp_path):
  result = read_env(tmp_path / 'nonexistent.env')
  assert result == {}


def test_read_env_value_with_equals(tmp_path):
  env_file = tmp_path / 'test.env'
  env_file.write_text('TOKEN=abc=def=ghi\n')
  result = read_env(env_file)
  assert result == {'TOKEN': 'abc=def=ghi'}