diff --git a/backend/src/apps/owasp/utils/score_calculator.py b/backend/src/apps/owasp/utils/score_calculator.py index 1382873bde..8a4d646b16 100644 --- a/backend/src/apps/owasp/utils/score_calculator.py +++ b/backend/src/apps/owasp/utils/score_calculator.py @@ -383,7 +383,7 @@ def recalculate_user(self, user: User) -> dict[str, str | int | bool]: }, ) - Certificate.issue_certificate(user, total_score, tier) + Certificate.issue_certificate(user, total_score, TierChoices(tier)) logger.info( "Recalculated score for %s: %s points (%s)", diff --git a/backend/tests/unit/apps/owasp/api/internal/nodes/certificate_test.py b/backend/tests/unit/apps/owasp/api/internal/nodes/certificate_test.py new file mode 100644 index 0000000000..b4b5ee3a6b --- /dev/null +++ b/backend/tests/unit/apps/owasp/api/internal/nodes/certificate_test.py @@ -0,0 +1,65 @@ +"""Tests for Certificate GraphQL node.""" + +from unittest.mock import Mock + +from apps.owasp.api.internal.nodes.certificate import CertificateNode +from tests.unit.apps.common.graphql_node_base_test import GraphQLNodeBaseTest + + +class TestCertificateNode(GraphQLNodeBaseTest): + """Test cases for CertificateNode class.""" + + def test_node_fields(self): + """Test node has expected fields.""" + field_names = {field.name for field in CertificateNode.__strawberry_definition__.fields} + expected_field_names = { + "github_user", + "id", + "is_verified", + "issued_at", + "score", + "tier", + } + assert field_names == expected_field_names + + def test_tier_resolver(self): + """Test tier resolver returns human-readable display name.""" + mock_cert = Mock() + mock_cert.get_tier_display.return_value = "Level 1" + + field = self._get_field_by_name("tier", CertificateNode) + result = field.base_resolver.wrapped_func(None, mock_cert) + + assert result == "Level 1" + mock_cert.get_tier_display.assert_called_once() + + def test_is_verified_resolver_active(self): + """Test is_verified returns True when certificate is verified.""" + mock_cert = Mock() + mock_cert.is_verified = True + + field = self._get_field_by_name("is_verified", CertificateNode) + result = field.base_resolver.wrapped_func(None, mock_cert) + + assert result is True + + def test_is_verified_resolver_not_verified(self): + """Test is_verified returns False when certificate is not verified.""" + mock_cert = Mock() + mock_cert.is_verified = False + + field = self._get_field_by_name("is_verified", CertificateNode) + result = field.base_resolver.wrapped_func(None, mock_cert) + + assert result is False + + def test_github_user_resolver(self): + """Test github_user resolver returns the related github_user instance.""" + mock_user = Mock() + mock_cert = Mock() + mock_cert.github_user = mock_user + + field = self._get_field_by_name("github_user", CertificateNode) + result = field.base_resolver.wrapped_func(None, mock_cert) + + assert result == mock_user diff --git a/backend/tests/unit/apps/owasp/api/internal/queries/certificate_test.py b/backend/tests/unit/apps/owasp/api/internal/queries/certificate_test.py new file mode 100644 index 0000000000..1f779a5962 --- /dev/null +++ b/backend/tests/unit/apps/owasp/api/internal/queries/certificate_test.py @@ -0,0 +1,76 @@ +from unittest.mock import MagicMock, patch + +import pytest +from django.core.exceptions import ValidationError + +from apps.owasp.api.internal.queries.certificate import CertificateQuery +from apps.owasp.models.crp.certificate import Certificate + + +class TestCertificateQuery: + """Test suite for CertificateQuery.""" + + def test_has_strawberry_definition(self): + """Test that CertificateQuery has valid Strawberry field definitions.""" + assert hasattr(CertificateQuery, "__strawberry_definition__") + field_names = [field.name for field in CertificateQuery.__strawberry_definition__.fields] + assert "certificate" in field_names + assert "my_certificates" in field_names + + @patch("apps.owasp.models.crp.certificate.Certificate.objects.select_related") + def test_certificate_found(self, mock_select_related): + """Test certificate resolution when certificate exists.""" + mock_cert = MagicMock(spec=Certificate) + mock_select_related.return_value.get.return_value = mock_cert + + result = CertificateQuery().certificate("CERT12345678") + + mock_select_related.assert_called_once_with("github_user") + mock_select_related.return_value.get.assert_called_once_with(id="CERT12345678") + assert result == mock_cert + + @pytest.mark.parametrize( + "exception", + [ + Certificate.DoesNotExist(), + ValidationError("Invalid ID format"), + ValueError("Invalid value"), + ], + ) + @patch("apps.owasp.models.crp.certificate.Certificate.objects.select_related") + def test_certificate_not_found_or_invalid(self, mock_select_related, exception): + """Test certificate resolution returns None when not found or on validation error.""" + mock_select_related.return_value.get.side_effect = exception + + result = CertificateQuery().certificate("INVALID_ID") + + assert result is None + + def test_my_certificates_user_without_github_user(self): + """Test my_certificates returns empty list when user has no github_user.""" + info = MagicMock() + info.context.request.user = MagicMock(spec=[]) # user has no github_user attr + + result = CertificateQuery().my_certificates(info) + + assert result == [] + + @patch("apps.owasp.models.crp.certificate.Certificate.objects.select_related") + def test_my_certificates_returns_active_certificates(self, mock_select_related): + """Test my_certificates returns the user's active certificates.""" + info = MagicMock() + mock_github_user = MagicMock() + info.context.request.user.github_user = mock_github_user + + mock_certs = [MagicMock(spec=Certificate)] + mock_qs = MagicMock() + mock_select_related.return_value = mock_qs + mock_qs.filter.return_value = mock_qs + mock_qs.order_by.return_value = mock_certs + + result = CertificateQuery().my_certificates(info) + + mock_select_related.assert_called_once_with("github_user") + mock_qs.filter.assert_called_once_with(github_user=mock_github_user, is_revoked=False) + mock_qs.order_by.assert_called_once_with("-issued_at") + assert result == mock_certs diff --git a/backend/tests/unit/apps/owasp/management/commands/owasp_crp_recalculate_scores_test.py b/backend/tests/unit/apps/owasp/management/commands/owasp_crp_recalculate_scores_test.py new file mode 100644 index 0000000000..2e0237fe5b --- /dev/null +++ b/backend/tests/unit/apps/owasp/management/commands/owasp_crp_recalculate_scores_test.py @@ -0,0 +1,77 @@ +from io import StringIO +from unittest.mock import MagicMock, patch + +import pytest +from django.core.management import call_command +from django.core.management.base import CommandError + +COMMAND_PATH = "apps.owasp.management.commands.owasp_crp_recalculate_scores" + + +class TestOwaspCrpRecalculateScores: + """Test suite for the owasp_crp_recalculate_scores management command.""" + + @patch(f"{COMMAND_PATH}.ContributionScoreCalculator") + def test_handle_success(self, mock_calculator_class): + """Test successful score recalculation with no failures.""" + mock_calculator = MagicMock() + mock_calculator.recalculate_all.return_value = { + "total": 10, + "created": 3, + "updated": 7, + "failed_count": 0, + } + mock_calculator_class.return_value = mock_calculator + + out = StringIO() + call_command("owasp_crp_recalculate_scores", stdout=out) + + mock_calculator_class.assert_called_once() + mock_calculator.recalculate_all.assert_called_once() + + output = out.getvalue() + assert "Starting score recalculation for all users..." in output + assert "Score recalculation complete:" in output + assert "- Total users: 10" in output + assert "- Created: 3" in output + assert "- Updated: 7" in output + assert "- Failed: 0" in output + + @patch(f"{COMMAND_PATH}.ContributionScoreCalculator") + def test_handle_with_failures(self, mock_calculator_class): + """Test score recalculation when certificate issuance failures occur.""" + mock_calculator = MagicMock() + mock_calculator.recalculate_all.return_value = { + "total": 5, + "created": 1, + "updated": 2, + "failed_count": 2, + "failures": [("alice", "Certificate error"), ("bob", "Network error")], + } + mock_calculator_class.return_value = mock_calculator + + out = StringIO() + with pytest.raises(CommandError, match=r"Failed to issue certificates for 2 user\(s\)"): + call_command("owasp_crp_recalculate_scores", stdout=out) + + output = out.getvalue() + assert "Failed to issue certificates for: alice, bob" in output + + @patch(f"{COMMAND_PATH}.ContributionScoreCalculator") + def test_handle_with_failures_no_failures_list(self, mock_calculator_class): + """Test score recalculation when failed_count > 0 but failures key is missing.""" + mock_calculator = MagicMock() + mock_calculator.recalculate_all.return_value = { + "total": 2, + "created": 0, + "updated": 1, + "failed_count": 1, + } + mock_calculator_class.return_value = mock_calculator + + out = StringIO() + with pytest.raises(CommandError, match=r"Failed to issue certificates for 1 user\(s\)"): + call_command("owasp_crp_recalculate_scores", stdout=out) + + output = out.getvalue() + assert "Failed to issue certificates for: " in output diff --git a/backend/tests/unit/apps/owasp/models/crp/__init__.py b/backend/tests/unit/apps/owasp/models/crp/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/tests/unit/apps/owasp/models/crp/certificate_test.py b/backend/tests/unit/apps/owasp/models/crp/certificate_test.py new file mode 100644 index 0000000000..cd0c5d9920 --- /dev/null +++ b/backend/tests/unit/apps/owasp/models/crp/certificate_test.py @@ -0,0 +1,114 @@ +from unittest.mock import MagicMock, patch + +import pytest + +from apps.github.models.user import User +from apps.owasp.exceptions import CertificateIssuanceError +from apps.owasp.models.crp.certificate import ( + CERTIFICATE_ID_ALPHABET, + CERTIFICATE_ID_LENGTH, + Certificate, + generate_certificate_id, +) +from apps.owasp.models.crp.recognition_enums import TierChoices + +MODEL_PATH = "apps.owasp.models.crp.certificate" + + +class TestCertificateModel: + """Test suite for Certificate model.""" + + def test_generate_certificate_id(self): + """Test generate_certificate_id produces a 12-char string from ALPHABET.""" + cert_id = generate_certificate_id() + assert len(cert_id) == CERTIFICATE_ID_LENGTH + assert all(c in CERTIFICATE_ID_ALPHABET for c in cert_id) + + def test_str_representation_active(self): + """Test __str__ for active certificate.""" + user = User(login="john_doe") + cert = Certificate(github_user=user, tier=TierChoices.LEVEL_1, is_revoked=False) + + assert str(cert) == "john_doe - LEVEL_1 Certificate (Active)" + + def test_str_representation_revoked(self): + """Test __str__ for revoked certificate.""" + user = User(login="jane_doe") + cert = Certificate(github_user=user, tier=TierChoices.LEVEL_2, is_revoked=True) + + assert str(cert) == "jane_doe - LEVEL_2 Certificate (Revoked)" + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch(f"{MODEL_PATH}.Certificate.objects") + @patch(f"{MODEL_PATH}.User.objects") + def test_issue_certificate_already_exists( + self, mock_user_objects, mock_cert_objects, mock_exit, mock_enter + ): + """Test issue_certificate returns early if active certificate already exists.""" + user = User(id=1, login="john_doe") + mock_user_objects.select_for_update.return_value.get.return_value = user + mock_cert_objects.filter.return_value.exists.return_value = True + + Certificate.issue_certificate(user, 150, TierChoices.LEVEL_2) + + mock_cert_objects.filter.assert_called_once_with( + github_user=user, tier=TierChoices.LEVEL_2, is_revoked=False + ) + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch(f"{MODEL_PATH}.Certificate.objects") + @patch(f"{MODEL_PATH}.CertificateProviderFactory") + @patch(f"{MODEL_PATH}.User.objects") + def test_issue_certificate_provider_resolution_error( + self, mock_user_objects, mock_factory, mock_cert_objects, mock_exit, mock_enter + ): + """Test issue_certificate raises on provider resolution error.""" + user = User(id=1, login="john_doe") + mock_user_objects.select_for_update.return_value.get.return_value = user + mock_cert_objects.filter.return_value.exists.return_value = False + mock_factory.get_provider.side_effect = ValueError("Unknown provider") + + with pytest.raises(CertificateIssuanceError): + Certificate.issue_certificate(user, 150, TierChoices.LEVEL_2) + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch(f"{MODEL_PATH}.Certificate.objects") + @patch(f"{MODEL_PATH}.CertificateProviderFactory") + @patch(f"{MODEL_PATH}.User.objects") + def test_issue_certificate_provider_issuance_exception( + self, mock_user_objects, mock_factory, mock_cert_objects, mock_exit, mock_enter + ): + """Test issue_certificate raises when provider issuance fails.""" + user = User(id=1, login="test_user") + mock_user_objects.select_for_update.return_value.get.return_value = user + mock_cert_objects.filter.return_value.exists.return_value = False + + mock_provider = MagicMock() + mock_provider.issue_certificate.side_effect = RuntimeError("PDF generation failed") + mock_factory.get_provider.return_value = mock_provider + + with pytest.raises(CertificateIssuanceError): + Certificate.issue_certificate(user, 150, TierChoices.LEVEL_2) + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch(f"{MODEL_PATH}.Certificate.objects") + @patch(f"{MODEL_PATH}.CertificateProviderFactory") + @patch(f"{MODEL_PATH}.User.objects") + def test_issue_certificate_success( + self, mock_user_objects, mock_factory, mock_cert_objects, mock_exit, mock_enter + ): + """Test successful certificate issuance.""" + user = User(id=1, login="test_user") + mock_user_objects.select_for_update.return_value.get.return_value = user + mock_cert_objects.filter.return_value.exists.return_value = False + + mock_provider = MagicMock() + mock_factory.get_provider.return_value = mock_provider + + Certificate.issue_certificate(user, 150, TierChoices.LEVEL_2) + + mock_provider.issue_certificate.assert_called_once_with(user, 150, TierChoices.LEVEL_2) diff --git a/backend/tests/unit/apps/owasp/models/crp/contribution_score_test.py b/backend/tests/unit/apps/owasp/models/crp/contribution_score_test.py new file mode 100644 index 0000000000..14038b777b --- /dev/null +++ b/backend/tests/unit/apps/owasp/models/crp/contribution_score_test.py @@ -0,0 +1,14 @@ +from apps.github.models.user import User +from apps.owasp.models.crp.contribution_score import ContributionScore +from apps.owasp.models.crp.recognition_enums import TierChoices + + +class TestContributionScoreModel: + """Test suite for ContributionScore model.""" + + def test_str_representation(self): + """Test __str__ for ContributionScore.""" + user = User(login="alice_dev") + score = ContributionScore(github_user=user, tier=TierChoices.LEVEL_3, value=350) + + assert str(score) == "alice_dev - LEVEL_3 (350 points)" diff --git a/backend/tests/unit/apps/owasp/models/crp/leaderboard_snapshot_test.py b/backend/tests/unit/apps/owasp/models/crp/leaderboard_snapshot_test.py new file mode 100644 index 0000000000..a586829738 --- /dev/null +++ b/backend/tests/unit/apps/owasp/models/crp/leaderboard_snapshot_test.py @@ -0,0 +1,24 @@ +from datetime import date + +from apps.github.models.user import User +from apps.owasp.models.crp.leaderboard_snapshot import LeaderboardSnapshot + + +class TestLeaderboardSnapshotModel: + """Test suite for LeaderboardSnapshot model.""" + + def test_str_representation(self): + """Test __str__ for LeaderboardSnapshot.""" + user = User(login="bob_coder") + snapshot = LeaderboardSnapshot( + github_user=user, + global_rank=5, + project_rank=2, + chapter_rank=1, + snapshot_date=date(2026, 8, 1), + ) + + assert ( + str(snapshot) + == "bob_coder - Global Rank: 5, Project Rank: 2, Chapter Rank: 1 (2026-08-01)" + ) diff --git a/backend/tests/unit/apps/owasp/models/crp/scoring_weight_test.py b/backend/tests/unit/apps/owasp/models/crp/scoring_weight_test.py new file mode 100644 index 0000000000..d5be4623e9 --- /dev/null +++ b/backend/tests/unit/apps/owasp/models/crp/scoring_weight_test.py @@ -0,0 +1,11 @@ +from apps.owasp.models.crp.recognition_enums import EventTypeChoices +from apps.owasp.models.crp.scoring_weight import ScoringWeight + + +class TestScoringWeightModel: + """Test suite for ScoringWeight model.""" + + def test_str_representation(self): + """Test __str__ for ScoringWeight using real choice display.""" + weight = ScoringWeight(event_type=EventTypeChoices.PR_MERGED, score=25) + assert str(weight) == "Pull Request Merged - 25 points" diff --git a/backend/tests/unit/apps/owasp/utils/certificate_provider_test.py b/backend/tests/unit/apps/owasp/utils/certificate_provider_test.py new file mode 100644 index 0000000000..d90684a86d --- /dev/null +++ b/backend/tests/unit/apps/owasp/utils/certificate_provider_test.py @@ -0,0 +1,48 @@ +from unittest.mock import MagicMock, patch + +import pytest + +from apps.owasp.models.crp.recognition_enums import TierChoices +from apps.owasp.utils.certificate_provider import ( + CertificateProviderFactory, + LocalCertificateProvider, +) + + +class TestLocalCertificateProvider: + """Test suite for LocalCertificateProvider.""" + + @patch("apps.owasp.models.crp.certificate.Certificate.objects.create") + def test_issue_certificate_creates_record(self, mock_create): + """Test issue_certificate creates a Certificate record in local DB.""" + mock_user = MagicMock() + provider = LocalCertificateProvider() + + provider.issue_certificate(mock_user, 200, TierChoices.LEVEL_2) + + mock_create.assert_called_once_with( + github_user=mock_user, + score=200, + tier=TierChoices.LEVEL_2, + ) + + +class TestCertificateProviderFactory: + """Test suite for CertificateProviderFactory.""" + + @patch("apps.owasp.utils.certificate_provider.settings") + def test_get_provider_local_success(self, mock_settings): + """Test get_provider returns LocalCertificateProvider when setting is 'local'.""" + mock_settings.CERTIFICATE_PROVIDER = "local" + + provider = CertificateProviderFactory.get_provider() + + assert isinstance(provider, LocalCertificateProvider) + + @patch("apps.owasp.utils.certificate_provider.settings") + def test_get_provider_unknown_raises_value_error(self, mock_settings): + """Test get_provider raises ValueError for unknown provider type.""" + mock_settings.CERTIFICATE_PROVIDER = "invalid_provider" + + with pytest.raises(ValueError, match="Unknown certificate provider: invalid_provider"): + CertificateProviderFactory.get_provider() diff --git a/backend/tests/unit/apps/owasp/utils/score_calculator_test.py b/backend/tests/unit/apps/owasp/utils/score_calculator_test.py new file mode 100644 index 0000000000..6266c015a2 --- /dev/null +++ b/backend/tests/unit/apps/owasp/utils/score_calculator_test.py @@ -0,0 +1,425 @@ +from datetime import date +from unittest.mock import MagicMock, patch + +import pytest + +from apps.github.models.user import User +from apps.owasp.exceptions import CertificateIssuanceError +from apps.owasp.models.crp.contribution_score import ContributionScore +from apps.owasp.models.crp.recognition_enums import TierChoices +from apps.owasp.utils.score_calculator import ContributionScoreCalculator + +CALCULATOR_PATH = "apps.owasp.utils.score_calculator" + + +class TestContributionScoreCalculator: + """Test suite for ContributionScoreCalculator.""" + + def _mock_recalculate_all_querysets(self, mock_user_class, mock_pr, mock_issue, users): + """Set up mock querysets for User, PullRequest, and Issue for recalculate_all tests.""" + mock_users_qs = MagicMock() + mock_users_qs.count.return_value = len(users) + mock_users_qs.__iter__.return_value = iter(users) + mock_users_qs.distinct.return_value.prefetch_related.return_value = mock_users_qs + mock_user_class.objects.filter.return_value = mock_users_qs + + pr_chain = mock_pr.objects.filter.return_value.values.return_value.annotate.return_value + pr_chain.values_list.return_value = [] + issue_chain = ( + mock_issue.objects.filter.return_value.values.return_value.annotate.return_value + ) + issue_chain.values_list.return_value = [] + + @patch(f"{CALCULATOR_PATH}.ScoringWeight") + def test_load_scoring_weights(self, mock_scoring_weight): + """Test load_scoring_weights retrieves active scoring weights from database.""" + w1 = MagicMock(event_type="pr_merged", score=20) + w2 = MagicMock(event_type="pr_opened", score=5) + mock_scoring_weight.objects.filter.return_value = [w1, w2] + + calc = ContributionScoreCalculator() + + mock_scoring_weight.objects.filter.assert_called_once_with(is_active=True) + assert calc.scoring_weights == {"pr_merged": 20, "pr_opened": 5} + + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 20, "pr_opened": 5}, + ) + def test_calculate_score(self, mock_load): + """Test calculate_score correctly computes breakdown and total score.""" + calc = ContributionScoreCalculator() + counts = {"pr_merged": 3, "pr_opened": 2, "unknown_event": 5} + + total_score, breakdown = calc.calculate_score(counts) + + assert total_score == 70 # (3 * 20) + (2 * 5) + (5 * 0) + assert breakdown == {"pr_merged": 60, "pr_opened": 10, "unknown_event": 0} + + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 20}, + ) + @patch.object(ContributionScoreCalculator, "get_contribution_breakdown") + def test_calculate(self, mock_breakdown, mock_load): + """Test calculate delegates to get_contribution_breakdown and sums the result.""" + user = User(login="test_user") + mock_breakdown.return_value = {"pr_merged": 40, "issue_completed": 10} + + calc = ContributionScoreCalculator() + start = date(2026, 1, 1) + end = date(2026, 6, 1) + total, breakdown = calc.calculate(user, start_date=start, end_date=end) + + assert total == 50 + assert breakdown == {"pr_merged": 40, "issue_completed": 10} + mock_breakdown.assert_called_once_with(user, start, end) + + @pytest.mark.parametrize( + ("score", "expected_tier"), + [ + (600, "level_4"), + (500, "level_4"), + (300, "level_3"), + (250, "level_3"), + (150, "level_2"), + (100, "level_2"), + (50, "level_1"), + (0, "level_1"), + ], + ) + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + def test_get_tier(self, mock_load, score, expected_tier): + """Test get_tier maps score thresholds to tier strings.""" + calc = ContributionScoreCalculator() + assert calc.get_tier(score) == expected_tier + + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + def test_get_tier_negative_score(self, mock_load): + """Test get_tier returns level_1 for negative scores.""" + calc = ContributionScoreCalculator() + assert calc.get_tier(-10) == "level_1" + + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_count_merged_pull_requests(self, mock_pr, mock_load): + """Test count_merged_pull_requests with and without date range.""" + user = User(login="test_user") + mock_qs = MagicMock() + mock_pr.objects.filter.return_value = mock_qs + mock_qs.filter.return_value = mock_qs + mock_qs.count.return_value = 5 + + calc = ContributionScoreCalculator() + + # Without date range + count = calc.count_merged_pull_requests(user) + assert count == 5 + mock_pr.objects.filter.assert_called_once_with( + author=user, + merged_at__isnull=False, + repository__is_fork=False, + repository__organization__is_owasp_related_organization=True, + ) + + # With date range + start = date(2026, 1, 1) + end = date(2026, 2, 1) + calc.count_merged_pull_requests(user, start_date=start, end_date=end) + assert mock_qs.filter.call_count == 2 + + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_count_opened_pull_requests(self, mock_pr, mock_load): + """Test count_opened_pull_requests with and without date range.""" + user = User(login="test_user") + mock_qs = MagicMock() + mock_pr.objects.filter.return_value = mock_qs + mock_qs.filter.return_value = mock_qs + mock_qs.count.return_value = 3 + + calc = ContributionScoreCalculator() + + # Without date range + count = calc.count_opened_pull_requests(user) + assert count == 3 + + # With date range + start = date(2026, 1, 1) + end = date(2026, 2, 1) + calc.count_opened_pull_requests(user, start_date=start, end_date=end) + assert mock_qs.filter.call_count == 2 + + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + @patch(f"{CALCULATOR_PATH}.Issue") + def test_count_completed_issues(self, mock_issue, mock_load): + """Test count_completed_issues with and without date range.""" + user = User(login="test_user") + mock_qs = MagicMock() + mock_issue.objects.filter.return_value = mock_qs + mock_qs.filter.return_value = mock_qs + mock_qs.count.return_value = 4 + + calc = ContributionScoreCalculator() + + # Without date range + count = calc.count_completed_issues(user) + assert count == 4 + + # With date range + start = date(2026, 1, 1) + end = date(2026, 2, 1) + calc.count_completed_issues(user, start_date=start, end_date=end) + assert mock_qs.filter.call_count == 2 + + @patch.object(ContributionScoreCalculator, "load_scoring_weights", return_value={}) + @patch.object(ContributionScoreCalculator, "count_merged_pull_requests", return_value=2) + @patch.object(ContributionScoreCalculator, "count_opened_pull_requests", return_value=1) + @patch.object(ContributionScoreCalculator, "count_completed_issues", return_value=3) + @patch.object( + ContributionScoreCalculator, + "calculate_score", + return_value=(100, {"pr_merged": 40}), + ) + def test_get_contribution_breakdown( + self, mock_calc_score, mock_issues, mock_opened, mock_merged, mock_load + ): + """Test get_contribution_breakdown gathers counts and calls calculate_score.""" + user = User(login="test_user") + calc = ContributionScoreCalculator() + + breakdown = calc.get_contribution_breakdown(user) + + assert breakdown == {"pr_merged": 40} + mock_calc_score.assert_called_once_with( + {"pr_merged": 2, "pr_opened": 1, "issue_completed": 3} + ) + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.BulkSaveModel") + @patch(f"{CALCULATOR_PATH}.User") + @patch(f"{CALCULATOR_PATH}.Issue") + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_recalculate_all_updates_and_creates_scores( + self, + mock_pr, + mock_issue, + mock_user_class, + mock_bulk_save_model, + mock_cert_class, + mock_load, + mock_exit, + mock_enter, + ): + """Test recalculate_all processes users, updates/creates scores, and issues certs.""" + user1 = User(login="user1") + existing_score = ContributionScore(github_user=user1, value=10, tier="level_1") + user1.contribution_score = existing_score + user2 = User(login="user2") + + self._mock_recalculate_all_querysets(mock_user_class, mock_pr, mock_issue, [user1, user2]) + + calc = ContributionScoreCalculator() + res = calc.recalculate_all() + + assert res["total"] == 2 + assert res["created"] == 1 + assert res["updated"] == 1 + assert res["failed_count"] == 0 + + mock_bulk_save_model.bulk_save.assert_called_once() + assert mock_cert_class.issue_certificate.call_count == 2 + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.BulkSaveModel") + @patch(f"{CALCULATOR_PATH}.User") + @patch(f"{CALCULATOR_PATH}.Issue") + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_recalculate_all_batching( + self, + mock_pr, + mock_issue, + mock_user_class, + mock_bulk_save_model, + mock_cert_class, + mock_load, + mock_exit, + mock_enter, + ): + """Test recalculate_all bulk saves when batch size limit is reached.""" + user1 = User(login="batch_user1") + user2 = User(login="batch_user2") + + self._mock_recalculate_all_querysets(mock_user_class, mock_pr, mock_issue, [user1, user2]) + + calc = ContributionScoreCalculator() + calc.BATCH_SIZE = 2 + res = calc.recalculate_all() + + assert res["total"] == 2 + assert res["created"] == 2 + mock_bulk_save_model.bulk_save.assert_called_once() + assert mock_cert_class.issue_certificate.call_count == 2 + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.BulkSaveModel") + @patch(f"{CALCULATOR_PATH}.User") + @patch(f"{CALCULATOR_PATH}.Issue") + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_recalculate_all_batching_certificate_failures( + self, + mock_pr, + mock_issue, + mock_user_class, + mock_bulk_save_model, + mock_cert_class, + mock_load, + mock_exit, + mock_enter, + ): + """Test recalculate_all error handling when batch size limit is reached.""" + user1 = User(login="batch_fail1") + user2 = User(login="batch_fail2") + + self._mock_recalculate_all_querysets(mock_user_class, mock_pr, mock_issue, [user1, user2]) + + mock_cert_class.issue_certificate.side_effect = [ + CertificateIssuanceError("Batch issue 1"), + CertificateIssuanceError("Batch issue 2"), + ] + + calc = ContributionScoreCalculator() + calc.BATCH_SIZE = 1 + res = calc.recalculate_all() + + assert res["total"] == 2 + assert res["failed_count"] == 2 + assert res["failures"][0][0] == "batch_fail1" + assert res["failures"][1][0] == "batch_fail2" + + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.BulkSaveModel") + @patch(f"{CALCULATOR_PATH}.User") + @patch(f"{CALCULATOR_PATH}.Issue") + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_recalculate_all_no_users( + self, + mock_pr, + mock_issue, + mock_user_class, + mock_bulk_save_model, + mock_cert_class, + mock_load, + ): + """Test recalculate_all when no users have contributions.""" + self._mock_recalculate_all_querysets(mock_user_class, mock_pr, mock_issue, []) + + calc = ContributionScoreCalculator() + res = calc.recalculate_all() + + assert res["total"] == 0 + assert res["created"] == 0 + assert res["updated"] == 0 + mock_bulk_save_model.bulk_save.assert_not_called() + mock_cert_class.issue_certificate.assert_not_called() + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.BulkSaveModel") + @patch(f"{CALCULATOR_PATH}.User") + @patch(f"{CALCULATOR_PATH}.Issue") + @patch(f"{CALCULATOR_PATH}.PullRequest") + def test_recalculate_all_handles_certificate_failures( + self, + mock_pr, + mock_issue, + mock_user_class, + mock_bulk_save_model, + mock_cert_class, + mock_load, + mock_exit, + mock_enter, + ): + """Test recalculate_all records certificate issuance errors.""" + user1 = User(login="failing_user") + existing_score = ContributionScore(github_user=user1, value=10, tier="level_1") + user1.contribution_score = existing_score + + self._mock_recalculate_all_querysets(mock_user_class, mock_pr, mock_issue, [user1]) + + mock_cert_class.issue_certificate.side_effect = CertificateIssuanceError("Issuance failed") + + calc = ContributionScoreCalculator() + res = calc.recalculate_all() + + assert res["total"] == 1 + assert res["failed_count"] == 1 + assert res["failures"][0][0] == "failing_user" + + @patch("django.db.transaction.Atomic.__enter__", return_value=None) + @patch("django.db.transaction.Atomic.__exit__", return_value=None) + @patch.object( + ContributionScoreCalculator, + "load_scoring_weights", + return_value={"pr_merged": 50}, + ) + @patch(f"{CALCULATOR_PATH}.Certificate") + @patch(f"{CALCULATOR_PATH}.ContributionScore.objects") + def test_recalculate_user( + self, mock_contrib_score_objects, mock_cert_class, mock_load, mock_exit, mock_enter + ): + """Test recalculate_user for a single user.""" + user = User(login="single_user") + + calc = ContributionScoreCalculator() + with patch.object(calc, "calculate", return_value=(150, {"pr_merged": 150})): + mock_contrib_score_objects.update_or_create.return_value = (MagicMock(), True) + + result = calc.recalculate_user(user) + + assert result == { + "total_score": 150, + "tier": "level_2", + "created": True, + } + mock_contrib_score_objects.update_or_create.assert_called_once_with( + github_user=user, + defaults={"value": 150, "tier": "level_2"}, + ) + mock_cert_class.issue_certificate.assert_called_once_with( + user, 150, TierChoices.LEVEL_2 + )