apps/progress/tests/test_skill_upgrade.py

"""Unit tests for the skill-upgrade / closed-learning-loop service.

See research/08-next-modules-build-plan.md §Module B.
"""
import pytest
from django.contrib.auth import get_user_model
from django.utils import timezone
from rest_framework.test import APIClient

from apps.progress.models import UserProgress
from apps.progress.services import (
    _suggested_proficiency,
    apply_upgrade,
    compute_upgrade_suggestions,
    dismiss_upgrade,
)
from apps.resources.models import Resource, SkillResource
from apps.skills.models import Skill, UserSkill

User = get_user_model()
pytestmark = pytest.mark.django_db


UPGRADE_LIST = '/api/progress/upgrade-suggestions/'


@pytest.fixture
def user():
    return User.objects.create_user(
        username='upg@x.com', email='upg@x.com', password='pw', name='U',
    )


@pytest.fixture
def auth_client(user):
    c = APIClient()
    c.force_authenticate(user=user)
    return c


def _skill(name='Python', category='Programming'):
    return Skill.objects.create(
        skill_name=name, category=category, difficulty_level='BEGINNER',
    )


def _resource(title='ML Course', r_type='COURSE', url=None):
    return Resource.objects.create(
        title=title, provider='X', url=url or f'https://e.com/{title}',
        difficulty_level='INTERMEDIATE', duration=100, type=r_type, rating=4.0,
    )


def _complete(user, resource, when=None):
    up = UserProgress.objects.create(
        user=user, resource=resource,
        status='COMPLETED',
        progress=100,
        completed_at=when or timezone.now(),
    )
    return up


def _set_skill(user, skill, proficiency, level='BEGINNER',
               dismissed_upgrade_at=None):
    return UserSkill.objects.create(
        user=user, skill=skill, proficiency=proficiency, user_level=level,
        dismissed_upgrade_at=dismissed_upgrade_at,
    )


class TestBumpMap:
    def test_zero_to_60(self):
        assert _suggested_proficiency(0) == 60

    def test_30_to_60(self):
        assert _suggested_proficiency(30) == 60

    def test_50_to_70(self):
        assert _suggested_proficiency(50) == 70

    def test_70_to_85(self):
        assert _suggested_proficiency(70) == 85

    def test_90_to_100(self):
        assert _suggested_proficiency(90) == 100

    def test_100_suppressed(self):
        assert _suggested_proficiency(100) is None


