Add unit tests for Overview and Data Quality Panel

Signed-off-by: Sampurna Pyne <sampurnapyne1710@gmail.com>

Author: Samk1710

insights/tests/test_data_quality_panel.py

@@ -0,0 +1,92 @@
+import datetime
+
+from django.test import TestCase
+from django.utils import timezone
+
+from insights.charts.data_quality_panel import data_quality_todos_resolutions_queryset
+from insights.charts.data_quality_panel import open_issues_to_datasource_queryset
+from vulnerabilities.models import AdvisoryToDoV2
+from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import ToDoRelatedAdvisoryV2
+
+
+class TestDataQualityPanelQuerysets(TestCase):
+    def setUp(self):
+        self.adv1 = AdvisoryV2.objects.create(
+            avid="github_osv/GHSA-1",
+            datasource_id="github_osv",
+            unique_content_id="1",
+            is_latest=True,
+            pipeline_id="github_osv_pipeline",
+            advisory_id="GHSA-1",
+            url="https://example.com/GHSA-1",
+        )
+        self.adv2 = AdvisoryV2.objects.create(
+            avid="nvd/CVE-2023-1234",
+            datasource_id="nvd",
+            unique_content_id="2",
+            is_latest=True,
+            pipeline_id="nvd_pipeline",
+            advisory_id="CVE-2023-1234",
+            url="https://nvd.nist.gov/vuln/detail/CVE-2023-1234",
+        )
+
+        now = timezone.now()
+
+        self.todo1 = AdvisoryToDoV2.objects.create(
+            alias="GHSA-1-TODO",
+            related_advisories_id="hash1",
+            issue_type="MISSING_AFFECTED_PACKAGE",
+            is_resolved=False,
+            created_at=now - datetime.timedelta(days=60),
+        )
+        ToDoRelatedAdvisoryV2.objects.create(todo=self.todo1, advisory=self.adv1)
+
+        self.todo2 = AdvisoryToDoV2.objects.create(
+            alias="CVE-2023-1234-TODO",
+            related_advisories_id="hash2",
+            issue_type="CONFLICTING_SEVERITY_SCORES",
+            is_resolved=False,
+            created_at=now - datetime.timedelta(days=45),
+        )
+        ToDoRelatedAdvisoryV2.objects.create(todo=self.todo2, advisory=self.adv2)
+
+        self.todo3 = AdvisoryToDoV2.objects.create(
+            alias="GHSA-1-TODO-2",
+            related_advisories_id="hash3",
+            issue_type="CONFLICTING_SEVERITY_SCORES",
+            is_resolved=True,
+            created_at=now - datetime.timedelta(days=60),
+            resolved_at=now - datetime.timedelta(days=15),
+        )
+        ToDoRelatedAdvisoryV2.objects.create(todo=self.todo3, advisory=self.adv1)
+
+    def test_open_issues_to_datasource_queryset(self):
+        """Test open issue query correctly aggregates unresolved to-dos by type and source."""
+        qs = list(open_issues_to_datasource_queryset())
+
+        self.assertEqual(len(qs), 2)
+
+        missing_pkg_todo = next(t for t in qs if t["issue_type"] == "MISSING_AFFECTED_PACKAGE")
+        self.assertEqual(missing_pkg_todo["advisories__datasource_id"], "github_osv")
+        self.assertEqual(missing_pkg_todo["count"], 1)
+
+        conflicting_sev_todo = next(
+            t for t in qs if t["issue_type"] == "CONFLICTING_SEVERITY_SCORES"
+        )
+        self.assertEqual(conflicting_sev_todo["advisories__datasource_id"], "nvd")
+        self.assertEqual(conflicting_sev_todo["count"], 1)
+
+    def test_data_quality_todos_resolutions_queryset(self):
+        """Test timeline query separately aggregates opened and resolved to-dos by month."""
+        open_todos_qs, resolved_todos_qs = data_quality_todos_resolutions_queryset()
+
+        open_todos = list(open_todos_qs)
+        resolved_todos = list(resolved_todos_qs)
+
+        self.assertEqual(sum(t["count"] for t in open_todos), 3)
+        self.assertEqual(sum(t["count"] for t in resolved_todos), 1)
+
+        resolved = resolved_todos[0]
+        self.assertEqual(resolved["advisories__datasource_id"], "github_osv")
+        self.assertEqual(resolved["count"], 1)

insights/tests/test_insights_snapshot_pipeline.py

@@ -1,10 +1,13 @@
 from django.test import TestCase
+from django.utils import timezone
 
 from insights.insights_snapshot_pipeline import InsightsSnapshotPipeline
 from insights.models import DailySnapshot
 from insights.tests.test_importer_panel import create_adv
 from vulnerabilities.models import AdvisorySeverity
+from vulnerabilities.models import AdvisoryToDoV2
 from vulnerabilities.models import PackageV2
