210 lines
13 KiB
Python
210 lines
13 KiB
Python
import csv
|
|
import io
|
|
import json
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.app import db
|
|
from backend.app.clients.jellystat import HistoryLimitError, JellystatClient, JellystatError
|
|
from backend.app.routers import insights as router
|
|
from backend.app.services import insights, monthly_reports as reports
|
|
from backend.app.services.jellyfin_identity import link_user
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
from backend.tests.test_insights import LIBRARIES, NOW, USER, play
|
|
|
|
|
|
class MonthlyPeriodTests(unittest.TestCase):
|
|
def test_default_is_last_complete_calendar_month(self):
|
|
period = reports.month_periods(None, NOW)
|
|
self.assertEqual(period['month'], '2026-08')
|
|
self.assertEqual(period['period_start'], '2026-08-01T00:00:00+00:00')
|
|
self.assertEqual(period['period_end'], '2026-09-01T00:00:00+00:00')
|
|
self.assertEqual(period['comparison_start'], '2026-07-01T00:00:00+00:00')
|
|
self.assertEqual(period['comparison_end'], period['period_start'])
|
|
self.assertFalse(period['is_partial'])
|
|
self.assertEqual(len(period['available_months']), 24)
|
|
|
|
def test_leap_year_and_year_rollover(self):
|
|
period = reports.month_periods('2024-02', datetime(2024, 3, 2, tzinfo=timezone.utc))
|
|
data = insights.summarize([], LIBRARIES, datetime.fromisoformat(period['period_start']),
|
|
datetime.fromisoformat(period['period_end']), end_exclusive=True)
|
|
self.assertEqual(len(data['daily']), 29)
|
|
self.assertEqual(data['daily'][-1]['date'], '2024-02-29')
|
|
period = reports.month_periods(None, datetime(2026, 1, 1, tzinfo=timezone.utc))
|
|
self.assertEqual(period['month'], '2025-12')
|
|
self.assertEqual(period['comparison_month'], '2025-11')
|
|
|
|
def test_partial_month_matches_elapsed_time_and_caps_short_month(self):
|
|
period = reports.month_periods('2026-09', NOW)
|
|
self.assertTrue(period['is_partial'])
|
|
self.assertEqual(period['comparison_end'], '2026-08-07T12:00:00+00:00')
|
|
self.assertFalse(period['comparison_capped'])
|
|
period = reports.month_periods('2026-03', datetime(2026, 3, 31, 12, tzinfo=timezone.utc))
|
|
self.assertEqual(period['comparison_end'], '2026-03-01T00:00:00+00:00')
|
|
self.assertTrue(period['comparison_capped'])
|
|
|
|
def test_utc_month_is_used_near_local_month_boundary(self):
|
|
local = datetime(2026, 9, 1, 0, 30, tzinfo=timezone(timedelta(hours=12)))
|
|
self.assertEqual(reports.month_periods(None, local)['month'], '2026-07')
|
|
|
|
def test_invalid_future_and_out_of_range_months_are_rejected(self):
|
|
for month in ['', '2026-9', '2026-00', '2026-13', '2026-10', '2024-09', '2026-08\r\nheader', '../../file']:
|
|
with self.subTest(month=month), self.assertRaises(ValueError):
|
|
reports.month_periods(month, NOW)
|
|
|
|
def test_changes_handle_zero_baselines_and_decreases(self):
|
|
self.assertEqual(reports.change(0, 0), {'current': 0, 'previous': 0, 'difference': 0, 'percent': 0})
|
|
self.assertIsNone(reports.change(5, 0)['percent'])
|
|
self.assertEqual(reports.change(30, 60)['percent'], -50)
|
|
|
|
def test_adjacent_months_never_double_count_boundary_play(self):
|
|
start = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
end = datetime(2026, 9, 1, tzinfo=timezone.utc)
|
|
row = play(ActivityDateInserted=start.isoformat())
|
|
previous = insights.summarize([row], LIBRARIES, reports.shift_month(start, -1), start, end_exclusive=True)
|
|
current = insights.summarize([row, row, play('next-month', ActivityDateInserted=end.isoformat())], LIBRARIES, start, end, end_exclusive=True)
|
|
self.assertEqual(previous['summary']['plays'], 0)
|
|
self.assertEqual(current['summary']['plays'], 1)
|
|
self.assertEqual(current['daily'][-1]['date'], '2026-08-31')
|
|
|
|
|
|
class MonthlyReportTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
reports._cache.clear()
|
|
db.create_user('viewer', 'Test-Password123!', auth_provider='jellyfin')
|
|
link_user('viewer', 'jf-viewer', 'http://jellyfin')
|
|
self.runtime = SimpleNamespace(jellyfin_base_url='http://jellyfin', jellyfin_api_key='PRIVATE-JF-KEY',
|
|
jellystat_base_url='http://jellystat', jellystat_api_key='PRIVATE-STATS-KEY')
|
|
self.runtime_patch = patch.object(reports, 'get_runtime_settings', return_value=self.runtime)
|
|
self.runtime_patch.start()
|
|
self.addCleanup(self.runtime_patch.stop)
|
|
self.clock = patch.object(reports, 'datetime', wraps=datetime)
|
|
self.clock.start().now.return_value = NOW
|
|
self.addCleanup(self.clock.stop)
|
|
|
|
def add_request(self, request_id, date, owner=42, status=2):
|
|
db.upsert_request_cache(request_id, request_id, 'movie', status, 'Request', 2026,
|
|
'viewer', 'viewer', owner, date, date, '{}')
|
|
|
|
async def test_report_uses_linked_identity_and_separates_periods_and_request_owners(self):
|
|
rows = [play('previous', PlaybackDuration=1800, ActivityDateInserted='2026-07-31T23:59:59Z'),
|
|
play('current', ActivityDateInserted='2026-08-01T00:00:00Z'),
|
|
play('future', ActivityDateInserted='2026-09-01T00:00:00Z')]
|
|
for request_id, date, owner in [(1, '2026-07-31T23:59:59Z', 42), (2, '2026-08-01T00:00:00Z', 42),
|
|
(3, '2026-08-15T00:00:00Z', 99), (4, '2026-09-01T00:00:00Z', 42)]:
|
|
self.add_request(request_id, date, owner)
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock, return_value=(rows, LIBRARIES)) as remote, \
|
|
patch.object(insights.JellyfinClient, 'get_users', new_callable=AsyncMock) as directory:
|
|
result = await reports.get_monthly_report(USER, '2026-08')
|
|
self.assertEqual(result['summary']['minutes'], 60)
|
|
self.assertEqual(result['previous_summary']['minutes'], 30)
|
|
self.assertEqual(result['changes']['minutes']['percent'], 100)
|
|
self.assertEqual(result['requests']['total'], 1)
|
|
self.assertEqual(result['previous_requests']['total'], 1)
|
|
self.assertEqual(result['requests']['recent'][0]['request_id'], 2)
|
|
self.assertEqual(remote.await_args.args[0], 'jf-viewer')
|
|
directory.assert_not_called()
|
|
self.assertNotIn('PRIVATE', json.dumps(result))
|
|
self.assertNotIn('artwork_item_id', json.dumps(result))
|
|
self.assertNotIn('jf-viewer', json.dumps(result))
|
|
|
|
async def test_partial_comparison_ignores_later_days_in_previous_month(self):
|
|
rows = [play('previous', ActivityDateInserted='2026-08-07T11:59:59Z'),
|
|
play('cutoff', ActivityDateInserted='2026-08-07T12:00:00Z'),
|
|
play('later', ActivityDateInserted='2026-08-30T12:00:00Z'),
|
|
play('current', ActivityDateInserted='2026-09-01T12:00:00Z')]
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock, return_value=(rows, LIBRARIES)):
|
|
result = await reports.get_monthly_report(USER, '2026-09')
|
|
self.assertEqual(result['previous_summary']['minutes'], 60)
|
|
self.assertEqual(result['summary']['minutes'], 60)
|
|
|
|
async def test_cache_isolated_by_identity_month_and_connections_but_requests_refresh(self):
|
|
db.create_user('second', 'Test-Password123!', auth_provider='jellyfin')
|
|
link_user('second', 'jf-second', 'http://jellyfin')
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock, return_value=([], LIBRARIES)) as remote:
|
|
await reports.get_monthly_report(USER, '2026-08')
|
|
self.add_request(1, '2026-08-15T12:00:00Z')
|
|
result = await reports.get_monthly_report(USER, '2026-08')
|
|
self.assertEqual(result['requests']['total'], 1)
|
|
self.assertEqual(remote.await_count, 1)
|
|
await reports.get_monthly_report({**USER, 'username': 'second'}, '2026-08')
|
|
await reports.get_monthly_report(USER, '2026-07')
|
|
self.runtime.jellystat_api_key = 'rotated-key'
|
|
await reports.get_monthly_report(USER, '2026-08')
|
|
self.runtime.jellystat_base_url = 'http://other-jellystat'
|
|
await reports.get_monthly_report(USER, '2026-08')
|
|
self.assertEqual(remote.await_count, 5)
|
|
|
|
async def test_unlinked_and_unconfigured_never_fetch_history(self):
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock) as remote:
|
|
result = await reports.get_monthly_report({**USER, 'username': 'unlinked', 'auth_provider': 'local'})
|
|
self.assertEqual(result['state'], 'unlinked')
|
|
self.runtime.jellystat_api_key = None
|
|
result = await reports.get_monthly_report(USER)
|
|
self.assertEqual(result['state'], 'not_configured')
|
|
remote.assert_not_called()
|
|
|
|
async def test_failed_history_is_not_cached_or_returned_as_a_partial_report(self):
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock, side_effect=HistoryLimitError()):
|
|
with self.assertRaises(HistoryLimitError):
|
|
await reports.get_monthly_report(USER)
|
|
self.assertEqual(reports._cache, {})
|
|
|
|
async def test_csv_preserves_unicode_and_quotes_but_blocks_formulas_and_private_fields(self):
|
|
rows = [play('csv', ActivityDateInserted='2026-08-10T12:00:00Z', NowPlayingItemName='=HYPERLINK("x")', Client='\t\ufeff@SUM(1,2)'),
|
|
play('unicode', NowPlayingItemId='second-film', ActivityDateInserted='2026-08-11T12:00:00Z', NowPlayingItemName='Amélie, "Paris"')]
|
|
with patch.object(JellystatClient, 'get_user_history', new_callable=AsyncMock, return_value=(rows, LIBRARIES)):
|
|
report = await reports.get_monthly_report(USER)
|
|
exported = reports.report_csv(report)
|
|
cells = [cell for row in csv.reader(io.StringIO(exported.lstrip('\ufeff'))) for cell in row]
|
|
self.assertIn('\'=HYPERLINK("x")', cells)
|
|
self.assertIn("'\t\ufeff@SUM(1,2)", cells)
|
|
self.assertIn('Amélie, "Paris"', cells)
|
|
for private in ['PRIVATE', 'jf-viewer', 'artwork/', 'token=', 'http://jellystat']:
|
|
self.assertNotIn(private, exported)
|
|
|
|
|
|
class MonthlyReportRouteTests(unittest.TestCase):
|
|
def client(self, authenticated=True):
|
|
app = FastAPI()
|
|
app.include_router(router.router)
|
|
if authenticated:
|
|
app.dependency_overrides[router.get_current_user] = lambda: {**USER, "features": {"stats": True}}
|
|
return TestClient(app)
|
|
|
|
def test_both_formats_require_auth_and_reject_scope_overrides(self):
|
|
for path in ['/insights/reports/monthly', '/insights/reports/monthly.csv']:
|
|
self.assertEqual(self.client(False).get(path).status_code, 401)
|
|
for query in ['userid=other', 'scope=server', 'user_id=1', 'month=2026-9', 'month=2026-08%0D%0Ax']:
|
|
self.assertEqual(self.client().get(path+'?'+query).status_code, 422)
|
|
|
|
def test_report_no_store_and_export_attachment_headers(self):
|
|
with patch.object(router, 'get_monthly_report', new_callable=AsyncMock, return_value={'state':'ready', 'month':'2026-08'}) as report, \
|
|
patch.object(router, 'report_csv', return_value='\ufeffMetric,Value\r\nMinutes,60\r\n'):
|
|
response = self.client().get('/insights/reports/monthly?month=2026-08')
|
|
self.assertEqual(response.headers['cache-control'], 'no-store')
|
|
report.assert_awaited_with({**USER, "features": {"stats": True}}, '2026-08')
|
|
response = self.client().get('/insights/reports/monthly.csv?month=2026-08')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.headers['content-type'], 'text/csv; charset=utf-8')
|
|
self.assertEqual(response.headers['cache-control'], 'no-store')
|
|
self.assertEqual(response.headers['x-content-type-options'], 'nosniff')
|
|
self.assertEqual(response.headers['content-disposition'], 'attachment; filename="magent-monthly-report-2026-08.csv"')
|
|
self.assertTrue(response.content.startswith(b'\xef\xbb\xbf'))
|
|
|
|
def test_errors_are_sanitized_and_unlinked_export_is_blocked(self):
|
|
for exception, status in [(JellystatError('PRIVATE'), 502), (HistoryLimitError('PRIVATE'), 422), (ValueError('PRIVATE'), 422)]:
|
|
with patch.object(router, 'get_monthly_report', new_callable=AsyncMock, side_effect=exception):
|
|
for suffix in ['', '.csv']:
|
|
response = self.client().get('/insights/reports/monthly'+suffix)
|
|
self.assertEqual(response.status_code, status)
|
|
self.assertNotIn('PRIVATE', response.text)
|
|
with patch.object(router, 'get_monthly_report', new_callable=AsyncMock, return_value={'state':'unlinked'}):
|
|
self.assertEqual(self.client().get('/insights/reports/monthly.csv').status_code, 409)
|