tests/test_qa_flow.py

"""Flow-level QA checks — converted from scripts/qa_flow_checks.py (F33).



The original script drove APIClient against the *configured* DB with no

isolation (manual `.delete()` cleanup + `cache.clear()` on the real cache),

so a crashed run could leave junk users/skills behind and wipe live throttle

state. These now run against the pytest test DB (`@pytest.mark.django_db` →

per-test rollback), and conftest's autouse `_clear_throttle_cache` gives each

throttle test a clean rate-limit bucket. Behavioural coverage is unchanged.

"""
from io import StringIO

import pytest
from django.contrib.auth import get_user_model
from django.core.management import call_command
from rest_framework.test import APIClient

from apps.progress.models import UserProgress
from apps.resources.models import Resource
from apps.roles.models import Role, UserTargetRole
from apps.skills.models import Skill, UserSkill

User = get_user_model()


@pytest.fixture
def seeded(db):
    """Full curated seed — skills, roles, resources."""
    call_command('seed_initial_skills', stdout=StringIO())
    call_command('seed_initial_roles', stdout=StringIO())
    call_command('seed_initial_resources', stdout=StringIO())
    return True


@pytest.mark.django_db
def test_register_email_casing_collision_and_login():
    """Lowercase register → 201; an upper-case variant of the same address →

    400 (case-insensitive uniqueness); login accepts any casing."""
    c = APIClient()
    r1 = c.post('/api/auth/register/', {
        'name': 'QA Casing', 'email': 'qa.casing@test.xx',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    assert r1.status_code == 201, r1.data

    r2 = c.post('/api/auth/register/', {
        'name': 'QA Casing', 'email': 'QA.Casing@Test.XX',
        'password': 'StrongPass123!', 'password_confirm': 'StrongPass123!',
    }, format='json')
    assert r2.status_code == 400, r2.data

    r3 = c.post('/api/auth/login/', {
        'email': 'QA.CASING@TEST.XX', 'password': 'StrongPass123!',
    }, format='json')
    assert r3.status_code == 200, r3.data


@pytest.mark.django_db
def test_cross_user_data_isolation(seeded):
    """User B never sees User A's UserSkill or UserProgress rows."""
    ua = User.objects.create_user(username='qa.a@test.xx', email='qa.a@test.xx',
                                  password='StrongPass123!', name='Alice')
    ub = User.objects.create_user(username='qa.b@test.xx', email='qa.b@test.xx',
                                  password='StrongPass123!', name='Bob')
    role = Role.objects.filter(is_active=True).first()
    UserTargetRole.objects.update_or_create(user=ua, role=role, defaults={'is_active': True})
    UserTargetRole.objects.update_or_create(user=ub, role=role, defaults={'is_active': True})

    skill = Skill.objects.first()
    UserSkill.objects.update_or_create(
        user=ua, skill=skill,
        defaults={'proficiency': 88, 'user_level': 'ADVANCED'},
    )

    ca = APIClient(); ca.force_authenticate(user=ua)
    cb = APIClient(); cb.force_authenticate(user=ub)

    a_skills = ca.get('/api/user-skills/').json()
    b_skills = cb.get('/api/user-skills/').json()
    assert isinstance(a_skills, list) and len(a_skills) >= 1
    assert not any(s['skill']['id'] == skill.id for s in b_skills)

    res = Resource.objects.first()
    UserProgress.objects.update_or_create(
        user=ua, resource=res,
        defaults={'status': 'IN_PROGRESS', 'progress': 55},
    )
    a_prog = ca.get('/api/progress/').json()
    b_prog = cb.get('/api/progress/').json()
    assert any(p.get('id') for p in a_prog)
    assert not any(p.get('resource') == res.id for p in b_prog)


@pytest.mark.django_db
def test_target_role_switch_reflected_in_analysis(seeded):
    """Switching the active target role changes the role the gap report is

    computed against."""
    ua = User.objects.create_user(username='qa.sw@test.xx', email='qa.sw@test.xx',
                                  password='StrongPass123!', name='Switch')
    r_old = Role.objects.filter(is_active=True).order_by('id').first()
    r_new = (Role.objects.filter(is_active=True)
             .exclude(id=r_old.id).order_by('id').first())
    UserTargetRole.objects.update_or_create(
        user=ua, role=r_old, defaults={'is_active': True})

    ca = APIClient(); ca.force_authenticate(user=ua)
    gap_old = ca.get('/api/analysis/').json()
    ca.post('/api/target-role/', {'role_id': r_new.id}, format='json')
    gap_new = ca.get('/api/analysis/').json()
    assert gap_old.get('role_id') == r_old.id
    assert gap_new.get('role_id') == r_new.id


@pytest.mark.django_db
def test_login_throttle_kicks_in_by_11th_attempt():
    """login scope = 10/minute → the 11th attempt in the window returns 429."""
    c = APIClient(REMOTE_ADDR='10.9.9.9')
    first_429 = None
    for i in range(1, 13):
        r = c.post('/api/auth/login/', {
            'email': 'doesnotexist@test.xx', 'password': 'wrong',
        }, format='json')
        if r.status_code == 429:
            first_429 = i
            break
    assert first_429 is not None, 'login never throttled'
    assert first_429 <= 11, f'throttled too late (attempt {first_429})'


@pytest.mark.django_db
def test_register_throttle_kicks_in_by_21st_attempt():
    """register scope = 20/hour → the 21st attempt in the window returns 429."""
    c = APIClient(REMOTE_ADDR='10.9.9.10')
    first_429 = None
    for i in range(1, 23):
        r = c.post('/api/auth/register/', {
            'name': 'x', 'email': f'throttle{i}@t.xx',
            'password': 'ShortButOk!9', 'password_confirm': 'ShortButOk!9',
        }, format='json')
        if r.status_code == 429:
            first_429 = i
            break
    assert first_429 is not None, 'register never throttled'
    assert first_429 <= 21, f'throttled too late (attempt {first_429})'


@pytest.mark.django_db
def test_skill_with_script_name_round_trips_as_raw_text():
    """A Skill whose name contains <script> is returned verbatim by

    /api/skills/ — the API must not mangle it; React escapes on render."""
    user = User.objects.create_user(username='qa.xss@test.xx', email='qa.xss@test.xx',
                                    password='StrongPass123!', name='XSS')
    Skill.objects.create(skill_name='<script>alert(1)</script>', category='X',
                         difficulty_level='BEGINNER')
    c = APIClient(); c.force_authenticate(user=user)
    body = c.get('/api/skills/').json()
    rows = body['results'] if isinstance(body, dict) and 'results' in body else body
    assert any('<script>' in (row.get('skill_name') or '') for row in rows)


@pytest.mark.django_db
def test_resources_skill_filter_rejects_injection_string():
    """`?skill=1 OR 1=1` is rejected (400) by the NumberFilter before it can

    reach the ORM — no SQL injection surface."""
    user = User.objects.create_user(username='qa.inj@test.xx', email='qa.inj@test.xx',
                                    password='StrongPass123!', name='Inj')
    c = APIClient(); c.force_authenticate(user=user)
    r = c.get('/api/resources/?skill=1%20OR%201=1')
    assert r.status_code == 400, r.status_code