Implement command to detect non-compliant project levels and update score calculation

backend/apps/owasp/management/commands/owasp_aggregate_projects.py

@@ -1,5 +1,6 @@
 """A command to update OWASP projects data."""
 
+from django.core.management import call_command
 from django.core.management.base import BaseCommand
 
 from apps.owasp.models.project import Project
@@ -114,3 +115,10 @@ def handle(self, *_args, **options) -> None:
 
         # Bulk save data.
         Project.bulk_save(projects)
+
+        if offset == 0:
+            self.stdout.write(self.style.NOTICE("Updating project level compliance..."))
+            call_command("owasp_update_project_level_compliance")
+
+            self.stdout.write(self.style.NOTICE("Updating project health scores..."))
+            call_command("owasp_update_project_health_scores")

backend/apps/owasp/management/commands/owasp_update_project_health_scores.py

@@ -5,6 +5,8 @@
 from apps.owasp.models.project_health_metrics import ProjectHealthMetrics
 from apps.owasp.models.project_health_requirements import ProjectHealthRequirements
 
+LEVEL_NON_COMPLIANCE_PENALTY = 10.0
+
 
 class Command(BaseCommand):
     help = "Update OWASP project health scores."
@@ -57,6 +59,12 @@ def handle(self, *args, **options):
                 if int(getattr(metric, field)) <= int(getattr(requirements, field)):
                     score += weight
 
+            # level non-compliance penalty
+            if metric.level_non_compliant:
+                score -= LEVEL_NON_COMPLIANCE_PENALTY
+
+            score = max(score, 0.0)
+
             metric.score = score
             project_health_metrics.append(metric)
 

backend/apps/owasp/management/commands/owasp_update_project_level_compliance.py

@@ -0,0 +1,96 @@
+"""A command to detect non-compliant OWASP project levels."""
+
+import logging
+import re
+from decimal import Decimal, InvalidOperation
+
+import requests
+from django.core.management.base import BaseCommand
+
+from apps.owasp.models import ProjectHealthMetrics
+from apps.owasp.utils.project_level import map_level
+
+logger = logging.getLogger(__name__)
+
+LEVELS_URL = (
+    "https://raw.githubusercontent.com/OWASP/owasp.github.io/main/_data/project_levels.json"
+)
+
+
+def clean_name(name: str) -> str:
+    """Normalize project names for matching with OWASP official data."""
+    name = name.lower().replace("owasp", "")
+    return re.sub(r"[^a-z0-9]+", "", name)
+
+
+class Command(BaseCommand):
+    help = "Update project level compliance."
+
+    def handle(self, *args, **options):
+        self.stdout.write("Updating project level compliance...")
+
+        response = requests.get(LEVELS_URL, timeout=15)
+        response.raise_for_status()
+
+        official = response.json()
+
+        by_repo: dict[str, Decimal] = {}
+        by_name: dict[str, Decimal] = {}
+
+        for item in official:
+            raw_level = item.get("level")
+            try:
+                level = Decimal(str(raw_level))
+            except (InvalidOperation, TypeError, ValueError):
+                logger.debug("Skipping invalid project level: %s", raw_level)
+                continue
+
+            repo = item.get("repo")
+            if repo:
+                by_repo[repo.lower()] = level
+
+            name = item.get("name")
+            if name:
+                by_name[clean_name(name)] = level
+
+        metrics = ProjectHealthMetrics.objects.select_related("project")
+
+        updated_metrics = []
+
+        for metric in metrics:
+            project = metric.project
+
+            official_level = None
+            if project.repo_url:
+                slug = project.repo_url.rstrip("/").split("/")[-1].lower()
+                official_level = by_repo.get(slug)
+
+            if official_level is None:
+                official_level = by_name.get(clean_name(project.name))
+
+            if official_level is None:
+                continue
+
+            expected_level = map_level(official_level)
+            if expected_level is None:
+                continue
+
+            metric.level_non_compliant = project.level != expected_level
+            if metric.level_non_compliant:
+                self.stdout.write(
+                    self.style.WARNING(
+                        f"Level mismatch: {project.name} "
+                        f"local={project.level} "
+                        f"official={expected_level}"
+                    )
+                )
+
+            updated_metrics.append(metric)
+
+        if updated_metrics:
+            ProjectHealthMetrics.bulk_save(
+                updated_metrics,
+                fields=["level_non_compliant"],
+            )
+
+        self.stdout.write(f"Project level compliance updated for {len(updated_metrics)} metrics.")

backend/apps/owasp/migrations/0070_projecthealthmetrics_level_non_compliant.py

@@ -0,0 +1,17 @@
+# Generated by Django 6.0.1 on 2026-01-11 19:03
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("owasp", "0069_alter_project_contribution_data_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="projecthealthmetrics",
+            name="level_non_compliant",
+            field=models.BooleanField(default=False, verbose_name="Is level non-compliant"),
+        ),
+    ]

