apps/analysis/tests/test_recommendations_stress.py

"""Stress / adversarial regression tests for the recommendation ranker.

Pins the level-tier guarantee (c808446) and the level-aware anchor (6ac0e63)
across the full proficiency range, at scale, and under hostile endpoint input.
These complement the behavioural tests in test_recommendations.py — here we
sweep boundaries and verify invariants hold rather than spot-check cases.

hypothesis is not installed, so the sweeps are parametrized explicitly.
"""
import pytest
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient

from apps.analysis.services import (
    DIFFICULTY_ORDER,
    MAX_LIMIT_PER_SKILL,
    NEAR_MAX_DISTANCE,
    _score,
    compute_recommendations,
)
from apps.resources.models import Resource, SkillResource
from apps.roles.models import Role, RoleSkill, UserTargetRole
from apps.skills.models import Skill, UserSkill
from apps.skills.utils import proficiency_to_level

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

RECS_URL = '/api/recommendations/'


@pytest.fixture
def user():
    return User.objects.create_user(
        username='stress@example.com', email='stress@example.com',
        password='password123', name='Stress Tester',
    )


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


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


def _role_with(skill_specs, name_hint='Role'):
    role = Role.objects.create(
        role_name=f'{name_hint}-{id(skill_specs)}',
        industry='Tech', is_active=True,
    )
    for name, level, weight, mandatory in skill_specs:
        RoleSkill.objects.create(
            role=role, skill=_skill(name),
            required_level=level, weight=weight, is_mandatory=mandatory,
        )
    return role


def _set_prof(user, skill_name, proficiency, level='BEGINNER'):
    skill = Skill.objects.get(skill_name=skill_name)
    UserSkill.objects.update_or_create(
        user=user, skill=skill,
        defaults={'proficiency': proficiency, 'user_level': level},
    )


def _resource(title, r_type, difficulty='INTERMEDIATE', rating=3.5, url=None,
              duration=60):
    return Resource.objects.create(
        title=title, provider='TestProvider',
        url=url or f'https://example.com/{title.replace(" ", "-").lower()}',
        difficulty_level=difficulty, duration=duration, type=r_type, rating=rating,
    )


def _link(skill_name, resource, relevance=1.0):
    SkillResource.objects.create(
        skill=Skill.objects.get(skill_name=skill_name),
        resource=resource, relevance_score=relevance,
    )


def _recs_for(user, role, skill_name, **kwargs):
    skill_id = Skill.objects.get(skill_name=skill_name).id
    rec = compute_recommendations(user, role, **kwargs)
    return rec['recommendations'].get(skill_id, []), skill_id


class TestTierMatrix:
    """Equal-quality resources at every difficulty → ordering is driven purely
    by level distance from the user's anchor, near strictly before far."""

    @pytest.mark.parametrize('prof,anchor', [
        (30, 'BEGINNER'),
        (50, 'INTERMEDIATE'),
        (80, 'ADVANCED'),
    ])
    def test_equal_quality_orders_by_distance_then_id(self, user, prof, anchor):
        # ADVANCED-required so the skill stays a gap at every proficiency tested.
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])
        _set_prof(user, 'Python', prof)
        resources = {}
        for diff in ('BEGINNER', 'INTERMEDIATE', 'ADVANCED'):
            res = _resource(f'{diff} course', 'COURSE', difficulty=diff, rating=3.5)
            _link('Python', res, relevance=0.8)
            resources[res.id] = diff

        items, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        assert proficiency_to_level(prof) == anchor  # sanity on the fixture

        anchor_idx = DIFFICULTY_ORDER[anchor]

        def dist(diff):
            return abs(anchor_idx - DIFFICULTY_ORDER[diff])

        # distances are non-decreasing down the list (exact-quality case)
        dists = [dist(it['difficulty_level']) for it in items]
        assert dists == sorted(dists)
        # tier (near=0 ≤NEAR_MAX_DISTANCE, far=1) is non-decreasing → no far
        # resource precedes any near one.
        tiers = [0 if d <= NEAR_MAX_DISTANCE else 1 for d in dists]
        assert tiers == sorted(tiers)
        # the exact-level resource always ranks first.
        assert items[0]['difficulty_level'] == anchor


