tests/verify_claims.py

"""Verification tests for the code-review fixes.

Each test encodes the *expected post-fix behavior* for one of the original
review claims. A passing test means the fix is in place. Intentionally kept
outside the default pytest collection glob (see pytest.ini) — invoke with
`pytest tests/verify_claims.py` when you want an audit trail.
"""
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 apply_upgrade
from apps.resources.models import Resource, SkillResource
from apps.skills.models import Skill, UserSkill

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


REGISTER = '/api/auth/register/'


# ---------------------------------------------------------------------------
# FIX #1 — apply_upgrade is now idempotent per completion. One completion
# yields exactly one bump, then already_applied=True.
# ---------------------------------------------------------------------------
def test_fix_1_apply_upgrade_is_idempotent_per_completion():
    user = User.objects.create_user(
        username='claim1@x.com', email='claim1@x.com', password='pw', name='c1',
    )
    py = Skill.objects.create(
        skill_name='Python', category='Programming', difficulty_level='BEGINNER',
    )
    r = Resource.objects.create(
        title='R', provider='P', url='https://e.com/c1',
        difficulty_level='BEGINNER', duration=10, type='DOCS', rating=4.0,
    )
    SkillResource.objects.create(skill=py, resource=r, relevance_score=1.0)
    UserProgress.objects.create(
        user=user, resource=r, status='COMPLETED', progress=100,
        completed_at=timezone.now(),
    )

    results = [apply_upgrade(user, py.id) for _ in range(5)]
    final = UserSkill.objects.get(user=user, skill=py).proficiency

    # First call applies; all subsequent return already_applied=True.
    assert final == 60, f"expected one-rung bump to 60, got {final}"
    assert results[0]['applied'] is True
    for later in results[1:]:
        assert later['applied'] is False
        assert later['already_applied'] is True


# ---------------------------------------------------------------------------
# FIX #2 — password validators are now enforced at registration.
# ---------------------------------------------------------------------------
def test_fix_2_short_password_rejected():
    c = APIClient()
    r = c.post(REGISTER, data={
        'name': 'Weak', 'email': 'weak@x.com',
        'password': 'abc', 'password_confirm': 'abc',
    }, format='json')
    assert r.status_code == 400
    assert 'password' in r.data


def test_fix_2_common_password_rejected():
    c = APIClient()
    r = c.post(REGISTER, data={
        'name': 'Common', 'email': 'common@x.com',
        'password': 'password', 'password_confirm': 'password',
    }, format='json')
    assert r.status_code == 400


def test_fix_2_numeric_password_rejected():
    c = APIClient()
    r = c.post(REGISTER, data={
        'name': 'Numeric', 'email': 'numeric@x.com',
        'password': '12345678', 'password_confirm': '12345678',
    }, format='json')
    assert r.status_code == 400