backend/apps/owasp/models/project_health_metrics.py

@@ -45,6 +45,7 @@ class Meta:
     is_leader_requirements_compliant = models.BooleanField(
         verbose_name="Is leader requirements compliant", default=False
     )
+    level_non_compliant = models.BooleanField(verbose_name="Is level non-compliant", default=False)
     last_released_at = models.DateTimeField(verbose_name="Last released at", blank=True, null=True)
     last_committed_at = models.DateTimeField(
         verbose_name="Last committed at", blank=True, null=True

backend/apps/owasp/utils/project_level.py

@@ -0,0 +1,28 @@
+"""Utilities for mapping OWASP official project levels to Nest ProjectLevel enum."""
+
+from decimal import Decimal, InvalidOperation
+from typing import cast
+
+from apps.owasp.models.enums.project import ProjectLevel
+
+LEVEL_MAP = {
+    Decimal(4): ProjectLevel.FLAGSHIP,
+    Decimal("3.5"): ProjectLevel.FLAGSHIP,
+    Decimal(3): ProjectLevel.PRODUCTION,
+    Decimal(2): ProjectLevel.INCUBATOR,
+    Decimal(1): ProjectLevel.LAB,
+    Decimal(0): ProjectLevel.OTHER,
+}
+
+
+def map_level(level: Decimal) -> ProjectLevel | None:
+    """Map OWASP official numeric level to Nest ProjectLevel."""
+    try:
+        parsed_level = Decimal(str(level))
+    except (InvalidOperation, TypeError, ValueError):
+        return None
+
+    if parsed_level < 0:
+        return None
+
+    return cast("ProjectLevel | None", LEVEL_MAP.get(parsed_level))

backend/tests/apps/owasp/management/commands/owasp_aggregate_projects_test.py

@@ -78,9 +78,18 @@ def test_handle(self, mock_bulk_save, command, mock_project, offset, projects):
         with (
             mock.patch.object(Project, "active_projects", mock_active_projects),
             mock.patch("builtins.print") as mock_print,
+            mock.patch(
+                "apps.owasp.management.commands.owasp_aggregate_projects.call_command"
+            ) as mock_call_command,
         ):
             command.handle(offset=offset)
 
+        if offset == 0:
+            mock_call_command.assert_any_call("owasp_update_project_level_compliance")
+            mock_call_command.assert_any_call("owasp_update_project_health_scores")
+        else:
+            mock_call_command.assert_not_called()
+
         assert mock_bulk_save.called
         assert mock_print.call_count == projects - offset
 

backend/tests/apps/owasp/management/commands/owasp_update_project_health_scores_test.py

@@ -63,6 +63,7 @@ def test_handle_successful_update(self):
         mock_metric.project.name = "Test Project"
         mock_metric.is_funding_requirements_compliant = True
         mock_metric.is_leader_requirements_compliant = True
+        mock_metric.level_non_compliant = False
         self.mock_metrics.return_value.select_related.return_value = [mock_metric]
         self.mock_requirements.return_value = [mock_requirements]
         mock_requirements.level = "test_level"
@@ -79,6 +80,52 @@ def test_handle_successful_update(self):
                 "score",
             ],
         )
-        assert mock_metric.score == EXPECTED_SCORE
+        assert mock_metric.score == pytest.approx(EXPECTED_SCORE)
         assert "Updated project health scores successfully." in self.stdout.getvalue()
         assert "Updating score for project: Test Project" in self.stdout.getvalue()
+
+    def test_non_compliant_level_penalty(self):
+        """Test score penalty for non-compliant project."""
+        fields_weights = {
+            "age_days": (5, 6),
+            "contributors_count": (5, 6),
+            "forks_count": (5, 6),
+            "last_release_days": (5, 6),
+            "last_commit_days": (5, 6),
+            "open_issues_count": (7, 6),
+            "open_pull_requests_count": (5, 6),
+            "owasp_page_last_update_days": (5, 6),
+            "last_pull_request_days": (5, 6),
+            "recent_releases_count": (5, 6),
+            "stars_count": (5, 6),
+            "total_pull_requests_count": (5, 6),
+            "total_releases_count": (5, 6),
+            "unanswered_issues_count": (7, 6),
+            "unassigned_issues_count": (7, 6),
+        }
+
+        # Create mock metric
+        mock_metric = MagicMock(spec=ProjectHealthMetrics)
+        mock_requirements = MagicMock(spec=ProjectHealthRequirements)
+
+        for field, (metric_value, requirement_value) in fields_weights.items():
+            setattr(mock_metric, field, metric_value)
+            setattr(mock_requirements, field, requirement_value)
+
+        mock_metric.project.level = "test_level"
+        mock_metric.project.name = "Penalty Test Project"
+        mock_metric.is_funding_requirements_compliant = True
+        mock_metric.is_leader_requirements_compliant = True
+
+        # Mark project as non-compliant
+        mock_metric.level_non_compliant = True
+
+        self.mock_metrics.return_value.select_related.return_value = [mock_metric]
+        mock_requirements.level = "test_level"
+        self.mock_requirements.return_value = [mock_requirements]
+
+        with patch("sys.stdout", new=self.stdout):
+            call_command("owasp_update_project_health_scores")
+
+        assert mock_metric.score == pytest.approx(EXPECTED_SCORE - 10.0)
+        self.mock_bulk_save.assert_called_once()