class TestProficiencyBoundaries:
    """The anchor level used for ranking tracks proficiency_to_level exactly at
    every boundary (≤40 BEGINNER / ≤60 INTERMEDIATE / else ADVANCED)."""

    @pytest.mark.parametrize('prof', [0, 40, 41, 60, 61, 80])
    def test_anchor_matches_proficiency_to_level(self, user, prof):
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])  # gap up to 99
        _set_prof(user, 'Python', prof)
        for diff in ('BEGINNER', 'INTERMEDIATE', 'ADVANCED'):
            _link('Python', _resource(f'{diff} c', 'COURSE', difficulty=diff,
                                      rating=3.5), relevance=0.8)
        items, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        # The resource whose difficulty == the anchor level is the exact match
        # and therefore ranks first.
        assert items[0]['difficulty_level'] == proficiency_to_level(prof)

    def test_fully_met_skill_yields_no_recommendation(self, user):
        """prof 100 against the max threshold is MET → excluded entirely (a
        boundary the anchor sweep can't reach, since MET skills have no recs)."""
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])  # threshold 100
        _set_prof(user, 'Python', 100)
        _link('Python', _resource('adv c', 'COURSE', difficulty='ADVANCED'),
              relevance=0.8)
        items, skill_id = _recs_for(user, role, 'Python', limit_per_skill=10)
        rec = compute_recommendations(user, role, limit_per_skill=10)
        assert skill_id not in rec['recommendations']
        assert items == []


class TestLimitTierInteraction:
    def test_top_slots_fill_with_near_first(self, user):
        """4 near + 2 far, limit 3 → every returned item is near."""
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])
        _set_prof(user, 'Python', 30)  # BEGINNER anchor
        for i in range(2):
            _link('Python', _resource(f'beg{i}', 'COURSE', difficulty='BEGINNER',
                                      url=f'https://e.com/b{i}'), relevance=0.8)
            _link('Python', _resource(f'int{i}', 'COURSE', difficulty='INTERMEDIATE',
                                      url=f'https://e.com/i{i}'), relevance=0.8)
        for i in range(2):  # far
            _link('Python', _resource(f'adv{i}', 'COURSE', difficulty='ADVANCED',
                                      url=f'https://e.com/a{i}'), relevance=0.8)

        items, _ = _recs_for(user, role, 'Python', limit_per_skill=3)
        assert len(items) == 3
        assert all(it['difficulty_level'] in ('BEGINNER', 'INTERMEDIATE')
                   for it in items)

    def test_far_kept_when_slots_remain(self, user):
        """1 near + 3 far, limit 3 → near first, then 2 far (far not dropped
        while there are open slots)."""
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])
        _set_prof(user, 'Python', 30)  # BEGINNER anchor
        _link('Python', _resource('beg', 'COURSE', difficulty='BEGINNER'),
              relevance=0.8)
        for i in range(3):  # far (ADVANCED, dist 2)
            _link('Python', _resource(f'adv{i}', 'COURSE', difficulty='ADVANCED',
                                      url=f'https://e.com/a{i}'), relevance=0.8)

        items, _ = _recs_for(user, role, 'Python', limit_per_skill=3)
        assert len(items) == 3
        assert items[0]['difficulty_level'] == 'BEGINNER'
        assert [it['difficulty_level'] for it in items[1:]] == ['ADVANCED', 'ADVANCED']


class TestFarOnly:
    def test_far_only_catalog_still_returned(self, user):
        """ADVANCED learner, only BEGINNER material → all of it is returned, not
        filtered out. The tier is a sort key, never a filter."""
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])
        _set_prof(user, 'Python', 80)  # ADVANCED anchor; BEGINNER res = dist 2
        for i in range(3):
            _link('Python', _resource(f'beg{i}', 'COURSE', difficulty='BEGINNER',
                                      url=f'https://e.com/b{i}'), relevance=0.8)
        items, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        assert len(items) == 3
        assert all(it['difficulty_level'] == 'BEGINNER' for it in items)


