201 lines
10 KiB
Python
201 lines
10 KiB
Python
import importlib.util
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, Mock, patch
|
|
|
|
from icalendar import Calendar, Event
|
|
import recurring_ical_events
|
|
|
|
spec = importlib.util.spec_from_file_location('dashboard', Path(__file__).parents[1] / 'app.py')
|
|
app = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(app)
|
|
TOKEN = 'test-proxy-credential-that-is-long-enough'
|
|
CONFIG = {'dav_url': 'https://dav.example/dav', 'dav_user': 'reader', 'dav_password': 'private-dav',
|
|
'ha_url': 'http://ha', 'ha_access_token': 'private-ha', 'proxy_token': TOKEN}
|
|
|
|
|
|
def request(application, path='/api/dashboard', method='GET', token=TOKEN):
|
|
result = {}
|
|
def start_response(status, headers):
|
|
result['status'] = int(status.split()[0])
|
|
result['headers'] = dict(headers)
|
|
result['body'] = b''.join(application({'PATH_INFO': path, 'REQUEST_METHOD': method, 'HTTP_X_DASHBOARD_KEY': token}, start_response))
|
|
return result
|
|
|
|
|
|
class Events(unittest.TestCase):
|
|
def event(self, text):
|
|
return Event.from_ical('BEGIN:VEVENT\r\nUID:test\r\nSUMMARY:Test\r\n' + text + '\r\nEND:VEVENT')
|
|
|
|
def test_exclusive_all_day_end(self):
|
|
e = app.serialize_event(self.event('DTSTART;VALUE=DATE:20261004\r\nDTEND;VALUE=DATE:20261006'), 'Work', '#fff')
|
|
self.assertTrue(e['allDay'])
|
|
self.assertEqual((e['start'], e['end']), ('2026-10-04', '2026-10-06'))
|
|
|
|
def test_floating_time_and_duration(self):
|
|
e = app.serialize_event(self.event('DTSTART:20261004T090000\r\nDURATION:PT30M'), 'Work', '#fff')
|
|
self.assertEqual(e['start'], '2026-10-04T09:00:00-05:00')
|
|
self.assertEqual(e['end'], '2026-10-04T09:30:00-05:00')
|
|
|
|
def test_point_event_preserves_zero_duration(self):
|
|
e = app.serialize_event(self.event('DTSTART:20261004T090000'), 'Work', '#fff')
|
|
self.assertEqual(e['start'], e['end'])
|
|
self.assertTrue(e['point'])
|
|
|
|
def test_default_all_day_duration_is_one_day(self):
|
|
e = app.serialize_event(self.event('DTSTART;VALUE=DATE:20261004'), 'Work', '#fff')
|
|
self.assertEqual(e['end'], '2026-10-05')
|
|
|
|
def test_repeated_dst_hour_is_not_a_negative_interval(self):
|
|
e = app.serialize_event(self.event('DTSTART:20261101T064500Z\r\nDTEND:20261101T071500Z'), 'Work', '#fff')
|
|
self.assertEqual(e['start'], '2026-11-01T01:45:00-05:00')
|
|
self.assertEqual(e['end'], '2026-11-01T01:15:00-06:00')
|
|
self.assertFalse(e['point'])
|
|
|
|
def test_invalid_intervals_are_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
app.serialize_event(self.event('DTSTART:20261004T090000\r\nDTEND:20261004T080000'), 'Work', '#fff')
|
|
|
|
def test_cancelled_event_hidden(self):
|
|
self.assertIsNone(app.serialize_event(self.event('STATUS:CANCELLED\r\nDTSTART:20261004T090000'), 'Work', '#fff'))
|
|
|
|
def test_recurrence_exclusion_and_moved_instance(self):
|
|
calendar = Calendar.from_ical('''BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:recurring
|
|
SUMMARY:Daily
|
|
DTSTART:20261004T140000Z
|
|
DTEND:20261004T150000Z
|
|
RRULE:FREQ=DAILY;COUNT=4
|
|
EXDATE:20261005T140000Z
|
|
END:VEVENT
|
|
BEGIN:VEVENT
|
|
UID:recurring
|
|
RECURRENCE-ID:20261006T140000Z
|
|
SUMMARY:Moved
|
|
DTSTART:20261006T160000Z
|
|
DTEND:20261006T170000Z
|
|
END:VEVENT
|
|
END:VCALENDAR''')
|
|
occurrences = recurring_ical_events.of(calendar).between((2026, 10, 4), (2026, 10, 8))
|
|
events = [app.serialize_event(c, 'Work', '#fff') for c in occurrences]
|
|
self.assertEqual(len(events), 3)
|
|
self.assertEqual(next(e for e in events if e['title'] == 'Moved')['start'], '2026-10-06T11:00:00-05:00')
|
|
self.assertFalse(any(e['start'].startswith('2026-10-05') for e in events))
|
|
|
|
|
|
class Sources(unittest.TestCase):
|
|
def test_failed_collection_preserves_data_and_does_not_block_another(self):
|
|
store = app.Store()
|
|
store.publish('tasks:Inbox', [{'title': 'old'}], now=100)
|
|
store.fail('tasks:Inbox', 'TimeoutError')
|
|
store.publish('tasks:Next', [{'title': 'new'}], now=200)
|
|
data = store.snapshot(now=201)
|
|
self.assertEqual(data['tasks']['Inbox'][0]['title'], 'old')
|
|
self.assertEqual(data['tasks']['Next'][0]['title'], 'new')
|
|
self.assertEqual(data['sources']['tasks:Inbox']['status'], 'error')
|
|
self.assertEqual(data['sources']['tasks:Next']['status'], 'fresh')
|
|
|
|
def test_staleness_is_derived_without_a_worker_error(self):
|
|
store = app.Store(); store.publish('tasks:Inbox', [], now=100)
|
|
self.assertEqual(store.snapshot(now=1000)['sources']['tasks:Inbox']['status'], 'fresh')
|
|
self.assertEqual(store.snapshot(now=1001)['sources']['tasks:Inbox']['status'], 'stale')
|
|
|
|
def test_malformed_item_does_not_discard_good_collection_items(self):
|
|
reader = app.DavReader(CONFIG)
|
|
good = Mock(); good.icalendar_component = Event.from_ical('BEGIN:VEVENT\nUID:x\nDTSTART:20261004T090000\nSUMMARY:Good\nEND:VEVENT')
|
|
bad = Mock(); bad.icalendar_component = Event.from_ical('BEGIN:VEVENT\nUID:y\nSUMMARY:No date\nEND:VEVENT')
|
|
client = MagicMock(); calendar = Mock(); calendar.search.return_value = [good, bad]
|
|
with patch.object(reader, 'collection_url', return_value='https://dav.example/work'), patch.object(reader, 'client', return_value=client), patch.object(app.caldav, 'Calendar', return_value=calendar):
|
|
items, invalid = reader.fetch('Work', '#ffffff')
|
|
self.assertEqual(len(items), 1); self.assertEqual(invalid, 1)
|
|
self.assertTrue(calendar.search.call_args.kwargs['expand'])
|
|
|
|
def test_failure_retry_keeps_last_snapshot(self):
|
|
store = app.Store(); store.publish('tasks:Inbox', [{'title': 'cached'}], now=100)
|
|
stop = Mock(); stop.is_set.side_effect = [False, True]
|
|
with patch.object(app.random, 'uniform', return_value=0):
|
|
app.refresh_source('tasks:Inbox', Mock(side_effect=TimeoutError), store, stop)
|
|
self.assertEqual(store.snapshot(now=101)['tasks']['Inbox'][0]['title'], 'cached')
|
|
self.assertLessEqual(stop.wait.call_args.args[0], 30)
|
|
|
|
|
|
class Routes(unittest.TestCase):
|
|
def setUp(self):
|
|
self.store = app.Store(); self.application = app.Application(CONFIG, self.store)
|
|
|
|
def test_direct_access_is_denied_including_non_ascii_header(self):
|
|
for token in ('', 'wrong', 'é' * 32):
|
|
response = request(self.application, token=token)
|
|
self.assertEqual(response['status'], 403)
|
|
self.assertNotIn(b'events', response['body'])
|
|
|
|
def test_only_read_methods_and_fixed_routes(self):
|
|
self.assertEqual(request(self.application, method='POST')['status'], 405)
|
|
self.assertEqual(request(self.application, '/../app.py')['status'], 404)
|
|
self.assertEqual(request(self.application, '/run/secrets/dashboard.json')['status'], 404)
|
|
head = request(self.application, '/app.js', method='HEAD')
|
|
self.assertEqual(head['status'], 200); self.assertEqual(head['body'], b'')
|
|
self.assertGreater(int(head['headers']['Content-Length']), 0)
|
|
|
|
def test_snapshot_and_assets_never_contain_credentials(self):
|
|
for path in ('/api/dashboard', '/', '/app.js', '/style.css'):
|
|
body = request(self.application, path)['body']
|
|
for value in ('private-ha', 'private-dav', TOKEN):
|
|
self.assertNotIn(value.encode(), body)
|
|
|
|
def test_liveness_remains_up_and_readiness_rejects_stale_data(self):
|
|
self.assertEqual(request(self.application, '/health/live', token='')['status'], 200)
|
|
for key in app.REQUIRED:
|
|
self.store.publish(key, {'forecast': [], 'unit': '°F'} if key == 'weather:hourly' else [], now=100)
|
|
with patch.object(app.time, 'time', return_value=200):
|
|
self.assertEqual(request(self.application, '/health')['status'], 200)
|
|
with patch.object(app.time, 'time', return_value=1001):
|
|
self.assertEqual(request(self.application, '/health')['status'], 503)
|
|
self.assertEqual(request(self.application, '/health/live')['status'], 200)
|
|
|
|
def test_asset_version_in_html_matches_snapshot(self):
|
|
self.assertIn(app.VERSION.encode(), request(self.application, '/')['body'])
|
|
self.assertEqual(json.loads(request(self.application)['body'])['version'], app.VERSION)
|
|
|
|
def test_config_rejects_old_browser_session_and_missing_proxy_key(self):
|
|
app.validate_config(CONFIG)
|
|
old = dict(CONFIG); old.pop('ha_access_token'); old['ha_refresh_token'] = 'borrowed'
|
|
with self.assertRaises(ValueError): app.validate_config(old)
|
|
missing = dict(CONFIG); missing.pop('proxy_token')
|
|
with self.assertRaises(ValueError): app.validate_config(missing)
|
|
|
|
|
|
class Weather(unittest.TestCase):
|
|
def socket(self):
|
|
socket = Mock()
|
|
socket.recv.side_effect = [json.dumps({'type': 'auth_required'}), json.dumps({'type': 'auth_ok'}),
|
|
json.dumps({'id': 1, 'type': 'result', 'success': True}),
|
|
json.dumps({'id': 99, 'type': 'event', 'event': {'forecast': []}}),
|
|
json.dumps({'id': 1, 'type': 'event', 'event': {'forecast': [{'datetime': '2026-10-04T19:00:00-05:00', 'temperature': 70}]}})]
|
|
return socket
|
|
|
|
def test_hourly_is_independent_of_daily_and_uses_only_read_requests(self):
|
|
socket = self.socket(); response = MagicMock(); response.__enter__.return_value = response
|
|
response.json.return_value = {'state': 'sunny', 'attributes': {'temperature_unit': '°F'}}
|
|
with patch.object(app.requests, 'post') as post, patch.object(app.requests, 'get', return_value=response), patch.object(app.websocket, 'create_connection', return_value=socket):
|
|
result, invalid = app.fetch_weather(CONFIG, 'hourly')
|
|
post.assert_not_called(); self.assertEqual(len(result['forecast']), 1); self.assertEqual(invalid, 0)
|
|
messages = [json.loads(call.args[0]) for call in socket.send.call_args_list]
|
|
self.assertEqual([m['type'] for m in messages], ['auth', 'weather/subscribe_forecast'])
|
|
self.assertEqual(messages[-1]['forecast_type'], 'hourly')
|
|
socket.close.assert_called_once()
|
|
|
|
def test_daily_does_not_depend_on_hourly_state_endpoint(self):
|
|
with patch.object(app.requests, 'get') as get, patch.object(app.websocket, 'create_connection', return_value=self.socket()):
|
|
result, invalid = app.fetch_weather(CONFIG, 'daily')
|
|
get.assert_not_called(); self.assertEqual(len(result), 1)
|
|
|
|
def test_absolute_deadline_cannot_be_extended_by_unrelated_messages(self):
|
|
socket = Mock()
|
|
with patch.object(app.time, 'monotonic', return_value=31):
|
|
with self.assertRaises(TimeoutError): app.receive(socket, deadline=30)
|
|
socket.recv.assert_not_called()
|