import json import inspect import unittest from unittest.mock import patch import fgai.dashboard as dashboard from fgai.dashboard import HTML, _ollama_models class DashboardTests(unittest.TestCase): def test_ollama_models_returns_sorted_model_names(self): class Response: def __enter__(self): return self def __exit__(self, *_args): return False def read(self): return json.dumps({"models": [{"name": "qwen3:8b", "size": 2}, {"name": "llama3.1", "size": 1}]}).encode("utf-8") with patch("fgai.dashboard.request.urlopen", return_value=Response()): result = _ollama_models() self.assertEqual(result["status"], "ok") self.assertEqual([item["name"] for item in result["models"]], ["llama3.1", "qwen3:8b"]) def test_dashboard_includes_how_to_view(self): self.assertIn('data-tab="howto"', HTML) self.assertIn('data-view="howto"', HTML) self.assertIn("How To Use SignalScope", HTML) def test_dashboard_can_update_existing_recommended_profiles(self): self.assertIn("Update profile", HTML) self.assertIn("function mergeProfile", HTML) def test_dashboard_can_delete_stream_profiles(self): self.assertIn("Remove", HTML) self.assertIn("function deleteStreamProfile", HTML) self.assertIn("delete-stream-profile", HTML) self.assertIn("danger-action", HTML) self.assertIn("profile-readiness", HTML) def test_dashboard_has_direct_stream_selection_save(self): self.assertIn("Save streams", HTML) self.assertIn("function saveStreamSelection", HTML) def test_dashboard_includes_operator_guidance(self): self.assertIn("Operator Guidance", HTML) self.assertIn("function buildActionPlan", HTML) def test_dashboard_links_evidence_to_graylog(self): self.assertIn("function graylogEvidenceLink", HTML) self.assertIn("/search?rangetype=relative", HTML) def test_dashboard_has_correlation_explorer(self): self.assertIn("correlationExplorer", HTML) self.assertIn("function renderCorrelationExplorer", HTML) self.assertIn("correlationHitboxes", HTML) def test_dashboard_shows_refreshing_mcp_as_polling_badge(self): self.assertIn("function mcpCapability", HTML) self.assertIn("connected, ${suffix}", HTML) def test_dashboard_has_visible_incident_action_status(self): self.assertIn("incidentNotice", HTML) self.assertIn("Incident marked", HTML) def test_dashboard_can_clear_stored_incident_states(self): self.assertIn('id="clearResolvedIncidents"', HTML) self.assertIn('id="clearAllIncidents"', HTML) self.assertIn("/api/incidents/clear", HTML) def test_dashboard_exposes_review_state_reset_endpoints(self): source = inspect.getsource(dashboard) self.assertIn("/api/feedback/clear", source) self.assertIn("/api/review-state/clear", source) def test_dashboard_exposes_feedback_revoke_endpoint(self): source = inspect.getsource(dashboard) self.assertIn("/api/feedback/revoke", source) self.assertIn(".revoke(", source) def test_dashboard_exposes_feedback_bulk_endpoint(self): source = inspect.getsource(dashboard) self.assertIn("/api/feedback/bulk", source) self.assertIn(".add_many(", source) def test_dashboard_refresh_guards_against_overlapping_calls(self): self.assertIn("refreshInFlight", HTML) self.assertIn("if (refreshInFlight) return;", HTML) def test_dashboard_open_details_survive_refresh_via_toggle_listener(self): self.assertIn("openDetailIds", HTML) self.assertIn("addEventListener('toggle'", HTML) def test_dashboard_annotates_field_deviations_with_cluster_size(self): self.assertIn("cluster_size", HTML) self.assertIn("-entity pattern", HTML) self.assertIn("cluster_dampened", HTML) def test_dashboard_shows_shared_entity_baselines(self): self.assertIn("shared-entity-baselines", HTML) self.assertIn("Shared Entity Baselines", HTML) self.assertIn("shared_entity_baselines", HTML) def test_dashboard_shows_detector_health(self): self.assertIn("detector-health", HTML) self.assertIn("Detector Health", HTML) self.assertIn("detector_health", HTML) def test_review_actions_carry_detector_for_precision_tracking(self): self.assertIn('data-detector="${esc(r.detector', HTML) self.assertIn("detector: button.dataset.detector", HTML) self.assertIn("detector: pendingReview.detector", HTML) self.assertIn("detector: row.detector", HTML) def test_dashboard_shows_shape_mismatch_warnings(self): self.assertIn('id="profileShapeWarnings"', HTML) self.assertIn("profile_shape_warnings", HTML) self.assertIn("looks like free text", HTML) self.assertIn("Open profile editor", HTML) def test_dashboard_has_anomaly_bulk_bar(self): self.assertIn('id="anomalyBulkBar"', HTML) self.assertIn('anomaly-select', HTML) self.assertIn('id="selectAllAnomalies"', HTML) self.assertIn('id="anomalyBulkSubmit"', HTML) def test_dashboard_anomaly_bulk_requires_expiry_and_only_ever_posts_entity_scope(self): self.assertIn('at least 1 day', HTML) start = HTML.index("getElementById('anomalyBulkSubmit')") end = HTML.index('};', start) submit_handler = HTML[start:end] self.assertIn("scope: 'entity'", submit_handler) self.assertIn('days <= 0', submit_handler) self.assertIn('confirm(', submit_handler) def test_dashboard_field_deviations_are_collapsible_and_open_by_default(self): start = HTML.index("Field Baseline Deviations") preceding = HTML[max(0, start - 80):start] self.assertIn('
', preceding) self.assertIn('id="findingSummaryBadge"', HTML) def test_dashboard_sorts_likely_worth_confirming_rows_first(self): self.assertIn('const suggestPriority', HTML) self.assertIn('suggestPriority(left) - suggestPriority(right)', HTML) def test_dashboard_has_generalization_suggestion_panel(self): self.assertIn('id="feedbackGeneralizationSuggestions"', HTML) def test_dashboard_generalization_merge_button_confirms_before_widening_scope(self): self.assertIn('merge-feedback-generalization', HTML) self.assertIn('Merge into one field-wide rule', HTML) self.assertIn('will be revoked', HTML) def test_dashboard_generalization_merge_only_ever_posts_field_scope(self): start = HTML.index("querySelectorAll('.merge-feedback-generalization')") end = HTML.index('}));', start) merge_handler = HTML[start:end] self.assertIn("scope: 'field'", merge_handler) self.assertNotIn("scope: 'entity'", merge_handler) self.assertNotIn("scope: 'stream'", merge_handler) def test_dashboard_exposes_feedback_generalization_setting(self): self.assertIn('name="feedback_generalization_min_values"', HTML) def test_dashboard_suggests_review_from_detector_history(self): self.assertIn("label:'Suggested', render:suggestReview", HTML) self.assertIn("likely false positive", HTML) self.assertIn("likely worth confirming", HTML) self.assertIn("not enough history", HTML) def test_dashboard_secondary_findings_panels_are_collapsible(self): self.assertIn('Related Activity Across Sources', HTML) self.assertIn('Block Candidates', HTML) self.assertIn('Threat Intelligence', HTML) self.assertIn('Policy Findings', HTML) self.assertIn('id="relatedActivity"', HTML) def test_dashboard_diagnostics_sections_are_collapsible(self): self.assertIn("collapsibleSection('stream-coverage'", HTML) self.assertIn("collapsibleSection('top-services'", HTML) self.assertIn("class=\"collapsible-section\"", HTML) self.assertIn("diag-stream-coverage", HTML) def test_dashboard_table_scroll_position_survives_refresh(self): self.assertIn("tableScrollLeft", HTML) self.assertIn("data-table-id", HTML) self.assertIn("addEventListener('scroll'", HTML) def test_dashboard_has_bulk_review_bar(self): self.assertIn('id="bulkReviewBar"', HTML) self.assertIn('id="selectAllDeviations"', HTML) self.assertIn("deviation-select", HTML) self.assertIn('id="bulkReviewSubmit"', HTML) def test_bulk_review_scope_never_offers_broad_scopes(self): start = HTML.index('id="bulkReviewScope"') end = HTML.index('', start) bulk_scope_html = HTML[start:end] self.assertIn('value="value"', bulk_scope_html) self.assertIn('value="field"', bulk_scope_html) self.assertNotIn('value="entity"', bulk_scope_html) self.assertNotIn('value="stream"', bulk_scope_html) def test_dashboard_has_active_feedback_rules_panel(self): self.assertIn('id="feedbackRules"', HTML) self.assertIn("revoke-feedback-rule", HTML) self.assertIn('id="feedbackRuleSummary"', HTML) def test_dashboard_has_feedback_rule_status_filter(self): self.assertIn('id="feedbackRuleStatusFilter"', HTML) self.assertIn('value="false_positive">False positive', HTML) def test_dashboard_has_scoped_review_form(self): self.assertIn('id="reviewScope"', HTML) self.assertIn('value="entity"', HTML) self.assertIn('value="stream"', HTML) self.assertIn("Broad-scope rules must expire", HTML) def test_dashboard_does_not_clear_streams_when_picker_is_unloaded(self): self.assertIn("const streamValues = Object.values(window.streamSelection || {})", HTML) self.assertIn("if (streamValues.length)", HTML) self.assertIn("values.graylog_streams = streamValues", HTML) def test_dashboard_can_recover_stream_selection_from_profiles(self): self.assertIn("No enabled streams were saved", HTML) self.assertIn("streams with existing profiles are selected for recovery", HTML) def test_dashboard_shows_profile_advisor_fallback_as_notice(self): self.assertIn("heuristic_fallback", HTML) self.assertIn("Profile advisor: heuristic fallback", HTML) self.assertIn("heuristic fallback", HTML) def test_dashboard_labels_ollama_assessment_errors(self): self.assertIn("function llmCapability", HTML) self.assertIn("assessment unavailable", HTML) self.assertIn("Ollama assessment: unavailable", HTML) def test_dashboard_exposes_ollama_assessment_timeout_setting(self): self.assertIn('name="llm_timeout"', HTML) self.assertIn("llm_timeout: config.llm_timeout || 180", HTML) def test_dashboard_exposes_field_confidence_and_auto_apply_settings(self): self.assertIn('name="field_confidence_threshold"', HTML) self.assertIn('name="profile_advisor_interval_seconds"', HTML) self.assertIn('name="field_confidence_retention_days"', HTML) self.assertIn('name="auto_apply_profile_suggestions"', HTML) self.assertIn("field_confidence_threshold: config.field_confidence_threshold ?? 60", HTML) if __name__ == "__main__": unittest.main()