Add tests

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

Author: Samk1710

insights/tests/test_importer_panel.py

@@ -0,0 +1,68 @@
+from django.test import TestCase
+
+from insights.charts.importer_panel import exploits_queryset
+from insights.charts.importer_panel import packages_queryset
+from vulnerabilities.models import AdvisoryExploit
+from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import ImpactedPackage
+from vulnerabilities.models import ImpactedPackageAffecting
+from vulnerabilities.models import PackageV2
+
+
+def create_adv(avid, unique_id):
+    return AdvisoryV2.objects.create(
+        avid=f"github_osv/{avid}",
+        datasource_id="github_osv",
+        unique_content_id=unique_id,
+        is_latest=True,
+        pipeline_id="github_osv_pipeline",
+        advisory_id=avid,
+        url="https://example.com/" + avid,
+    )
+
+
+class TestImporterPanelQuerysets(TestCase):
+    def test_packages_queryset(self):
+        advisory_with_package = create_adv("GHSA-1", "1")
+        pkg1 = PackageV2.objects.create(type="npm", name="test1", version="1.0.0")
+        impact1 = ImpactedPackage.objects.create(advisory=advisory_with_package)
+        ImpactedPackageAffecting.objects.create(impacted_package=impact1, package=pkg1)
+
+        advisory_with_ghost_package = create_adv("GHSA-2", "2")
+        pkg2 = PackageV2.objects.create(type="npm", name="test2", version="2.0.0", is_ghost=True)
+        impact2 = ImpactedPackage.objects.create(advisory=advisory_with_ghost_package)
+        ImpactedPackageAffecting.objects.create(impacted_package=impact2, package=pkg2)
+
+        advisory_without_package = create_adv("GHSA-3", "3")
+
+        qs = list(packages_queryset())
+
+        self.assertEqual(len(qs), 1)
+        stats = qs[0]
+        self.assertEqual(stats["datasource_id"], "github_osv")
+        self.assertEqual(stats["total_advisories"], 3)
+        self.assertEqual(stats["advisories_with_packages"], 2)  # First two have packages
+        self.assertEqual(stats["advisories_with_ghost_packages"], 1)
+
+    def test_exploits_queryset(self):
+        advisory_with_kev = create_adv("GHSA-1", "1")
+        AdvisoryExploit.objects.create(advisory=advisory_with_kev, data_source="KEV")
+
+        advisory_with_metasploit = create_adv("GHSA-2", "2")
+        AdvisoryExploit.objects.create(advisory=advisory_with_metasploit, data_source="Metasploit")
+
+        advisory_with_exploitdb = create_adv("GHSA-3", "3")
+        AdvisoryExploit.objects.create(advisory=advisory_with_exploitdb, data_source="Exploit-DB")
+
+        advisory_without_exploits = create_adv("GHSA-4", "4")
+
+        qs = list(exploits_queryset())
+
+        self.assertEqual(len(qs), 1)
+        stats = qs[0]
+        self.assertEqual(stats["datasource_id"], "github_osv")
+
+        # GHSA-4 shouldn't count in these
+        self.assertEqual(stats["advisories_with_kev"], 1)
+        self.assertEqual(stats["advisories_with_metasploit"], 1)
+        self.assertEqual(stats["advisories_with_exploitdb"], 1)

insights/tests/test_insights_snapshot_pipeline.py

@@ -0,0 +1,37 @@
+from django.test import TestCase
+
+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 PackageV2
+
+
+class TestInsightsSnapshotPipeline(TestCase):
+    def test_pipeline_execution_with_data(self):
+        """Test pipeline populates insights based on existing data."""
+
+        PackageV2.objects.create(type="pypi", name="django", version="1.0.0")
+        PackageV2.objects.create(type="npm", name="lodash", version="1.0.0")
+
+        # Create Advisory for Importer and Severity charts
+        advisory_1 = create_adv("GHSA-1234", "1")
+        severity_1 = AdvisorySeverity.objects.create(scoring_system="cvssv3.1", value="9.8")
+        advisory_1.severities.add(severity_1)
+
+        pipeline = InsightsSnapshotPipeline()
+        pipeline.execute()
+
+        self.assertEqual(DailySnapshot.objects.count(), 1)
+        snapshot = DailySnapshot.objects.first()
+
+        # Verify captured package types (pypi, npm) and 'global'
+        self.assertEqual(snapshot.package_insights.count(), 3)
+
+        # Verify ImporterInsight was created for 'github_osv'
+        self.assertEqual(snapshot.importer_insights.count(), 1)
+        self.assertTrue(snapshot.importer_insights.filter(importer="github_osv").exists())
+
+        # Verify SeverityInsight was generated
+        self.assertTrue(hasattr(snapshot, "severity_insight"))
+        self.assertEqual(snapshot.severity_insight.buckets[9], 1)

insights/tests/test_package_panel.py