class TestComputeSuggestions:
    def test_completed_resource_creates_suggestion_when_user_skill_absent(
            self, user):
        ml = _skill('Machine Learning')
        r = _resource()
        SkillResource.objects.create(skill=ml, resource=r, relevance_score=1.0)
        _complete(user, r)

        suggestions = compute_upgrade_suggestions(user)
        assert len(suggestions) == 1
        s = suggestions[0]
        assert s['skill_id'] == ml.id
        assert s['current_proficiency'] == 0
        assert s['suggested_proficiency'] == 60
        assert s['suggested_level'] == 'INTERMEDIATE'

    def test_only_completed_status_triggers(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        UserProgress.objects.create(
            user=user, resource=r, status='IN_PROGRESS', progress=50,
        )
        assert compute_upgrade_suggestions(user) == []

    def test_suggestion_suppressed_at_100(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _set_skill(user, py, proficiency=100, level='ADVANCED')
        _complete(user, r)
        assert compute_upgrade_suggestions(user) == []

    def test_never_downgrade(self, user):
        """User at 85 → bump map suggests 100. But if current >= target,
        suppress. (Table avoids downgrade by design; we also verify the
        never-downgrade guard.)"""
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _set_skill(user, py, proficiency=95, level='ADVANCED')
        _complete(user, r)
        suggestions = compute_upgrade_suggestions(user)
        assert len(suggestions) == 1
        assert suggestions[0]['suggested_proficiency'] == 100

    def test_multiple_skills_linked_to_one_resource(self, user):
        py = _skill('Python')
        ml = _skill('Machine Learning')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r, relevance_score=0.7)
        SkillResource.objects.create(skill=ml, resource=r, relevance_score=1.0)
        _complete(user, r)
        suggestions = compute_upgrade_suggestions(user)
        assert {s['skill_id'] for s in suggestions} == {py.id, ml.id}

    def test_dismissal_hides_suggestion(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        # Completion first, then dismissal AFTER → suppresses.
        _complete(user, r, when=timezone.now() - timezone.timedelta(days=2))
        _set_skill(user, py, proficiency=0,
                   dismissed_upgrade_at=timezone.now())
        assert compute_upgrade_suggestions(user) == []

    def test_dismissal_self_heals_on_newer_completion(self, user):
        """Dismissing a prior suggestion does not block a later completion
        from resurfacing the suggestion."""
        py = _skill('Python')
        r1 = _resource(title='First', url='https://e.com/first')
        r2 = _resource(title='Second', url='https://e.com/second')
        SkillResource.objects.create(skill=py, resource=r1)
        SkillResource.objects.create(skill=py, resource=r2)

        earlier = timezone.now() - timezone.timedelta(days=5)
        _complete(user, r1, when=earlier)
        _set_skill(user, py, proficiency=0,
                   dismissed_upgrade_at=timezone.now()
                   - timezone.timedelta(days=2))

        # A NEW completion post-dismissal re-surfaces the suggestion.
        _complete(user, r2, when=timezone.now())

        suggestions = compute_upgrade_suggestions(user)
        assert len(suggestions) == 1
        assert suggestions[0]['skill_id'] == py.id


class TestApply:
    def test_apply_creates_user_skill_if_absent(self, user):
        ml = _skill('ML')
        r = _resource()
        SkillResource.objects.create(skill=ml, resource=r)
        _complete(user, r)

        result = apply_upgrade(user, ml.id)
        assert result['applied'] is True
        us = UserSkill.objects.get(user=user, skill=ml)
        assert us.proficiency == 60
        assert us.user_level == 'INTERMEDIATE'

    def test_apply_bumps_proficiency_and_recomputes_level(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _set_skill(user, py, proficiency=70, level='ADVANCED')
        _complete(user, r)

        result = apply_upgrade(user, py.id)
        assert result['applied'] is True
        us = UserSkill.objects.get(user=user, skill=py)
        assert us.proficiency == 85
        # proficiency_to_level(85) → ADVANCED (since > 60)
        assert us.user_level == 'ADVANCED'

    def test_apply_is_idempotent(self, user):
        """Repeated applies without a new completion must be no-ops.
        One completion → one rung of the bump ladder, then already_applied.
        """
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _complete(user, r)

        first = apply_upgrade(user, py.id)
        assert first['applied'] is True

        second = apply_upgrade(user, py.id)
        assert second['applied'] is False
        assert second['already_applied'] is True

        us = UserSkill.objects.get(user=user, skill=py)
        assert us.proficiency == 60  # did not climb past one rung

    def test_apply_resurfaces_on_new_completion(self, user):
        """A completion after the last apply unlocks the next bump."""
        py = _skill('Python')
        r1 = _resource(title='First', url='https://e.com/first')
        r2 = _resource(title='Second', url='https://e.com/second')
        SkillResource.objects.create(skill=py, resource=r1)
        SkillResource.objects.create(skill=py, resource=r2)

        _complete(user, r1, when=timezone.now() - timezone.timedelta(days=2))
        first = apply_upgrade(user, py.id)
        assert first['applied'] is True
        assert UserSkill.objects.get(user=user, skill=py).proficiency == 60

        _complete(user, r2, when=timezone.now())  # newer than the cursor
        second = apply_upgrade(user, py.id)
        assert second['applied'] is True
        assert UserSkill.objects.get(user=user, skill=py).proficiency == 85

    def test_apply_when_no_completion_is_a_noop(self, user):
        py = _skill('Python')
        result = apply_upgrade(user, py.id)
        assert result['applied'] is False
        assert result['already_applied'] is False

    def test_apply_when_already_at_100_is_already_applied(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _set_skill(user, py, proficiency=100, level='ADVANCED')
        _complete(user, r)

        result = apply_upgrade(user, py.id)
        assert result['applied'] is False
        assert result['already_applied'] is True

    def test_apply_advances_dismissal_cursor(self, user):
        """After an apply, dismissed_upgrade_at is advanced to the latest
        completion timestamp. Semantically this means the suggestion is
        suppressed until a newer completion arrives — matching the dismissed
        flow. A prior user-triggered dismissal must not block an apply that is
        acting on a newer completion."""
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        earlier = timezone.now() - timezone.timedelta(days=2)
        _set_skill(user, py, proficiency=0, dismissed_upgrade_at=earlier)
        completion_at = timezone.now()
        _complete(user, r, when=completion_at)

        result = apply_upgrade(user, py.id)
        assert result['applied'] is True
        us = UserSkill.objects.get(user=user, skill=py)
        assert us.dismissed_upgrade_at == completion_at

        # Functional check: the applied suggestion doesn't resurface on a
        # re-compute without a new completion.
        assert compute_upgrade_suggestions(user) == []


class TestDismiss:
    def test_dismiss_sets_timestamp_and_hides(self, user):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _complete(user, r)
        _set_skill(user, py, proficiency=0)

        result = dismiss_upgrade(user, py.id)
        assert result['dismissed'] is True
        us = UserSkill.objects.get(user=user, skill=py)
        assert us.dismissed_upgrade_at is not None
        assert compute_upgrade_suggestions(user) == []

    def test_dismiss_creates_user_skill_if_absent(self, user):
        py = _skill('Python')
        dismiss_upgrade(user, py.id)
        us = UserSkill.objects.get(user=user, skill=py)
        assert us.proficiency == 0
        assert us.dismissed_upgrade_at is not None


class TestEndpoints:
    def test_list_endpoint(self, user, auth_client):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _complete(user, r)

        r_resp = auth_client.get(UPGRADE_LIST)
        assert r_resp.status_code == 200
        assert r_resp.data['suggestions'][0]['skill_id'] == py.id

    def test_apply_endpoint(self, user, auth_client):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _complete(user, r)

        r_resp = auth_client.post(f'{UPGRADE_LIST}{py.id}/apply/')
        assert r_resp.status_code == 200
        assert r_resp.data['applied'] is True
        assert UserSkill.objects.get(user=user, skill=py).proficiency == 60

    def test_dismiss_endpoint(self, user, auth_client):
        py = _skill('Python')
        r = _resource()
        SkillResource.objects.create(skill=py, resource=r)
        _complete(user, r)

        r_resp = auth_client.post(f'{UPGRADE_LIST}{py.id}/dismiss/')
        assert r_resp.status_code == 200
        assert r_resp.data['dismissed'] is True

        # List should now be empty.
        r_resp = auth_client.get(UPGRADE_LIST)
        assert r_resp.data['suggestions'] == []

    def test_apply_unknown_skill_404(self, auth_client):
        r_resp = auth_client.post(f'{UPGRADE_LIST}99999/apply/')
        assert r_resp.status_code == 404

    def test_requires_auth(self):
        c = APIClient()
        assert c.get(UPGRADE_LIST).status_code == 401