class TestVolumeAndStability:
    def test_large_catalog_limits_and_is_deterministic(self, user):
        role = _role_with([('Python', 'INTERMEDIATE', 1.0, True)])
        for i in range(50):
            _link('Python', _resource(f'r{i}', 'COURSE', url=f'https://e.com/{i}'),
                  relevance=0.8)
        a, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        b, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        assert len(a) == 10
        assert [it['resource_id'] for it in a] == [it['resource_id'] for it in b]


class TestScoreFieldNonMonotonic:
    def test_far_resource_can_outscore_yet_rank_below(self, user):
        """The score field is exposed and equals _score(...) rounded, but a far
        resource may carry a HIGHER score than a near one and still rank below
        it — documenting that the tier key intentionally breaks score-rank
        monotonicity (a guarantee, not a bug)."""
        role = _role_with([('Python', 'ADVANCED', 1.0, True)])
        _set_prof(user, 'Python', 30)  # BEGINNER anchor
        # near: weak INTERMEDIATE DOCS
        _link('Python', _resource('int docs', 'DOCS', difficulty='INTERMEDIATE',
                                  rating=4.0), relevance=0.7)
        # far: strong ADVANCED VIDEO
        _link('Python', _resource('adv video', 'VIDEO', difficulty='ADVANCED',
                                  rating=5.0), relevance=1.0)

        items, _ = _recs_for(user, role, 'Python', limit_per_skill=10)
        assert [it['difficulty_level'] for it in items] == ['INTERMEDIATE', 'ADVANCED']
        near, far = items[0], items[1]
        # the near resource ranks first despite a lower score
        assert near['score'] < far['score']
        # score equals the documented formula, rounded to 4dp
        assert near['score'] == round(
            _score('DOCS', 'BEGINNER', 'INTERMEDIATE', 0.7, 4.0), 4)
        assert far['score'] == round(
            _score('VIDEO', 'BEGINNER', 'ADVANCED', 1.0, 5.0), 4)


class TestEndpointAdversarial:
    def test_huge_limit_clamped(self, user, auth_client):
        role = _role_with([('Python', 'INTERMEDIATE', 1.0, True)])
        UserTargetRole.objects.create(user=user, role=role, is_active=True)
        for i in range(MAX_LIMIT_PER_SKILL + 5):
            _link('Python', _resource(f'r{i}', 'COURSE', url=f'https://e.com/{i}'),
                  relevance=0.8)
        r = auth_client.get(f'{RECS_URL}?limit=99999999')
        assert r.status_code == 200, r.data
        sid = Skill.objects.get(skill_name='Python').id
        assert len(r.data['recommendations'][sid]) == MAX_LIMIT_PER_SKILL

    @pytest.mark.parametrize('bad', ['-1', 'abc', '999999'])
    def test_bad_role_param_never_500(self, user, auth_client, bad):
        """A hostile ?role= value resolves to a clean 404, never a 500."""
        r = auth_client.get(f'{RECS_URL}?role={bad}')
        assert r.status_code == 404, (bad, r.status_code)

    def test_xss_skill_and_title_round_trip_unescaped(self, user, auth_client):
        """The API returns raw values (no HTML-escaping) — the React client is
        responsible for escaping on render. Confirms the payload is not double-
        encoded and that no injection is performed server-side."""
        xss = '<script>alert(1)</script>'
        role = _role_with([(xss, 'INTERMEDIATE', 1.0, True)])
        UserTargetRole.objects.create(user=user, role=role, is_active=True)
        _link(xss, _resource(xss, 'COURSE'), relevance=0.9)

        r = auth_client.get(RECS_URL)
        assert r.status_code == 200
        sid = Skill.objects.get(skill_name=xss).id
        assert r.data['recommendations'][sid][0]['title'] == xss
        # rendered JSON carries the raw string, not an HTML-escaped one.
        assert b'<script>alert(1)</script>' in r.content
        assert b'&lt;script&gt;' not in r.content