alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ec7ebd7013803dbae8ff187aa4c81ecf99b72986
manage/backend/tests/.pi-map.md
T
Developer ec7ebd7013 chore: regenerate stale .pi-map.md / .pi-map.index.md project maps 
Project map had drifted: claimed recharts unused (was used), ObservabilityPage.tsx
exists (refactored to service-tabs/), missed JellyfinNowPlayingWidget +
authentik/backups service types. Regenerate to reflect post-change reality:
new Qbittorrent/LineSeriesChart/Prometheus{Gauge,Mean}Widget files,
service_data.py/qbittorrent_store.py, canonical prometheus-charting +
service-storage specs, archived changes.
2026-07-09 14:05:56 +00:00

52 lines
46 KiB
Markdown
Raw Blame History

# backend/tests
dir: backend/tests
index: backend/tests/.pi-map.index.md
## role
Comprehensive test suite providing unit and integration tests that validate the backend's API endpoints, external service clients, data persistence, and utility functions of a media library management dashboard.
## files
- __init__.py | Swaps the position of two tmux panes within a window or between windows | dep: tmux, sh
- test_api.py | Integration tests for FastAPI endpoints using TestClient with mocked SSH and Jellyfin clients to validate API layer behavior without real remote connections. | exp: class:TestHealth, method:test_health(self, test_client), call:test_client.get, call:response.json, method:test_metrics_endpoint(self, test_client), call:test_client.get, call:response.headers["content-type"].startswith, method:test_request_id_header(self, test_client), call:test_client.get, call:len, class:TestDashboard, method:test_counts(self, test_client), call:test_client.get, call:response.json, method:test_libraries(self, test_client), call:test_client.get, call:response.json, call:len, method:test_activity(self, test_client), call:test_client.get, call:response.json, call:len, call:next, method:test_now_playing_alias(self, test_client), call:test_client.get, call:response.json, call:len, method:test_jellyfin_endpoints_return_503_when_not_configured(self, test_client), call:app.dependency_overrides.pop, call:test_client.get, call:response.json, call:MagicMock, class:TestSettingsReset, method:test_reset_local_database_requires_full_confirmation(self, test_client, tmp_path), call:SettingsStore, call:store.ensure_defaults, call:media_db.write_text, call:media_wal.write_text, call:__import__, call:SimpleNamespace, call:test_client.post, call:app.dependency_overrides.pop, method:test_reset_local_database_wipes_state_and_leaves_no_machines(self, test_client, tmp_path), call:SettingsStore, call:store.ensure_defaults, call:media_db.write_text, call:media_wal.write_text, call:__import__, call:SimpleNamespace, call:test_client.post, call:app.dependency_overrides.pop, call:response.json, call:media_db.exists, call:media_wal.exists, call:store.get_machine, call:len, call:store.list_machines, class:TestFiles, method:test_list_directory(self, test_client), call:test_client.get, call:response.json, method:test_ffprobe(self, test_client), call:test_client.get, call:response.json, method:test_stat(self, test_client), call:test_client.get, call:response.json, method:test_resolve_path(self, test_client), call:test_client.get, call:response.json, method:test_list_failure(self, test_client, mock_ssh), call:CommandResult, call:test_client.get, class:TestMediaIndexApi, method:test_status_includes_build_progress(self, test_client, tmp_path), call:MediaIndex, call:index.init_schema, call:index.set_metadata, call:patch, call:test_client.get, call:response.json, method:test_build_returns_started_when_background_build_is_queued(self, test_client, tmp_path, mock_jellyfin), call:MediaIndex, call:patch, call:FakeProcess, call:test_client.post, call:response.json, call:start_worker.assert_called_once, call:app.dependency_overrides.pop, method:test_stop_requests_cancel(self, test_client, tmp_path), call:MediaIndex, call:index.init_schema, call:index.set_metadata, call:patch, call:test_client.post, call:response.json, call:test_client.get("/api/media/status").json, call:app.dependency_overrides.pop, method:test_force_stop_terminates_worker(self, test_client, tmp_path), call:MediaIndex, call:index.init_schema, call:index.set_metadata, call:patch, call:test_client.post, call:response.json, call:killpg.assert_called, call:test_client.get("/api/media/status").json, call:app.dependency_overrides.pop, method:test_force_stop_returns_conflict_when_idle(self, test_client, tmp_path), call:MediaIndex, call:index.init_schema, call:test_client.post, call:app.dependency_overrides.pop, class:TestJobs, method:test_list_templates(self, test_client), call:test_client.get, call:response.json, call:len, method:test_run_job(self, test_client, mock_ssh), call:CommandResult, call:test_client.post, call:response.json, method:test_run_unknown_job(self, test_client), call:test_client.post, class:TestMonitoring, method:test_prometheus_targets_empty(self, test_client), call:test_client.get, call:response.json, method:test_prometheus_targets_returns_enabled_ssh_node_exporter(self, test_client), call:store.upsert_machine, call:test_client.get, call:response.json, call:len, class:TestResolveServiceRecord, method:_store(self, rows), call:MagicMock, call:next, method:test_service_id_match_returns_record(self), call:self._store, call:patch, call:resolve_service_record, call:mock_build.assert_called_once_with, method:test_service_id_type_mismatch_returns_none(self), call:self._store, call:resolve_service_record, method:test_service_id_disabled_returns_none(self), call:self._store, call:resolve_service_record, method:test_no_service_id_returns_first_enabled(self), call:self._store, call:patch, call:resolve_service_record, call:mock_build.assert_called_once_with, method:test_no_service_id_and_none_enabled_returns_none(self), call:self._store, call:resolve_service_record, class:TestSettingsMachines, method:test_machine_appears_in_prometheus_targets(self, test_client), call:store.upsert_machine, call:test_client.get("/api/monitoring/prometheus-targets").json, call:len, method:test_delete_machine_removed_from_prometheus_targets(self, test_client), call:store.upsert_machine, call:test_client.delete, call:test_client.get("/api/monitoring/prometheus-targets").json, class:TestAlertmanager, method:test_alerts_endpoint_when_not_configured(self, test_client), call:test_client.get, call:response.json, method:test_alerts_endpoint_when_unreachable(self, test_client), call:_am_service, call:patch, call:Exception, call:test_client.get, call:response.json, method:test_alerts_endpoint_returns_summary(self, test_client), call:_am_service, call:MagicMock, call:patch, call:test_client.get, call:response.json, method:test_alerts_endpoint_sends_bearer_token(self, test_client), call:_am_service, call:MagicMock, call:patch, call:test_client.get, method:test_alertmanager_status_when_not_configured(self, test_client), call:test_client.get, call:response.json, method:test_alertmanager_status_when_unreachable(self, test_client), call:_am_service, call:patch, call:Exception, call:test_client.get, call:response.json, method:test_alertmanager_status_returns_cluster_info(self, test_client), call:_am_service, call:MagicMock, call:patch, call:test_client.get, call:response.json, class:TestAlertmanagerWebhook, method:test_webhook_receives_payload_and_logs(self, test_client, caplog), call:caplog.at_level, call:test_client.post, call:response.json, class:TestPrometheusStatus, method:test_prometheus_status_when_not_configured(self, test_client), call:test_client.get, call:response.json, method:test_prometheus_status_when_unreachable(self, test_client), call:ServiceRecord, call:patch, call:Exception, call:test_client.get, call:response.json, method:test_prometheus_status_returns_version(self, test_client), call:ServiceRecord, call:MagicMock, call:patch, call:test_client.get, call:response.json, func:mock_jellyfin(), call:MagicMock, func:mock_ssh(), call:MagicMock, call:CommandResult, call:json.dumps, func:test_client(mock_jellyfin, mock_ssh, tmp_path), call:SettingsStore, call:SimpleNamespace, call:patch, call:TestClient, call:app.dependency_overrides.clear, func:_am_service(), call:cfg.update, call:ServiceRecord | dep: json, types, unittest.mock, pytest, fastapi.testclient, media_library_viewer_api.clients.ssh, media_library_viewer_api.dependencies, media_library_viewer_api.main, media_library_viewer_api.routers.media, media_library_viewer_api.services.media_index, media_library_viewer_api.services.settings_store, media_library_viewer_api.widgets.sources, media_library_viewer_api (main, dependencies, routers, services, clients, widgets)
- test_authentik_client.py | Tests the AuthentikClient class and its associated directory/users API endpoint, covering URL normalization, pagination handling, and error scenarios. | exp: class:TestAuthentikClient, method:test_base_url_normalizes_trailing_slash(self) → None, call:AuthentikClient, method:test_base_url_strips_api_v3_suffix(self) → None, call:AuthentikClient, method:test_bearer_header_is_set(self) → None, call:AuthentikClient, method:test_empty_base_url_raises(self) → None, call:pytest.raises, call:AuthentikClient, method:test_empty_api_token_raises(self) → None, call:pytest.raises, call:AuthentikClient, class:TestAuthentikUsersEndpoint, method:test_not_configured_returns_empty_with_error(self, store: SettingsStore) → None, call:TestClient, call:client.get, call:response.json, func:_encryption_key(monkeypatch: pytest.MonkeyPatch) → None, call:monkeypatch.setenv, call:reset_encryption_key_cache, func:store(tmp_path: Path) → SettingsStore, call:SettingsStore, call:s.ensure_defaults, call:app.dependency_overrides.clear | dep: pathlib, unittest.mock, pytest, cryptography.fernet, fastapi.testclient, media_library_viewer_api.clients.authentik, media_library_viewer_api.dependencies, media_library_viewer_api.main, media_library_viewer_api.services.secrets, media_library_viewer_api.services.settings_store
- test_backups.py | Tests dashboard backup statistics retrieval and backup report submission endpoints with isolated SQLite database instances. | exp: func:test_dashboard_backups(), call:tempfile.TemporaryDirectory, call:Path, call:SettingsStore, call:store.init_schema, call:TestClient, call:client.get, call:response.json, func:test_post_backup_report(), call:tempfile.TemporaryDirectory, call:Path, call:SettingsStore, call:store.init_schema, call:TestClient, call:get_api_key, call:client.post, call:response.json | dep: tempfile, pathlib, fastapi.testclient, media_library_viewer_api.main, media_library_viewer_api.services.settings_store, media_library_viewer_api.services, media_library_viewer_api.auth
- test_config.py | Unit tests for Pydantic-based Settings configuration loading from environment variables | exp: class:TestSettings, method:test_defaults(self), call:patch.dict, call:Settings, method:test_from_env(self), call:patch.dict, call:Settings, method:test_media_root_property(self), call:patch.dict, call:Settings, method:test_path_prefix_property(self), call:patch.dict, call:Settings, method:test_ssh_key_path_joins_directory_and_name(self), call:patch.dict, call:Settings | dep: os, unittest.mock, media_library_viewer_api.config
- test_dashboards.py | Tests CRUD operations for named dashboards, including slug uniqueness and explicit slug handling. | exp: func:_client(tmp_path: Path) → TestClient, call:SettingsStore, call:TestClient, func:test_create_and_list_dashboards(tmp_path: Path), call:_client, call:client.post, call:resp.json, call:client.get("/api/dashboards").json, call:len, call:app.dependency_overrides.clear, func:test_update_dashboard(tmp_path: Path), call:_client, call:client.post("/api/dashboards", json={"label": "First"}).json, call:client.put( f"/api/dashboards/{created['id']}", json={"label": "Renamed", "payload": {"widgets": ["w1"]}}, ).json, call:app.dependency_overrides.clear, func:test_delete_dashboard(tmp_path: Path), call:_client, call:client.post("/api/dashboards", json={"label": "Temp"}).json, call:client.delete, call:client.get("/api/dashboards").json, call:app.dependency_overrides.clear, func:test_slug_collision_appends_suffix(tmp_path: Path), call:_client, call:client.post("/api/dashboards", json={"label": "Overview"}).json, call:app.dependency_overrides.clear, func:test_explicit_slug_respected(tmp_path: Path), call:_client, call:client.post( "/api/dashboards", json={"label": "My Dashboard", "slug": "custom-slug"}, ).json, call:app.dependency_overrides.clear, func:test_update_nonexistent_returns_404(tmp_path: Path), call:_client, call:client.put, call:app.dependency_overrides.clear | dep: pathlib, fastapi.testclient, media_library_viewer_api.dependencies, media_library_viewer_api.main, media_library_viewer_api.services.settings_store
- test_domain_media.py | Unit tests for media normalization helpers that transform raw media item data into structured display formats. | exp: class:TestFirstMediaSource, method:test_empty_item(self), call:first_media_source, method:test_no_sources(self), call:first_media_source, method:test_returns_first(self), call:first_media_source, class:TestMediaStreams, method:test_empty(self), call:media_streams, method:test_all_streams(self), call:len, call:media_streams, method:test_filter_by_type(self), call:len, call:media_streams, class:TestStreamValue, method:test_first_key(self), call:stream_value, method:test_second_key(self), call:stream_value, method:test_none_values(self), call:stream_value, method:test_empty_string(self), call:stream_value, method:test_no_match(self), call:stream_value, class:TestIsHdrItem, method:test_sdr_item(self), call:is_hdr_item, method:test_hdr10(self), call:is_hdr_item, method:test_dolby_vision(self), call:is_hdr_item, method:test_bt2020_transfer(self), call:is_hdr_item, method:test_empty_item(self), call:is_hdr_item, class:TestFormatDateAdded, method:test_none(self), call:format_date_added, method:test_empty(self), call:format_date_added, method:test_iso_format(self), call:format_date_added, class:TestTimestampDateAdded, method:test_none(self), call:timestamp_date_added, method:test_empty(self), call:timestamp_date_added, method:test_valid(self), call:timestamp_date_added, call:isinstance, class:TestFormatRateBitsDecimal, method:test_none(self), call:format_rate_bits_decimal, method:test_empty(self), call:format_rate_bits_decimal, method:test_kbps(self), call:format_rate_bits_decimal, method:test_mbps(self), call:format_rate_bits_decimal, class:TestNormalizeMediaItem, method:test_movie(self), call:normalize_media_item, method:test_episode(self), call:normalize_media_item, method:test_minimal_item(self), call:normalize_media_item, class:TestDisplayMediaRow, method:test_basic(self), call:display_media_row, method:test_sdr(self), call:display_media_row | dep: media_library_viewer_api.domain.media, pytest
- test_jellyseerr_client.py | Unit tests for the JellyseerrClient class covering user retrieval, pagination, and URL normalization. | exp: class:JellyseerrClientTests, method:setUp(self) → None, call:JellyseerrClient, call:MagicMock, method:test_jellyfin_users_accepts_wrapped_payload(self) → None, call:MagicMock, call:self.client.jellyfin_users, call:self.assertEqual, call:len, call:self.session.get.assert_called_once, method:test_jellyfin_users_accepts_list_payload(self) → None, call:MagicMock, call:self.client.jellyfin_users, call:self.assertEqual, method:test_users_uses_take_and_skip(self) → None, call:MagicMock, call:self.client.users, call:self.assertEqual, method:test_absolute_url_normalizes_relative_paths(self) → None, call:self.assertEqual, call:self.client.absolute_url | dep: unittest, unittest.mock, media_library_viewer_api.clients.jellyseerr, unittest.mock.MagicMock, media_library_viewer_api.clients.jellyseerr.JellyseerrClient
- test_jobs.py | Unit tests for job template rendering and safety validation in a media library API | exp: class:TestJobTemplate, method:test_render_basic(self), call:JobTemplate, call:template.render, method:test_render_quotes_spaces(self), call:JobTemplate, call:template.render, method:test_render_quotes_special_chars(self), call:JobTemplate, call:template.render, call:result.startswith, class:TestBuiltinTemplates, method:test_all_templates_exist(self), method:test_all_templates_renderable(self), call:JOB_TEMPLATES.items, call:template.render, method:test_no_destructive_in_phase1(self), call:JOB_TEMPLATES.items | dep: media_library_viewer_api.jobs, media_library_viewer_api.jobs (JobTemplate, JOB_TEMPLATES)
- test_mailer.py | Unit tests for SMTP mail helper functions including HTML-to-text conversion, email sending with attachments, TLS/SSL fallback modes, and error handling. | exp: class:_SMTPContext, method:__init__(self, smtp: MagicMock), method:__enter__(self), method:__exit__(self, exc_type, exc, tb), class:MailerTests, method:test_html_to_text_strips_tags(self) → None, call:html_to_text, call:self.assertIn, method:test_send_email_message_uses_smtp_with_attachments(self) → None, call:SimpleNamespace, call:MagicMock, call:_SMTPContext, call:patch, call:send_email_message, call:EmailAttachment, call:smtp_ssl.assert_not_called, call:smtp.starttls.assert_called_once, call:smtp.login.assert_called_once_with, call:smtp.send_message.assert_called_once, call:self.assertEqual, method:test_send_email_message_falls_back_to_fastmail_mode(self) → None, call:SimpleNamespace, call:MagicMock, call:TimeoutError, call:_SMTPContext, call:patch, call:send_email_message, call:self.assertEqual, call:len, call:fallback_smtp.send_message.assert_called_once, method:test_send_email_message_retries_with_smtp_username_when_from_is_rejected(self) → None, call:SimpleNamespace, call:MagicMock, call:smtplib.SMTPDataError, call:_SMTPContext, call:patch, call:send_email_message, call:smtp_ssl.assert_not_called, call:self.assertEqual, method:test_describe_smtp_error_handles_timeout(self) → None, call:describe_smtp_error, call:TimeoutError, call:self.assertIn, call:detail.lower | dep: smtplib, unittest, types, unittest.mock, media_library_viewer_api.services.mailer, types.SimpleNamespace, unittest.mock.MagicMock, unittest.mock.patch
- test_media_index.py | Unit tests for a SQLite-based media index service, validating query filtering/sorting/pagination, data replacement, metadata handling, and the media index build process. | exp: class:TestMediaIndexStatus, method:test_nonexistent(self, tmp_path), call:MediaIndex, call:idx.status, method:test_empty_index(self, index), call:index.init_schema, call:index.status, method:test_populated(self, populated_index), call:populated_index.status, class:TestMediaIndexQuery, method:test_query_all(self, populated_index), call:populated_index.query, call:len, method:test_filter_by_library(self, populated_index), call:populated_index.query, call:all, method:test_filter_by_type(self, populated_index), call:populated_index.query, method:test_search(self, populated_index), call:populated_index.query, method:test_hdr_filter(self, populated_index), call:populated_index.query, method:test_sdr_filter(self, populated_index), call:populated_index.query, method:test_sort_by_size_desc(self, populated_index), call:populated_index.query, method:test_sort_by_year_asc(self, populated_index), call:populated_index.query, method:test_pagination(self, populated_index), call:populated_index.query, call:len, method:test_display_row_format(self, populated_index), call:populated_index.query, class:TestMediaIndexReplace, method:test_replace_clears_old(self, index), call:index.replace_items, call:index.status, method:test_replace_scoped_by_service_id_preserves_other_services(self, index), call:index.replace_items, call:index.status, call:index.query, class:TestMediaIndexMetadata, method:test_set_and_read_metadata(self, index), call:index.set_metadata, call:index.status, class:TestMediaIndexBuildPaths, method:test_build_media_index_patches_remote_media_root(self, tmp_path), call:MediaIndex, call:build_media_index, call:FakeClient, call:index.query, method:test_build_media_index_reports_progress(self, tmp_path), call:self.calls.append, call:kwargs.get, call:MediaIndex, call:build_media_index, call:FakeClient, call:any, call:next, method:test_build_media_index_can_be_cancelled(self, tmp_path), call:self.calls.append, call:kwargs.get, call:events.append, call:MediaIndex, call:FakeClient, call:pytest.raises, call:build_media_index, call:any, func:index(tmp_path), call:MediaIndex, func:populated_index(index), call:index.replace_items | dep: typing, pytest, media_library_viewer_api.services.media_index
- test_observability.py | Tests for Prometheus observability metrics helpers that record backup run counters and timestamps | exp: func:_metric_samples(metric_name), call:tuple, call:s.labels.values, call:REGISTRY.collect, func:test_record_backup_run_increments_counter(), call:record_backup_run, call:_metric_samples, func:test_record_backup_run_sets_last_success_timestamp(), call:time.time, call:record_backup_run, call:_metric_samples | dep: time, prometheus_client, media_library_viewer_api.observability
- test_path_utils.py | Unit tests for path resolution utilities that handle remote media path prefixing, media root mapping, and combined path resolution. | exp: class:TestApplyRemotePathPrefix, method:test_empty_path(self), call:apply_remote_path_prefix, method:test_empty_prefix(self), call:apply_remote_path_prefix, method:test_whitespace_prefix(self), call:apply_remote_path_prefix, method:test_basic_prefix(self), call:apply_remote_path_prefix, method:test_already_prefixed(self), call:apply_remote_path_prefix, method:test_relative_path(self), call:apply_remote_path_prefix, method:test_trailing_slash_prefix(self), call:apply_remote_path_prefix, method:test_path_with_spaces(self), call:apply_remote_path_prefix, class:TestMapPathToMediaRoot, method:test_empty_path(self), call:map_path_to_media_root, method:test_empty_root(self), call:map_path_to_media_root, method:test_already_under_root(self), call:map_path_to_media_root, method:test_anchor_mapping(self), call:map_path_to_media_root, method:test_anchor_mapping_deeper(self), call:map_path_to_media_root, method:test_no_anchor_match(self), call:map_path_to_media_root, method:test_root_is_just_anchor(self), call:map_path_to_media_root, method:test_path_equals_root(self), call:map_path_to_media_root, class:TestResolveRemoteMediaPath, method:test_empty_path(self), call:resolve_remote_media_path, method:test_mapping_takes_priority(self), call:resolve_remote_media_path, method:test_fallback_prefix_when_no_anchor(self), call:resolve_remote_media_path, method:test_no_mapping_no_prefix(self), call:resolve_remote_media_path, method:test_already_resolved(self), call:resolve_remote_media_path | dep: media_library_viewer_api.path_utils
- test_prometheus_range.py | Unit tests for Prometheus range-query helper functions including step calculation and matrix normalization. | exp: class:TestStepForWindow, method:test_floor_of_fifteen_seconds(self) → None, call:step_for_window, method:test_custom_target_points(self) → None, call:step_for_window, class:TestNormalizePrometheusMatrix, method:test_empty_matrix(self) → None, call:normalize_prometheus_matrix, method:test_drops_dunder_labels_and_joins(self) → None, call:normalize_prometheus_matrix, call:len, method:test_falls_back_to_value_when_no_labels(self) → None, call:normalize_prometheus_matrix, method:test_dedup_collisions_with_suffix(self) → None, call:normalize_prometheus_matrix, method:test_non_numeric_sentinels_become_none(self) → None, call:normalize_prometheus_matrix, method:test_malformed_values_are_ignored_not_raised(self) → None, call:normalize_prometheus_matrix | dep: pytest, media_library_viewer_api.widgets.prometheus_range
- test_qbittorrent_client.py | Unit tests for the QbittorrentClient, verifying URL handling, login flow, cookie reuse, 403 re-login behavior, and data retrieval. | exp: class:QbittorrentClientTests, method:setUp(self) → None, call:QbittorrentClient, call:MagicMock, method:_login_response(self, text) → MagicMock, call:MagicMock, method:_get_response(self, json_data: dict, status_code) → MagicMock, call:MagicMock, method:test_base_url_appends_api_v2(self) → None, call:QbittorrentClient, call:self.assertEqual, method:test_base_url_keeps_existing_api_v2(self) → None, call:QbittorrentClient, call:self.assertEqual, method:test_base_url_strips_trailing_slash(self) → None, call:QbittorrentClient, call:self.assertEqual, method:test_empty_base_url_raises(self) → None, call:self.assertRaises, call:QbittorrentClient, method:test_empty_username_raises(self) → None, call:self.assertRaises, call:QbittorrentClient, method:test_login_posts_credentials(self) → None, call:self._login_response, call:self.client._login, call:self.session.post.assert_called_once, call:self.assertIn, call:self.assertEqual, call:self.assertTrue, method:test_login_failure_raises_runtime_error(self) → None, call:self._login_response, call:self.assertRaises, call:self.client._login, method:test_get_auto_logs_in_on_first_call(self) → None, call:self._login_response, call:self._get_response, call:self.client._get, call:self.session.post.assert_called_once, call:self.assertEqual, method:test_cookie_reuse_does_not_re_login(self) → None, call:self._get_response, call:self.client._get, call:self.session.post.assert_not_called, method:test_403_triggers_re_login(self) → None, call:MagicMock, call:self._get_response, call:self._login_response, call:self.client._get, call:self.assertEqual, call:self.session.post.assert_called_once, method:test_maindata_returns_full_payload(self) → None, call:self._get_response, call:self.client.maindata, call:self.assertEqual, call:len | dep: unittest, unittest.mock, requests, media_library_viewer_api.clients.qbittorrent
- test_service_data.py | Tests for ServiceDataHarness database lifecycle management (migrations, cascade delete) and QbittorrentSampleStore operations (append, window, prune). | exp: class:TestServiceDataHarnessMigrations, method:test_run_migrations_creates_tables(self, tmp_path), call:ServiceDataHarness, call:harness.register, call:harness.run_migrations, call:harness.connect, call:conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall, method:test_migrations_are_idempotent(self, tmp_path), call:ServiceDataHarness, call:harness.register, call:harness.run_migrations, method:test_alter_table_idempotency(self, tmp_path), call:StorageConcern, call:ServiceDataHarness, call:harness.register, call:harness.run_migrations, call:harness.connect, call:conn.execute("PRAGMA table_info(items)").fetchall, class:TestServiceDataHarnessCascadeDelete, method:test_cascade_delete_removes_only_matching_service(self, tmp_path), call:ServiceDataHarness, call:harness.register, call:harness.run_migrations, call:harness.connect, call:conn.commit, call:harness.cascade_delete, call:conn.execute("SELECT service_id, value FROM test_items ORDER BY id").fetchall, call:len, method:test_cascade_delete_skips_missing_concern_db(self, tmp_path), call:ServiceDataHarness, call:harness.register, call:harness.cascade_delete, class:TestQbittorrentSampleStore, method:test_append_and_window(self, store), call:store.append, call:store.window, call:len, method:test_window_with_since_ts(self, store), call:store.append, call:store.window, call:len, method:test_prune_enforces_max_samples(self, store), call:range, call:store.append, call:store.window, call:len, method:test_two_services_do_not_cross_contaminate(self, store), call:store.append, call:store.window, call:len | dep: pytest, media_library_viewer_api.services.qbittorrent_store, media_library_viewer_api.services.service_data
- test_services.py | Tests the service registry, including definitions, encryption/decryption, CRUD operations, and cascade deletes. | exp: func:_encryption_key(monkeypatch), call:monkeypatch.setenv, call:reset_encryption_key_cache, func:client(tmp_path), call:SettingsStore, call:store.ensure_defaults, call:SimpleNamespace, call:patch, call:TestClient, call:app.dependency_overrides.clear, func:test_registry_contains_eight_service_types(), call:set, func:test_jellyseerr_absorbed_into_jellyfin(), call:get_service_definition, func:test_backups_service_definition(), call:get_service_definition, func:test_authentik_service_definition(), call:get_service_definition, func:test_definitions_declare_widget_kinds(), call:get_service_definition, func:test_widget_kind_lookup(), call:get_widget_kind, func:test_service_config_schema_is_json_schema(), call:get_service_definition, func:test_encrypt_decrypt_round_trip(), call:encrypt_value, call:decrypt_value, func:test_encrypt_decrypt_secrets_dict(), call:encrypt_secrets, call:decrypt_secrets, func:test_missing_encryption_key_raises(monkeypatch), call:monkeypatch.delenv, call:reset_encryption_key_cache, call:pytest.raises, call:get_encryption_key, func:test_decrypt_with_wrong_key_raises(monkeypatch), call:encrypt_secrets, call:monkeypatch.setenv, call:Fernet.generate_key().decode, call:reset_encryption_key_cache, call:pytest.raises, call:decrypt_secrets, func:test_invalid_ciphertext_raises(), call:pytest.raises, call:decrypt_value, func:test_list_service_types(client), call:client.get, call:response.json, func:test_service_type_includes_secret_and_widget_metadata(client), call:client.get, call:next, call:response.json, call:set, func:_prometheus_payload(), call:payload.update, func:test_create_and_list_service(client), call:client.post, call:_prometheus_payload, call:response.json, call:client.get, call:len, func:test_list_instances_filters_by_type(client), call:client.post, call:_prometheus_payload, call:client.get, call:len, call:response.json, func:test_update_service_preserves_unsent_secrets(client), call:client.post("/api/services/instances", json=_prometheus_payload()).json, call:_prometheus_payload, call:client.put( f"/api/services/instances/{created['id']}", json={ "service_type": "prometheus", "name": "Renamed Prometheus", "config": {"base_url": "https://prometheus.example.com", "timeout_seconds": 10}, }, ).json, func:test_update_service_can_clear_secret(client), call:client.post("/api/services/instances", json=_prometheus_payload()).json, call:_prometheus_payload, call:client.put( f"/api/services/instances/{created['id']}", json={ "service_type": "prometheus", "name": "Production Prometheus", "config": {"base_url": "https://prometheus.example.com"}, "secrets": {"api_key": ""}, }, ).json, func:test_unknown_service_type_rejected(client), call:client.post, func:test_invalid_config_rejected(client), call:client.post, func:test_service_base_url_requires_http_schema(bad_url), call:get_service_definition, call:pytest.raises, call:model.model_validate, func:test_service_base_url_accepts_absolute_urls(service_type), call:get_service_definition, call:model.model_validate, func:test_unknown_secret_field_rejected(client), call:client.post, func:test_credential_key_in_config_rejected(client), call:client.post, func:test_update_nonexistent_returns_404(client), call:client.put, call:_prometheus_payload, func:test_update_id_mismatch_returns_400(client), call:client.post("/api/services/instances", json=_prometheus_payload()).json, call:_prometheus_payload, call:client.put, func:test_delete_service(client), call:client.post("/api/services/instances", json=_prometheus_payload()).json, call:_prometheus_payload, call:client.delete, call:client.get("/api/services/instances").json, func:test_delete_nonexistent_returns_404(client), call:client.delete, func:test_delete_service_cascades_to_widgets(client, tmp_path), call:store.upsert_service, call:store.connect, call:conn.execute("PRAGMA table_info(dashboard_widgets)").fetchall, call:store.delete_service, call:store.get_service, call:conn.execute( "SELECT COUNT(*) FROM dashboard_widgets WHERE service_id = ?", (service["id"],), ).fetchone, call:int, func:test_cascade_delete_removes_harness_data_across_concerns(tmp_path, monkeypatch), call:monkeypatch.setenv, call:str, call:reset_service_data_harness, call:get_service_data_harness, call:SettingsStore, call:store.ensure_defaults, call:store.upsert_service, call:QbittorrentSampleStore, call:sample_store.append, call:len, call:sample_store.window, call:store.delete_service, call:MediaIndex, call:harness.db_path, call:index.init_schema, call:index.replace_items, call:index.query, func:test_record_and_list_service_task_runs(client), call:store.upsert_service, call:store.record_service_task_run, call:store.list_service_task_runs, call:len, func:test_jellyseerr_migrates_into_single_jellyfin(tmp_path), call:SettingsStore, call:store.ensure_defaults, call:store.upsert_service, call:store.list_services, call:store.get_service, func:test_jellyseerr_dropped_when_no_jellyfin(tmp_path), call:SettingsStore, call:store.ensure_defaults, call:store.upsert_service, call:store.list_services, func:test_jellyseerr_migration_is_idempotent(tmp_path), call:SettingsStore, call:store.ensure_defaults, call:store.upsert_service, call:store.list_services | dep: types, unittest.mock, pytest, cryptography.fernet, fastapi.testclient, pydantic, media_library_viewer_api.dependencies, media_library_viewer_api.integrations.registry, media_library_viewer_api.main, media_library_viewer_api.services.secrets, media_library_viewer_api.services.settings_store, media_library_viewer_api.services.media_index_impl, media_library_viewer_api.services.qbittorrent_store, media_library_viewer_api.services.service_data, media_library_viewer_api (dependencies, integrations.registry, main, services.secrets, services.settings_store, services.media_index_impl, services.qbittorrent_store, services.service_data)
- test_ssh_client.py | Tests that RemoteSSHClient uses existing known hosts file and sets appropriate host key policies without reprobing | exp: func:test_connect_uses_existing_known_hosts_without_reprobing(tmp_path), call:known_hosts_path.write_text, call:MagicMock, call:patch, call:object, call:RemoteSSHClient, call:str, call:client.connect, call:ssh_client.load_system_host_keys.assert_called_once_with, call:ssh_client.load_host_keys.assert_called_once_with, call:ssh_client.set_missing_host_key_policy.assert_called_once, call:ssh_client.connect.assert_called_once | dep: unittest.mock, media_library_viewer_api.clients.ssh, paramiko, pytest
- test_targets.py | Tests the Prometheus Node Exporter target discovery logic for various machine configurations. | exp: class:TestBuildNodeExporterTargets, method:test_disabled_machine_excluded(self, store: SettingsStore), call:store.upsert_machine, call:build_node_exporter_targets, method:test_ssh_enabled_machine_included(self, store: SettingsStore), call:store.upsert_machine, call:build_node_exporter_targets, call:len, method:test_scrape_host_defaults_to_machine_host(self, store: SettingsStore), call:store.upsert_machine, call:build_node_exporter_targets, method:test_local_machine_excluded(self, store: SettingsStore), call:store.upsert_machine, call:build_node_exporter_targets, method:test_missing_host_excluded(self, store: SettingsStore), call:store.upsert_machine, call:build_node_exporter_targets, func:store(tmp_path: Path) → SettingsStore, call:SettingsStore, call:db.init_schema | dep: pathlib, pytest, media_library_viewer_api.services.settings_store, media_library_viewer_api.services.targets
- test_utils.py | Unit tests for media library formatting utility functions including duration, bitrate, file size, stream metadata, and timestamp conversions | exp: class:TestTicksToMinutes, method:test_none(self), call:ticks_to_minutes, method:test_zero(self), call:ticks_to_minutes, method:test_one_hour(self), call:ticks_to_minutes, method:test_90_minutes(self), call:ticks_to_minutes, method:test_rounding(self), call:int, call:ticks_to_minutes, class:TestHumanSize, method:test_none(self), call:human_size, method:test_zero(self), call:human_size, method:test_bytes(self), call:human_size, method:test_kilobytes(self), call:human_size, method:test_megabytes(self), call:human_size, method:test_gigabytes(self), call:human_size, method:test_terabytes(self), call:human_size, class:TestIsKnownVideoFile, method:test_none(self), call:is_known_video_file, method:test_empty(self), call:is_known_video_file, method:test_mkv(self), call:is_known_video_file, method:test_mp4(self), call:is_known_video_file, method:test_txt(self), call:is_known_video_file, method:test_srt(self), call:is_known_video_file, method:test_spaces_in_path(self), call:is_known_video_file, class:TestFormatDuration, method:test_none(self), call:format_duration, method:test_empty(self), call:format_duration, method:test_seconds(self), call:format_duration, method:test_hours(self), call:format_duration, method:test_string_input(self), call:format_duration, class:TestFormatBitrate, method:test_none(self), call:format_bitrate, method:test_empty(self), call:format_bitrate, method:test_low(self), call:format_bitrate, method:test_kbps(self), call:format_bitrate, method:test_mbps(self), call:format_bitrate, method:test_string_input(self), call:format_bitrate, class:TestFfprobeSummary, method:test_empty(self), call:ffprobe_format_summary, method:test_with_format(self), call:ffprobe_format_summary, class:TestSummarizeStreams, method:test_video_streams(self), call:summarize_video_streams, call:len, method:test_audio_streams(self), call:summarize_audio_streams, call:len, method:test_subtitle_streams(self), call:summarize_subtitle_streams, call:len, method:test_summarize_all(self), call:summarize_streams, call:len, class:TestTimestampToLocal, method:test_none(self), call:timestamp_to_local, method:test_valid(self), call:timestamp_to_local | dep: media_library_viewer_api.utils, pytest
- test_widgets.py | Tests the CRUD operations and data-fetching logic for a dashboard widget system, covering both standalone built-in widgets and service-bound widgets. | exp: func:_encryption_key(monkeypatch), call:monkeypatch.setenv, func:client(tmp_path), call:SettingsStore, call:store.ensure_defaults, call:SimpleNamespace, call:patch, call:TestClient, call:app.dependency_overrides.clear, func:_make_prometheus_service(client), call:config.update, call:client.post( "/api/services/instances", json={"service_type": "prometheus", "name": name, "config": config, "enabled": True}, ).json, func:test_list_builtin_kinds(client), call:client.get, call:response.json, func:test_create_and_read_static_widget(client), call:client.post, call:response.json, call:client.get("/api/widgets/instances").json, call:len, func:test_create_backups_widget(client), call:client.post, func:test_widget_filtering_by_service_id_and_scope(client), call:_make_prometheus_service, call:client.post, call:client.get("/api/widgets/instances").json, call:len, call:client.get(f"/api/widgets/instances?service_id={service['id']}").json, call:client.get("/api/widgets/instances?scope=dashboard").json, call:client.get("/api/widgets/instances?scope=service").json, func:test_unknown_builtin_kind_rejected(client), call:client.post, func:test_credential_key_in_config_rejected(client), call:client.post, func:test_create_service_bound_widget(client), call:_make_prometheus_service, call:client.post, call:response.json, func:test_service_bound_widget_unknown_kind_rejected(client), call:_make_prometheus_service, call:client.post, func:test_service_bound_widget_service_not_found_rejected(client), call:client.post, func:test_service_bound_widget_invalid_config_rejected(client), call:_make_prometheus_service, call:client.post, func:test_update_and_delete_widget(client), call:client.post( "/api/widgets/instances", json={"widget_kind": "static", "title": "Note", "config": {"text": "a"}}, ).json, call:client.put( f"/api/widgets/instances/{created['id']}", json={"widget_kind": "static", "title": "Note2", "config": {"text": "b"}}, ).json, call:client.delete, call:client.get("/api/widgets/instances").json, func:test_update_nonexistent_returns_404(client), call:client.put, func:test_update_id_mismatch_returns_400(client), call:client.post( "/api/widgets/instances", json={"widget_kind": "static", "title": "x", "config": {}}, ).json, call:client.put, func:test_fetch_static_widget_data(client), call:client.post( "/api/widgets/instances", json={"widget_kind": "static", "title": "Note", "config": {"text": "hello"}}, ).json, call:client.get, call:response.json, func:test_fetch_backups_widget_data(client), call:client.post( "/api/widgets/instances", json={"widget_kind": "backups", "title": "Backups", "config": {}}, ).json, call:client.get, call:response.json, func:test_fetch_widget_service_not_found(client), call:_make_prometheus_service, call:client.post( "/api/widgets/instances", json={ "service_id": service["id"], "widget_kind": "metric", "title": "x", "config": {"promql": "up"}, }, ).json, call:client.delete, call:client.get("/api/widgets/instances").json, func:test_fetch_widget_service_disabled(client), call:_make_prometheus_service, call:client.post( "/api/widgets/instances", json={ "service_id": service["id"], "widget_kind": "metric", "title": "x", "config": {"promql": "up"}, }, ).json, call:client.put, call:client.get, call:response.json, func:test_fetch_widget_not_found(client), call:client.get, func:test_alertmanager_adapter_summarizes_alerts(), call:AlertmanagerWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, func:test_alertmanager_adapter_applies_severity_filter(), call:AlertmanagerWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, func:test_alertmanager_adapter_sends_bearer_token(), call:AlertmanagerWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, func:test_alertmanager_adapter_missing_service(), call:AlertmanagerWidgetSource, call:adapter.fetch, func:test_static_adapter(), call:StaticWidgetSource, call:adapter.fetch, func:test_backups_adapter(client), call:patch, call:BackupsWidgetSource, call:adapter.fetch, func:test_ssh_task_adapter_missing_service(), call:SshTaskWidgetSource, call:adapter.fetch, func:test_ssh_task_adapter_records_history_on_run(client), call:store.upsert_task, call:store.upsert_service, call:SimpleNamespace, call:SshTaskWidgetSource, call:ServiceRecord, call:patch, call:adapter.fetch, call:store.list_service_task_runs, call:len, func:test_jellyfin_definition_has_now_playing_widget(), call:get_service_definition, func:test_prometheus_chart_adapter_runs_range_query(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:call.args[0].endswith, call:{"start", "end", "step"}.issubset, func:test_prometheus_chart_adapter_requires_promql(), call:PrometheusWidgetSource, call:ServiceRecord, call:adapter.fetch, func:test_prometheus_chart_adapter_degrades_on_http_error(), call:PrometheusWidgetSource, call:ServiceRecord, call:patch, call:req_mod.ConnectionError, call:adapter.fetch, call:result["error"].lower, func:test_jellyfin_now_playing_filters_active_sessions(), call:JellyfinWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:len, func:test_jellyfin_activity_shows_all_sessions(), call:JellyfinWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:len, func:widget_ref_client(monkeypatch), call:monkeypatch.setenv, call:Fernet.generate_key().decode, call:reset_encryption_key_cache, call:SettingsStore, call:str, call:Path, call:tempfile.mkdtemp, call:store.ensure_defaults, call:TestClient, call:app.dependency_overrides.pop, func:get_store_override(), func:test_widget_reference_lifecycle(widget_ref_client), call:store.upsert_service, call:store.list_services, call:store.upsert_widget, call:client.post, call:resp.json, call:client.get, call:len, call:client.delete, call:store.get_widget, func:test_widget_reference_detach(widget_ref_client), call:store.upsert_service, call:store.list_services, call:store.upsert_widget, call:client.post, call:resp.json, call:client.get("/api/widgets/references", params={"dashboard_scope": "main"}).json, call:len, call:store.get_widget, func:test_widget_reference_update_sort_order(widget_ref_client), call:store.upsert_widget, call:client.post, call:resp.json, call:client.put, call:store.get_widget, func:test_prometheus_gauge_adapter_returns_scalar(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:call.args[0].endswith, func:test_prometheus_gauge_adapter_rejects_multi_series(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:result["error"].lower, func:test_prometheus_gauge_adapter_requires_promql(), call:PrometheusWidgetSource, call:ServiceRecord, call:adapter.fetch, func:test_prometheus_mean_adapter_computes_average(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, func:test_prometheus_mean_adapter_rejects_multi_series(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, call:result["error"].lower, func:test_prometheus_mean_adapter_skips_nan_values(), call:PrometheusWidgetSource, call:ServiceRecord, call:SimpleNamespace, call:patch, call:adapter.fetch, func:test_prometheus_mean_adapter_requires_promql(), call:PrometheusWidgetSource, call:ServiceRecord, call:adapter.fetch, func:_fake_qbit_maindata(), func:test_qbittorrent_totals_counts_all_torrents(), call:QbittorrentWidgetSource, call:ServiceRecord, call:patch, call:_fake_qbit_maindata, call:adapter.fetch, func:test_qbittorrent_active_filters_dl_ul_only(), call:QbittorrentWidgetSource, call:ServiceRecord, call:patch, call:_fake_qbit_maindata, call:adapter.fetch, call:len, func:test_qbittorrent_speed_appends_and_returns_series(tmp_path), call:ServiceDataHarness, call:str, call:harness.register, call:harness.run_migrations, call:QbittorrentWidgetSource, call:ServiceRecord, call:patch, call:_fake_qbit_maindata, call:QbittorrentSampleStore, call:adapter.fetch, call:len, func:test_qbittorrent_adapter_missing_service(), call:QbittorrentWidgetSource, call:adapter.fetch, func:test_qbittorrent_adapter_missing_credentials(), call:QbittorrentWidgetSource, call:ServiceRecord, call:adapter.fetch, func:test_qbittorrent_adapter_timeout(), call:QbittorrentWidgetSource, call:ServiceRecord, call:patch, call:_asyncio.sleep, call:(_ for _ in ()).throw, call:TimeoutError, call:adapter.fetch, func:_slow(), call:_asyncio.sleep | dep: types, unittest.mock, pytest, cryptography.fernet, fastapi.testclient, media_library_viewer_api.dependencies, media_library_viewer_api.main, media_library_viewer_api.services.settings_store, media_library_viewer_api.widgets.sources, media_library_viewer_api.integrations.registry, media_library_viewer_api.services.secrets, tempfile, pathlib, media_library_viewer_api.services.qbittorrent_store, media_library_viewer_api.services.service_data, media_library_viewer_api (main, dependencies, services, widgets)
## arch
Layered testing approach using pytest conventions, with TestClient for FastAPI endpoint integration tests, mocked external clients (SSH, Jellyfin, Authentik, qBittorrent, Jellyseerr) for isolated API-layer testing, and direct unit tests for Pydantic configs, SQLite database operations, Prometheus metrics, and path/formatting utilities.
## tags
test, service, widget, call:test, media, prometheus, call:response.json, client
## symbols
- TestHealth
- TestDashboard
- TestSettingsReset
- TestFiles
- TestMediaIndexApi
- TestJobs
- TestMonitoring
- TestResolveServiceRecord
## workflows
- change tests behavior
read: __init__.py, test_api.py, test_authentik_client.py
- change tests CLI
read: test_authentik_client.py, test_jellyseerr_client.py, test_qbittorrent_client.py
- change tests config
read: test_config.py
## dirty
-
Reference in New Issue View Git Blame Copy Permalink