def test_fix_2_strong_password_accepted():
    c = APIClient()
    r = c.post(REGISTER, data={
        'name': 'Strong', 'email': 'strong@x.com',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    assert r.status_code == 201


# ---------------------------------------------------------------------------
# FIX #3 — email uniqueness is now case-insensitive (normalized lowercase).
# Login accepts either case.
# ---------------------------------------------------------------------------
def test_fix_3_case_variant_duplicate_rejected():
    c = APIClient()
    r1 = c.post(REGISTER, data={
        'name': 'A', 'email': 'DupCase@x.com',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    assert r1.status_code == 201
    # Stored as lowercase.
    assert User.objects.get(email__iexact='dupcase@x.com').email == 'dupcase@x.com'

    r2 = c.post(REGISTER, data={
        'name': 'B', 'email': 'dupcase@x.com',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    assert r2.status_code == 400
    assert User.objects.filter(email__iexact='dupcase@x.com').count() == 1


def test_fix_3_login_case_insensitive():
    c = APIClient()
    c.post(REGISTER, data={
        'name': 'Mixed', 'email': 'Mixed@X.com',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    r = c.post('/api/auth/login/', data={
        'email': 'MIXED@x.COM', 'password': 'StrongPass123!',
    }, format='json')
    assert r.status_code == 200, r.data


# ---------------------------------------------------------------------------
# FIX #8 — resource list no longer N+1s on has_checkpoints. A single SQL
# statement (the Exists subquery) covers the whole page.
# ---------------------------------------------------------------------------
def test_fix_8_resource_list_no_n_plus_one_on_has_checkpoints():
    from django.db import connection
    from django.test.utils import CaptureQueriesContext

    def _count_cp_queries(row_count: int) -> int:
        skill = Skill.objects.create(
            skill_name=f'S{row_count}', category='C', difficulty_level='BEGINNER',
        )
        for i in range(row_count):
            r = Resource.objects.create(
                title=f'R{row_count}_{i}', provider='P',
                url=f'https://e.com/{row_count}_{i}',
                difficulty_level='BEGINNER', duration=10, type='DOCS', rating=4.0,
            )
            SkillResource.objects.create(skill=skill, resource=r, relevance_score=1.0)

        user = User.objects.create_user(
            username=f'cp{row_count}@x.com', email=f'cp{row_count}@x.com',
            password='pw', name='n',
        )
        c = APIClient()
        c.force_authenticate(user=user)

        with CaptureQueriesContext(connection) as ctx:
            resp = c.get('/api/resources/')
        assert resp.status_code == 200

        return sum(
            1 for q in ctx.captured_queries
            if 'resources_resourcecheckpoint' in q['sql'].lower()
        )

    # N+1 signature = query count scales with row count. Exists() annotation
    # folds the lookup into the main SELECT, so the count is constant.
    queries_for_5 = _count_cp_queries(5)
    queries_for_10 = _count_cp_queries(10)
    assert queries_for_5 == queries_for_10, (
        f"query count should not scale with row count; "
        f"got {queries_for_5} for 5 rows and {queries_for_10} for 10 rows"
    )
    # And the total checkpoint-related query count should be tiny (≤ 1).
    assert queries_for_5 <= 1, queries_for_5


# ---------------------------------------------------------------------------
# FIX #5 — ranking docstring arithmetic now matches _score output.
# ---------------------------------------------------------------------------
def test_fix_5_ranking_scores_match_docstring():
    from apps.analysis.services import _score
    docs_score = _score('DOCS', 'INTERMEDIATE', 'INTERMEDIATE', 1.0, 3.5)
    video_score = _score('VIDEO', 'INTERMEDIATE', 'INTERMEDIATE', 0.3, 3.5)
    assert abs(docs_score - 32.5) < 0.01
    assert abs(video_score - 28.0) < 0.01
    assert docs_score > video_score


# ---------------------------------------------------------------------------
# FIX #25 — re-POST of the same target role now bumps selected_at.
# ---------------------------------------------------------------------------
def test_fix_25_same_role_repost_bumps_selected_at():
    from apps.roles.models import Role, UserTargetRole
    from time import sleep

    user = User.objects.create_user(
        username='tr@x.com', email='tr@x.com', password='pw', name='tr',
    )
    role = Role.objects.create(role_name='R', industry='Tech', is_active=True)

    c = APIClient()
    c.force_authenticate(user=user)
    c.post('/api/target-role/', {'role_id': role.id}, format='json')

    first = UserTargetRole.objects.get(user=user, is_active=True)
    first_ts = first.selected_at

    sleep(0.05)
    c.post('/api/target-role/', {'role_id': role.id}, format='json')

    again = UserTargetRole.objects.get(user=user, is_active=True)
    assert again.pk == first.pk  # same row reused (no duplicate created)
    assert again.selected_at > first_ts  # timestamp refreshed


# ---------------------------------------------------------------------------
# FIX #6 / #7 — list endpoint shapes are explicit design choices.
# UserProgress and Roles are both intentionally unpaginated (bounded
# cardinality). Document the contract so the frontend doesn't need to handle
# two shapes.
# ---------------------------------------------------------------------------
def test_fix_6_progress_list_remains_plain_list():
    user = User.objects.create_user(
        username='pag@x.com', email='pag@x.com', password='pw', name='pag',
    )
    c = APIClient()
    c.force_authenticate(user=user)
    r = c.get('/api/progress/')
    assert r.status_code == 200
    assert isinstance(r.data, list)


def test_fix_7_roles_list_remains_plain_list():
    from apps.roles.models import Role

    user = User.objects.create_user(
        username='rag@x.com', email='rag@x.com', password='pw', name='rag',
    )
    Role.objects.create(role_name='X', industry='Tech', is_active=True)

    c = APIClient()
    c.force_authenticate(user=user)
    r = c.get('/api/roles/')
    assert r.status_code == 200
    assert isinstance(r.data, list)


# ---------------------------------------------------------------------------
# FIX (bonus) — only /api/recommendations/ is routed; the duplicate under
# /api/analysis/recommendations/ has been removed.
# ---------------------------------------------------------------------------
def test_fix_bonus_recommendations_not_duplicated():
    user = User.objects.create_user(
        username='dup@x.com', email='dup@x.com', password='pw', name='d',
    )
    c = APIClient()
    c.force_authenticate(user=user)

    # Canonical path still works (400 because no target role is set).
    r = c.get('/api/recommendations/')
    assert r.status_code in (200, 400)

    # Duplicate path should be gone (404).
    r_dup = c.get('/api/analysis/recommendations/')
    assert r_dup.status_code == 404