|
|
@@ -0,0 +1,617 @@
|
|
|
+"""
|
|
|
+SQLite backend tests for all StoreBacked stores.
|
|
|
+
|
|
|
+Each test injects a SQLiteBackend directly via backend= so no CalDAV or config
|
|
|
+is involved. A fresh temp DB is created per test via the sqlite_backend fixture.
|
|
|
+"""
|
|
|
+import json
|
|
|
+import pytest
|
|
|
+from datetime import date, timedelta
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+from room_cal.store._sqlite import SQLiteBackend
|
|
|
+from room_cal.closed import Closed, ClosedData, _build_title, _build_description as _closed_desc
|
|
|
+from room_cal.discount import Discount, DiscountData, DiscountType
|
|
|
+from room_cal.reservations import Reservation
|
|
|
+from room_cal.user_cal import UserCal
|
|
|
+from room_cal.pricing.price_calendar import CalPrice
|
|
|
+
|
|
|
+
|
|
|
+# ── fixtures ───────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def sqlite_backend(tmp_path):
|
|
|
+ return SQLiteBackend(tmp_path / 'test.db', 'Room')
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def house_backend(tmp_path):
|
|
|
+ return SQLiteBackend(tmp_path / 'test.db', 'House')
|
|
|
+
|
|
|
+
|
|
|
+# ── SQLiteBackend direct ───────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_backend_create_fetch_delete(sqlite_backend):
|
|
|
+ from room_cal.store._backend import EventRecord
|
|
|
+ from datetime import datetime
|
|
|
+
|
|
|
+ rec = EventRecord(
|
|
|
+ uid='u1',
|
|
|
+ start=datetime(2024, 6, 1),
|
|
|
+ end=datetime(2024, 6, 2),
|
|
|
+ all_day=True,
|
|
|
+ title='Test Event',
|
|
|
+ description='desc',
|
|
|
+ x_data='{"k": 1}',
|
|
|
+ )
|
|
|
+ await sqlite_backend.create(rec)
|
|
|
+ results = await sqlite_backend.fetch_range(datetime(2024, 5, 31), datetime(2024, 6, 3))
|
|
|
+ assert len(results) == 1
|
|
|
+ assert results[0].uid == 'u1'
|
|
|
+ assert results[0].title == 'Test Event'
|
|
|
+ assert results[0].x_data == '{"k": 1}'
|
|
|
+
|
|
|
+ await sqlite_backend.delete('u1')
|
|
|
+ results = await sqlite_backend.fetch_range(datetime(2024, 5, 31), datetime(2024, 6, 3))
|
|
|
+ assert results == []
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_backend_update(sqlite_backend):
|
|
|
+ from room_cal.store._backend import EventRecord
|
|
|
+ from datetime import datetime
|
|
|
+
|
|
|
+ rec = EventRecord(
|
|
|
+ uid='u2',
|
|
|
+ start=datetime(2024, 7, 10),
|
|
|
+ end=datetime(2024, 7, 11),
|
|
|
+ all_day=True,
|
|
|
+ title='Original',
|
|
|
+ description='',
|
|
|
+ x_data=None,
|
|
|
+ )
|
|
|
+ await sqlite_backend.create(rec)
|
|
|
+ updated = await sqlite_backend.update('u2', title='Updated', x_data='{"v": 2}')
|
|
|
+ assert updated.title == 'Updated'
|
|
|
+ assert updated.x_data == '{"v": 2}'
|
|
|
+
|
|
|
+ results = await sqlite_backend.fetch_range(datetime(2024, 7, 9), datetime(2024, 7, 12))
|
|
|
+ assert results[0].title == 'Updated'
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_backend_fetch_range_boundary(sqlite_backend):
|
|
|
+ """fetch_range uses exclusive end — event touching the boundary is included."""
|
|
|
+ from room_cal.store._backend import EventRecord
|
|
|
+ from datetime import datetime
|
|
|
+
|
|
|
+ rec = EventRecord(
|
|
|
+ uid='u3',
|
|
|
+ start=datetime(2024, 8, 1),
|
|
|
+ end=datetime(2024, 8, 2),
|
|
|
+ all_day=True,
|
|
|
+ title='Boundary',
|
|
|
+ description='',
|
|
|
+ x_data=None,
|
|
|
+ )
|
|
|
+ await sqlite_backend.create(rec)
|
|
|
+
|
|
|
+ assert await sqlite_backend.fetch_range(datetime(2024, 8, 1), datetime(2024, 8, 2)) != []
|
|
|
+ assert await sqlite_backend.fetch_range(datetime(2024, 8, 2), datetime(2024, 8, 3)) == []
|
|
|
+ assert await sqlite_backend.fetch_range(datetime(2024, 7, 31), datetime(2024, 8, 1, 0, 0, 1)) != []
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_backend_multiple_calendars_isolated(tmp_path):
|
|
|
+ """Two backends sharing the same DB file but different calendar_names don't see each other's events."""
|
|
|
+ from room_cal.store._backend import EventRecord
|
|
|
+ from datetime import datetime
|
|
|
+
|
|
|
+ db = tmp_path / 'shared.db'
|
|
|
+ room = SQLiteBackend(db, 'Room')
|
|
|
+ house = SQLiteBackend(db, 'House')
|
|
|
+
|
|
|
+ rec = EventRecord(
|
|
|
+ uid='r1', start=datetime(2024, 9, 1), end=datetime(2024, 9, 2),
|
|
|
+ all_day=True, title='Room Event', description='', x_data=None,
|
|
|
+ )
|
|
|
+ await room.create(rec)
|
|
|
+
|
|
|
+ room_results = await room.fetch_range(datetime(2024, 8, 31), datetime(2024, 9, 3))
|
|
|
+ house_results = await house.fetch_range(datetime(2024, 8, 31), datetime(2024, 9, 3))
|
|
|
+ assert len(room_results) == 1
|
|
|
+ assert house_results == []
|
|
|
+
|
|
|
+
|
|
|
+# ── Closed ─────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_set_and_get(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ target = date(2024, 3, 15)
|
|
|
+ services = ['airbnb', 'booking']
|
|
|
+
|
|
|
+ await closed.set(target, ClosedData(services=services, source={'domain': 'test', 'name': 'Test', 'id': 'c1'}))
|
|
|
+
|
|
|
+ stored = await closed.get(target)
|
|
|
+ assert stored is not None
|
|
|
+ assert sorted(stored.services) == sorted(services)
|
|
|
+ assert stored.source == {'domain': 'test', 'name': 'Test', 'id': 'c1'}
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_event_title_and_description(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ target = date(2024, 3, 16)
|
|
|
+
|
|
|
+ await closed.set(target, ClosedData(services=['airbnb', 'booking'], source={'domain': 'test', 'name': 'Test => ', 'id': 'c2'}))
|
|
|
+
|
|
|
+ event = await closed._find(target)
|
|
|
+ assert event is not None
|
|
|
+ assert event.title == 'Closed: All'
|
|
|
+ assert event.description == 'AirBnB: Closed\nBooking.com: Closed\nSource: Test => c2'
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_delete_via_empty_services(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ target = date(2024, 3, 17)
|
|
|
+
|
|
|
+ await closed.set(target, ClosedData(services=['airbnb']))
|
|
|
+ assert await closed.get(target) is not None
|
|
|
+
|
|
|
+ await closed.set(target, ClosedData(services=[]))
|
|
|
+ assert await closed.get(target) is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_set_is_idempotent(sqlite_backend):
|
|
|
+ """Calling set twice with the same data produces one event, not two."""
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ target = date(2024, 3, 18)
|
|
|
+
|
|
|
+ data = ClosedData(services=['airbnb'], source={'domain': 'test', 'id': 'c3'})
|
|
|
+ await closed.set(target, data)
|
|
|
+ await closed.set(target, data)
|
|
|
+
|
|
|
+ events = await closed.list(target, target)
|
|
|
+ assert len(events) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_set_updates_on_change(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ target = date(2024, 3, 19)
|
|
|
+
|
|
|
+ await closed.set(target, ClosedData(services=['airbnb']))
|
|
|
+ await closed.set(target, ClosedData(services=['airbnb', 'booking']))
|
|
|
+
|
|
|
+ stored = await closed.get(target)
|
|
|
+ assert sorted(stored.services) == ['airbnb', 'booking']
|
|
|
+ assert len(await closed.list(target, target)) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_list_range(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+
|
|
|
+ for day in range(1, 4):
|
|
|
+ await closed.set(date(2024, 4, day), ClosedData(services=['airbnb']))
|
|
|
+
|
|
|
+ events = await closed.list(date(2024, 4, 1), date(2024, 4, 3))
|
|
|
+ assert len(events) == 3
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_closed_get_range(sqlite_backend):
|
|
|
+ closed = Closed(backend=sqlite_backend)
|
|
|
+ d1, d2 = date(2024, 4, 10), date(2024, 4, 11)
|
|
|
+
|
|
|
+ await closed.set(d1, ClosedData(services=['airbnb']))
|
|
|
+ await closed.set(d2, ClosedData(services=['booking']))
|
|
|
+
|
|
|
+ result = await closed.get_range(d1, d2)
|
|
|
+ assert d1 in result and d2 in result
|
|
|
+ assert result[d1].services == ['airbnb']
|
|
|
+ assert result[d2].services == ['booking']
|
|
|
+
|
|
|
+
|
|
|
+# ── Discount ───────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_flat_round_trip(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 1)
|
|
|
+ source = {'domain': 'test', 'name': 'Test => ', 'id': 'flat-1'}
|
|
|
+
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.FLAT, value=21, source=source))
|
|
|
+
|
|
|
+ stored = await discount.get(target)
|
|
|
+ assert stored is not None
|
|
|
+ assert stored.type == DiscountType.FLAT
|
|
|
+ assert stored.value == 21
|
|
|
+ assert stored.source == source
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_percent_round_trip(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 2)
|
|
|
+ source = {'domain': 'test', 'name': 'Test => ', 'id': 'pct-1'}
|
|
|
+
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.PERCENT, value=15, source=source))
|
|
|
+
|
|
|
+ stored = await discount.get(target)
|
|
|
+ assert stored is not None
|
|
|
+ assert stored.type == DiscountType.PERCENT
|
|
|
+ assert stored.value == 15
|
|
|
+
|
|
|
+ event = await discount._find(target)
|
|
|
+ assert event.title == 'Discount: 15%'
|
|
|
+ assert 'Source: Test => pct-1' in event.description
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_flat_event_title(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 3)
|
|
|
+
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.FLAT, value=20, source={}))
|
|
|
+
|
|
|
+ event = await discount._find(target)
|
|
|
+ assert event.title == 'Discount: $20'
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_delete_on(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 4)
|
|
|
+
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.FLAT, value=10, source={}))
|
|
|
+ assert await discount.get(target) is not None
|
|
|
+
|
|
|
+ await discount.delete_on(target)
|
|
|
+ assert await discount.get(target) is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_set_is_idempotent(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 5)
|
|
|
+ data = DiscountData(type=DiscountType.FLAT, value=5, source={})
|
|
|
+
|
|
|
+ await discount.set(target, data)
|
|
|
+ await discount.set(target, data)
|
|
|
+
|
|
|
+ events = await discount.list(target, target)
|
|
|
+ assert len(events) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_set_updates_value(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ target = date(2024, 5, 6)
|
|
|
+
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.FLAT, value=10, source={}))
|
|
|
+ await discount.set(target, DiscountData(type=DiscountType.FLAT, value=20, source={}))
|
|
|
+
|
|
|
+ stored = await discount.get(target)
|
|
|
+ assert stored.value == 20
|
|
|
+ assert len(await discount.list(target, target)) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_discount_get_range(sqlite_backend):
|
|
|
+ discount = Discount(backend=sqlite_backend)
|
|
|
+ d1, d2 = date(2024, 5, 10), date(2024, 5, 11)
|
|
|
+
|
|
|
+ await discount.set(d1, DiscountData(type=DiscountType.FLAT, value=5, source={}))
|
|
|
+ await discount.set(d2, DiscountData(type=DiscountType.PERCENT, value=10, source={}))
|
|
|
+
|
|
|
+ result = await discount.get_range(d1, d2)
|
|
|
+ assert result[d1].type == DiscountType.FLAT
|
|
|
+ assert result[d2].type == DiscountType.PERCENT
|
|
|
+
|
|
|
+
|
|
|
+# ── Reservation ────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_set_and_get(sqlite_backend):
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ checkin = date(2024, 6, 1)
|
|
|
+ checkout = checkin + timedelta(days=2)
|
|
|
+
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': 'Alice Test',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': 'RES-001',
|
|
|
+ 'checkin_date': checkin,
|
|
|
+ 'checkout_date': checkout,
|
|
|
+ })
|
|
|
+
|
|
|
+ stored = await res.get('RES-001', checkin, checkout)
|
|
|
+ assert stored is not None
|
|
|
+ assert stored.guest_name == 'Alice Test'
|
|
|
+ assert stored.reservation_id == 'RES-001'
|
|
|
+ assert stored.service == 'airbnb'
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_generates_wifi_and_door(sqlite_backend):
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ checkin = date(2024, 6, 3)
|
|
|
+ checkout = checkin + timedelta(days=1)
|
|
|
+
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': 'Bob Test',
|
|
|
+ 'service': 'booking',
|
|
|
+ 'reservation_id': 'RES-002',
|
|
|
+ 'checkin_date': checkin,
|
|
|
+ 'checkout_date': checkout,
|
|
|
+ })
|
|
|
+
|
|
|
+ stored = await res.get('RES-002', checkin, checkout)
|
|
|
+ assert stored.wifi_key is not None and stored.wifi_key != ''
|
|
|
+ assert stored.door_code is not None and stored.door_code != ''
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_preserves_wifi_on_update(sqlite_backend):
|
|
|
+ """Re-setting a reservation preserves wifi_key/door_code from the first write."""
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ checkin = date(2024, 6, 5)
|
|
|
+ checkout = checkin + timedelta(days=2)
|
|
|
+ data = {
|
|
|
+ 'guest_name': 'Carol Test',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': 'RES-003',
|
|
|
+ 'checkin_date': checkin,
|
|
|
+ 'checkout_date': checkout,
|
|
|
+ }
|
|
|
+
|
|
|
+ await res.set(data)
|
|
|
+ first = await res.get('RES-003', checkin, checkout)
|
|
|
+
|
|
|
+ await res.set(data)
|
|
|
+ second = await res.get('RES-003', checkin, checkout)
|
|
|
+
|
|
|
+ assert first.wifi_key == second.wifi_key
|
|
|
+ assert first.door_code == second.door_code
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_list(sqlite_backend):
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ checkin = date(2024, 6, 10)
|
|
|
+
|
|
|
+ for i in range(3):
|
|
|
+ c_in = checkin + timedelta(days=i * 5)
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': f'Guest {i}',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': f'RES-L{i}',
|
|
|
+ 'checkin_date': c_in,
|
|
|
+ 'checkout_date': c_in + timedelta(days=2),
|
|
|
+ })
|
|
|
+
|
|
|
+ events = await res.list(checkin, checkin + timedelta(days=30))
|
|
|
+ assert len(events) == 3
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_delete(sqlite_backend):
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ checkin = date(2024, 6, 20)
|
|
|
+ checkout = checkin + timedelta(days=1)
|
|
|
+
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': 'Dave Test',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': 'RES-DEL',
|
|
|
+ 'checkin_date': checkin,
|
|
|
+ 'checkout_date': checkout,
|
|
|
+ })
|
|
|
+
|
|
|
+ event = await res._find('RES-DEL', checkin, checkout)
|
|
|
+ assert event is not None
|
|
|
+ await res.delete(event)
|
|
|
+ assert await res.get('RES-DEL', checkin, checkout) is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reservation_dates_shift_within_same_range(sqlite_backend):
|
|
|
+ """Shifting dates within the same search range triggers the delete+recreate path."""
|
|
|
+ res = Reservation(backend=sqlite_backend)
|
|
|
+ # Both checkin/checkout fall in the same week so _find can see the old event
|
|
|
+ checkin1, checkout1 = date(2024, 7, 1), date(2024, 7, 3)
|
|
|
+ checkin2, checkout2 = date(2024, 7, 2), date(2024, 7, 4)
|
|
|
+
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': 'Eve Test',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': 'RES-SHIFT',
|
|
|
+ 'checkin_date': checkin1,
|
|
|
+ 'checkout_date': checkout1,
|
|
|
+ })
|
|
|
+
|
|
|
+ # Second set uses new dates but _find scans new range — old event at [1,3) is in [2,4) range
|
|
|
+ await res.set({
|
|
|
+ 'guest_name': 'Eve Test',
|
|
|
+ 'service': 'airbnb',
|
|
|
+ 'reservation_id': 'RES-SHIFT',
|
|
|
+ 'checkin_date': checkin2,
|
|
|
+ 'checkout_date': checkout2,
|
|
|
+ })
|
|
|
+
|
|
|
+ stored = await res.get('RES-SHIFT', checkin2, checkout2)
|
|
|
+ assert stored is not None
|
|
|
+ assert stored.reservation_id == 'RES-SHIFT'
|
|
|
+
|
|
|
+
|
|
|
+# ── UserCal ────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_usercal_set_and_find_services(house_backend):
|
|
|
+ ucal = UserCal(backend=house_backend)
|
|
|
+ service_key = 'airbnb'
|
|
|
+ reservation_id = 'UC-001'
|
|
|
+ checkin = date(2024, 8, 1)
|
|
|
+ checkout = checkin + timedelta(days=3)
|
|
|
+
|
|
|
+ await ucal.set(service_key, reservation_id, checkin, checkout)
|
|
|
+
|
|
|
+ events = await ucal.find_services(checkin, checkout)
|
|
|
+ match = next((e for e in events if json.loads(e.x_data).get('reservation_id') == reservation_id), None)
|
|
|
+ assert match is not None
|
|
|
+ assert json.loads(match.x_data)['service_key'] == service_key
|
|
|
+ assert match.all_day
|
|
|
+ assert match.start.date() == checkin
|
|
|
+ assert match.end.date() == checkout
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_usercal_set_is_idempotent(house_backend):
|
|
|
+ ucal = UserCal(backend=house_backend)
|
|
|
+ checkin = date(2024, 8, 5)
|
|
|
+ checkout = checkin + timedelta(days=2)
|
|
|
+
|
|
|
+ await ucal.set('airbnb', 'UC-002', checkin, checkout)
|
|
|
+ await ucal.set('airbnb', 'UC-002', checkin, checkout)
|
|
|
+
|
|
|
+ events = await ucal.find_services(checkin, checkout)
|
|
|
+ matches = [e for e in events if json.loads(e.x_data).get('reservation_id') == 'UC-002']
|
|
|
+ assert len(matches) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_usercal_find_closed_excludes_service_events(house_backend):
|
|
|
+ """find_closed only returns 'Room Closed' events; service events are excluded."""
|
|
|
+ ucal = UserCal(backend=house_backend)
|
|
|
+ checkin = date(2024, 8, 10)
|
|
|
+ checkout = checkin + timedelta(days=2)
|
|
|
+
|
|
|
+ await ucal.set('airbnb', 'UC-003', checkin, checkout)
|
|
|
+
|
|
|
+ closed_events = await ucal.find_closed(checkin, checkout)
|
|
|
+ assert all(e.title == 'Room Closed' for e in closed_events)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_usercal_delete(house_backend):
|
|
|
+ ucal = UserCal(backend=house_backend)
|
|
|
+ checkin = date(2024, 8, 15)
|
|
|
+ checkout = checkin + timedelta(days=1)
|
|
|
+
|
|
|
+ await ucal.set('booking', 'UC-004', checkin, checkout)
|
|
|
+
|
|
|
+ events = await ucal.find_services(checkin, checkout)
|
|
|
+ for e in events:
|
|
|
+ if json.loads(e.x_data).get('reservation_id') == 'UC-004':
|
|
|
+ await ucal.delete(e)
|
|
|
+
|
|
|
+ remaining = await ucal.find_services(checkin, checkout)
|
|
|
+ assert not any(json.loads(e.x_data).get('reservation_id') == 'UC-004' for e in remaining)
|
|
|
+
|
|
|
+
|
|
|
+# ── CalPrice ───────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_and_get_prices(sqlite_backend):
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 1)
|
|
|
+
|
|
|
+ prices = await cal.set(target, 100, discount_rec=None)
|
|
|
+ assert prices.base == 100
|
|
|
+ assert prices.net == 100
|
|
|
+
|
|
|
+ readback = await cal._get_prices(target)
|
|
|
+ assert readback is not None
|
|
|
+ assert readback.net == 100
|
|
|
+ assert readback.base == 100
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_includes_service_prices(sqlite_backend):
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 2)
|
|
|
+
|
|
|
+ prices = await cal.set(target, 80, discount_rec=None)
|
|
|
+ assert len(prices.service_prices) > 0
|
|
|
+ for svc, price in prices.service_prices.items():
|
|
|
+ assert isinstance(price, (int, float))
|
|
|
+ assert price > 0
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_with_flat_discount(sqlite_backend):
|
|
|
+ """CalPrice.set applies a flat discount when passed via discount_rec."""
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 3)
|
|
|
+ discount = DiscountData(type=DiscountType.FLAT, value=10, source={})
|
|
|
+
|
|
|
+ prices = await cal.set(target, 100, discount_rec=discount)
|
|
|
+ assert prices.net == 90
|
|
|
+ assert prices.discount == 10
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_with_percent_discount(sqlite_backend):
|
|
|
+ """CalPrice.set applies a percent discount when passed via discount_rec."""
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 4)
|
|
|
+ discount = DiscountData(type=DiscountType.PERCENT, value=10, source={})
|
|
|
+
|
|
|
+ prices = await cal.set(target, 100, discount_rec=discount)
|
|
|
+ assert prices.net == 90
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_is_idempotent(sqlite_backend):
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 5)
|
|
|
+
|
|
|
+ await cal.set(target, 75, discount_rec=None)
|
|
|
+ await cal.set(target, 75, discount_rec=None)
|
|
|
+
|
|
|
+ events = await cal.list(target, target)
|
|
|
+ assert len(events) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_set_updates_price(sqlite_backend):
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ target = date(2024, 9, 6)
|
|
|
+
|
|
|
+ await cal.set(target, 60, discount_rec=None)
|
|
|
+ await cal.set(target, 80, discount_rec=None)
|
|
|
+
|
|
|
+ readback = await cal._get_prices(target)
|
|
|
+ assert readback.net == 80
|
|
|
+ assert len(await cal.list(target, target)) == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_get_fills_defaults_for_missing_dates(sqlite_backend):
|
|
|
+ """CalPrice.get returns a default price for dates with no event."""
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ start = date(2024, 10, 1)
|
|
|
+ end = date(2024, 10, 3)
|
|
|
+
|
|
|
+ result = await cal.get(start, end)
|
|
|
+ assert len(result) == 3
|
|
|
+ for prices in result.values():
|
|
|
+ assert prices is not None
|
|
|
+ assert prices.net > 0
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_calprice_list(sqlite_backend):
|
|
|
+ cal = CalPrice(backend=sqlite_backend)
|
|
|
+ start = date(2024, 10, 5)
|
|
|
+
|
|
|
+ for i in range(3):
|
|
|
+ await cal.set(start + timedelta(days=i), 70 + i, discount_rec=None)
|
|
|
+
|
|
|
+ events = await cal.list(start, start + timedelta(days=2))
|
|
|
+ assert len(events) == 3
|