@@ -0,0 +1,56 @@
+from django.test import TestCase
+
+from insights.charts.package_panel import ecosystem_distribution_queryset
+from insights.charts.package_panel import top_packages_queryset
+from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import ImpactedPackage
+from vulnerabilities.models import ImpactedPackageAffecting
+from vulnerabilities.models import PackageV2
+
+
+class TestPackagePanelQuerysets(TestCase):
+    def test_ecosystem_distribution_queryset(self):
+        PackageV2.objects.create(type="npm", name="test1", version="1.0.0")
+        PackageV2.objects.create(type="npm", name="test2", version="1.0.0")
+        PackageV2.objects.create(type="pypi", name="test3", version="1.0.0")
+        PackageV2.objects.create(type="gem", name="test4", version="1.0.0")
+
+        tracked_types = ["npm", "pypi"]
+
+        qs = list(ecosystem_distribution_queryset(tracked_types))
+        stats = {item["type"]: item["total_package_count"] for item in qs}
+
+        self.assertEqual(len(stats), 2)  # gem should be excluded
+        self.assertEqual(stats["npm"], 2)
+        self.assertEqual(stats["pypi"], 1)
+
+    def test_top_packages_queryset(self):
+        django_v1 = PackageV2.objects.create(type="pypi", name="django", version="1.0.0")
+        django_v2 = PackageV2.objects.create(type="pypi", name="django", version="2.0.0")
+        flask_v1 = PackageV2.objects.create(type="pypi", name="flask", version="1.0.0")
+
+        def create_impact(advisory_id, package):
+            adv = AdvisoryV2.objects.create(
+                avid=f"github_osv/{advisory_id}",
+                datasource_id="github_osv",
+                unique_content_id=advisory_id,
+                is_latest=True,
+                pipeline_id="github_osv_pipeline",
+                advisory_id=advisory_id,
+                url=f"https://github.com/advisories/{advisory_id}",
+            )
+            impact = ImpactedPackage.objects.create(advisory=adv)
+            ImpactedPackageAffecting.objects.create(impacted_package=impact, package=package)
+
+        create_impact("GHSA-1111", django_v1)
+        create_impact("GHSA-2222", django_v2)
+        create_impact("GHSA-3333", django_v2)
+        create_impact("GHSA-4444", flask_v1)
+
+        qs = list(top_packages_queryset("pypi"))
+
+        self.assertEqual(len(qs), 2)
+        self.assertEqual(qs[0]["name"], "django")
+        self.assertEqual(qs[0]["count"], 3)
+        self.assertEqual(qs[1]["name"], "flask")
+        self.assertEqual(qs[1]["count"], 1)

insights/tests/test_severity_panel.py

@@ -0,0 +1,41 @@
+from django.test import TestCase
+
+from insights.charts.severity_panel import iter_severity_insights
+from vulnerabilities.models import AdvisorySeverity
+from vulnerabilities.models import AdvisoryV2
+
+
+class TestSeverityPanelQuerysets(TestCase):
+    def test_iter_severity_insights(self):
+        advisory1 = 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",
+        )
+        severity1 = AdvisorySeverity.objects.create(scoring_system="cvssv3.1", value="9.8")
+        advisory1.severities.add(severity1)
+
+        advisory2 = AdvisoryV2.objects.create(
+            avid="github_osv/GHSA-2",
+            datasource_id="github_osv",
+            unique_content_id="2",
+            is_latest=True,
+            pipeline_id="github_osv_pipeline",
+            advisory_id="GHSA-2",
+            url="https://example.com/GHSA-2",
+        )
+        severity2 = AdvisorySeverity.objects.create(scoring_system="cvssv3.1", value="4.5")
+        advisory2.severities.add(severity2)
+
+        qs = list(iter_severity_insights())
+
+        self.assertEqual(len(qs), 1)
+        insight = qs[0]
+
+        self.assertEqual(insight.buckets[9], 1)
+        self.assertEqual(insight.buckets[4], 1)
+        self.assertEqual(sum(insight.buckets), 2)

insights/tests/test_views.py

@@ -0,0 +1,45 @@
+from django.test import TestCase
+from django.urls import reverse
+
+from insights.models import DailySnapshot
+from vulnerabilities.models import PackageV2
+
+
+class TestInsightsViews(TestCase):
+    def test_dashboard_redirects_to_default_panel(self):
+        """Root insights URL should redirect to the first panel (overview_panel)."""
+        response = self.client.get("/insights/")
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(response.url.endswith("/insights/overview_panel/"))
+
+    def test_dashboard_invalid_panel_redirects(self):
+        """Invalid panel ID should redirect to default."""
+        response = self.client.get("/insights/non_existent_panel/")
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(response.url.endswith("/insights/overview_panel/"))
+
+    def test_dashboard_valid_panel_loads(self):
+        """Valid panels should load with a 200 OK."""
+        response = self.client.get("/insights/package_panel/")
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "package_panel")
+
+    def test_dashboard_search_package_panel(self):
+        """Test CWE search on package panel."""
+        PackageV2.objects.create(type="pypi", name="django", version="1.0.0")
+        response = self.client.get("/insights/package_panel/?q=pkg:pypi/django")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("search_results_dict", response.context)
+
+    def test_dashboard_search_severity_panel(self):
+        """Test Severity search on severity panel."""
+        PackageV2.objects.create(type="pypi", name="flask", version="1.0.0")
+        response = self.client.get("/insights/severity_panel/?q=pkg:pypi/flask")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("search_results_dict", response.context)
+
+    def test_dashboard_search_no_results(self):
+        """Test search with no matching packages."""
+        response = self.client.get("/insights/severity_panel/?q=pkg:pypi/does_not_exist")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.context["search_error"], "No data available for this query.")