+from vulnerabilities.models import ToDoRelatedAdvisoryV2
 
 
 class TestInsightsSnapshotPipeline(TestCase):
@@ -16,9 +19,20 @@ def test_pipeline_execution_with_data(self):
 
         # Create Advisory for Importer and Severity charts
         advisory_1 = create_adv("GHSA-1234", "1")
+        advisory_1.date_published = timezone.now()
+        advisory_1.save()
         severity_1 = AdvisorySeverity.objects.create(scoring_system="cvssv3.1", value="9.8")
         advisory_1.severities.add(severity_1)
 
+        todo = AdvisoryToDoV2.objects.create(
+            alias="GHSA-1234-TODO",
+            related_advisories_id="hash1",
+            issue_type="MISSING_AFFECTED_PACKAGE",
+            is_resolved=False,
+            created_at=timezone.now(),
+        )
+        ToDoRelatedAdvisoryV2.objects.create(todo=todo, advisory=advisory_1)
+
         pipeline = InsightsSnapshotPipeline()
         pipeline.execute()
 
@@ -35,3 +49,12 @@ def test_pipeline_execution_with_data(self):
         # Verify SeverityInsight was generated
         self.assertTrue(hasattr(snapshot, "severity_insight"))
         self.assertEqual(snapshot.severity_insight.buckets[9], 1)
+
+        self.assertTrue(hasattr(snapshot, "overview"))
+        self.assertEqual(snapshot.overview.total_advisories, 1)
+        self.assertEqual(len(snapshot.overview.last_30days), 31)
+        self.assertEqual(snapshot.overview.yearly_insights.count(), 1)
+
+        self.assertEqual(snapshot.data_quality_issue_types.count(), 1)
+        self.assertEqual(snapshot.data_quality_issue_types.first().datasource_id, "github_osv")
+        self.assertEqual(snapshot.data_quality_todos_resolutions.count(), 1)

insights/tests/test_overview_panel.py

@@ -0,0 +1,71 @@
+import datetime
+
+from django.test import TestCase
+from django.utils import timezone
+
+from insights.charts.overview_panel import daily_growth_counts
+from insights.charts.overview_panel import kpi_card_queryset
+from insights.charts.overview_panel import yearly_distribution_queryset
+from insights.tests.test_importer_panel import create_adv
+from vulnerabilities.models import PackageV2
+
+
+class TestOverviewPanelQuerysets(TestCase):
+    def setUp(self):
+        PackageV2.objects.create(type="pypi", name="django", version="1.0.0")
+        PackageV2.objects.create(type="npm", name="lodash", version="1.0.0")
+
+        self.now = timezone.now()
+
+        adv1 = create_adv("GHSA-1234", "1")
+        adv1.date_published = self.now - datetime.timedelta(days=365 * 2)
+        adv1.date_collected = self.now - datetime.timedelta(days=5)
+        adv1.datasource_id = "github_osv"
+        adv1.save()
+
+        adv2 = create_adv("GHSA-5678", "2")
+        adv2.date_published = self.now
+        adv2.date_collected = self.now
+        adv2.datasource_id = "github_osv"
+        adv2.save()
+
+        adv3 = create_adv("CVE-2023-7777", "3")
+        adv3.date_published = self.now - datetime.timedelta(days=10)
+        adv3.date_collected = self.now
+        adv3.datasource_id = "nvd"
+        adv3.save()
+
+        adv4 = create_adv("CVE-2024-9999", "4")
+        adv4.date_published = self.now
+        adv4.date_collected = self.now
+        adv4.datasource_id = "nvd"
+        adv4.save()
+
+        adv5 = create_adv("CVE-2024-8888", "5")
+        adv5.date_published = self.now
+        adv5.date_collected = self.now
+        adv5.datasource_id = "nvd"
+        adv5.save()
+
+    def test_kpi_card_queryset(self):
+        """Test KPI queries"""
+        stats = kpi_card_queryset()
+        self.assertEqual(stats["total_advisories"], 5)
+        self.assertEqual(stats["total_packages"], 2)
+        self.assertEqual(stats["total_data_sources"], 2)
+
+    def test_yearly_distribution_queryset(self):
+        """Test yearly distribution query aggregates advisories by publish year."""
+        yearly_records = list(yearly_distribution_queryset())
+        self.assertEqual(len(yearly_records), 2)
+
+        this_year = next(y for y in yearly_records if y["year"] == self.now.year)
+        two_years_ago = next(y for y in yearly_records if y["year"] == self.now.year - 2)
+
+        self.assertEqual(this_year["count"], 4)
+        self.assertEqual(two_years_ago["count"], 1)
+
+    def test_daily_growth_counts(self):
+        """Test daily ingestion query"""
+        counts = daily_growth_counts(self.now)
+        self.assertEqual(sum(counts), 5)