backend/tests/apps/owasp/management/commands/owasp_update_project_level_compliance_test.py

@@ -0,0 +1,89 @@
+from unittest.mock import MagicMock, patch
+
+from django.core.management import call_command
+from django.test import SimpleTestCase
+
+from apps.owasp.models.enums.project import ProjectLevel
+
+
+class TestProjectLevelCompliance(SimpleTestCase):
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.bulk_save"
+    )
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.objects"
+    )
+    @patch("apps.owasp.management.commands.owasp_update_project_level_compliance.requests.get")
+    def test_parses_official_level_data(self, mock_get, mock_metrics, mock_bulk_save):
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.json.return_value = [
+            {"name": "Parsed Project", "repo": "parsed-project", "level": "3.5"}
+        ]
+
+        metric = MagicMock()
+        metric.project.name = "Parsed Project"
+        metric.project.level = ProjectLevel.FLAGSHIP
+        metric.level_non_compliant = False
+
+        mock_metrics.select_related.return_value = [metric]
+
+        call_command("owasp_update_project_level_compliance")
+
+        assert metric.level_non_compliant is False
+        mock_bulk_save.assert_called_once()
+
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.bulk_save"
+    )
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.objects"
+    )
+    @patch("apps.owasp.management.commands.owasp_update_project_level_compliance.requests.get")
+    def test_marks_non_compliant_when_levels_differ(self, mock_get, mock_metrics, mock_bulk_save):
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.json.return_value = [
+            {"name": "Test Project", "repo": "test-project", "level": "4"}
+        ]
+
+        metric = MagicMock()
+        metric.project.name = "Test Project"
+        metric.project.level = ProjectLevel.PRODUCTION
+        metric.level_non_compliant = False
+
+        mock_metrics.select_related.return_value = [metric]
+
+        call_command("owasp_update_project_level_compliance")
+
+        assert metric.level_non_compliant is True
+        mock_bulk_save.assert_called_once()
+
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.bulk_save"
+    )
+    @patch(
+        "apps.owasp.management.commands.owasp_update_project_level_compliance."
+        "ProjectHealthMetrics.objects"
+    )
+    @patch("apps.owasp.management.commands.owasp_update_project_level_compliance.requests.get")
+    def test_remains_compliant_when_levels_match(self, mock_get, mock_metrics, mock_bulk_save):
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.json.return_value = [
+            {"name": "Test Project", "repo": "test-project", "level": "3"}
+        ]
+
+        metric = MagicMock()
+        metric.project.name = "Test Project"
+        metric.project.level = ProjectLevel.PRODUCTION
+        metric.level_non_compliant = False
+
+        mock_metrics.select_related.return_value = [metric]
+
+        call_command("owasp_update_project_level_compliance")
+
+        assert metric.level_non_compliant is False
+        mock_bulk_save.assert_called_once()

backend/tests/apps/owasp/utils/project_level_test.py

@@ -0,0 +1,27 @@
+from decimal import Decimal
+
+from django.test import SimpleTestCase
+
+from apps.owasp.models.enums.project import ProjectLevel
+from apps.owasp.utils.project_level import map_level
+
+
+class ProjectLevelMappingTest(SimpleTestCase):
+    def test_flagship_levels(self):
+        assert map_level(Decimal(4)) == ProjectLevel.FLAGSHIP
+        assert map_level(Decimal("3.5")) == ProjectLevel.FLAGSHIP
+
+    def test_production_level(self):
+        assert map_level(Decimal(3)) == ProjectLevel.PRODUCTION
+
+    def test_incubator_level(self):
+        assert map_level(Decimal(2)) == ProjectLevel.INCUBATOR
+
+    def test_lab_level(self):
+        assert map_level(Decimal(1)) == ProjectLevel.LAB
+
+    def test_other_level(self):
+        assert map_level(Decimal(0)) == ProjectLevel.OTHER
+
+    def test_negative_level_returns_none(self):
+        assert map_level(Decimal(-1)) is None