apps/roles/tests/test_roles.py

import pytest
from rest_framework import status
from rest_framework.test import APIClient
from django.contrib.auth import get_user_model

from apps.skills.models import Skill
from apps.roles.models import Role, RoleSkill, UserTargetRole

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

ROLES_URL = '/api/roles/'
TARGET_ROLE_URL = '/api/target-role/'


@pytest.fixture
def auth_client():
    user = User.objects.create_user(
        username="testuser@example.com",
        email="testuser@example.com",
        password="password123",
        name="Test User"
    )
    client = APIClient()
    # Force authenticate user bypasses the JWT requirement for fast testing
    client.force_authenticate(user=user)
    return client


@pytest.fixture
def user(auth_client):
    return User.objects.first()


@pytest.fixture
def skill():
    return Skill.objects.create(skill_name="Python", category="Programming", difficulty_level="BEGINNER")


@pytest.fixture
def role(skill):
    r = Role.objects.create(role_name="Backend Developer", industry="Tech", is_active=True)
    RoleSkill.objects.create(role=r, skill=skill, required_level="INTERMEDIATE", weight=1.0, is_mandatory=True)
    return r


@pytest.fixture
def inactive_role(skill):
    r = Role.objects.create(role_name="Deprecated Dev", industry="Tech", is_active=False)
    RoleSkill.objects.create(role=r, skill=skill, required_level="BEGINNER")
    return r


class TestRoleCatalog:
    def test_list_roles_returns_active_only(self, auth_client, role, inactive_role):
        response = auth_client.get(ROLES_URL)
        assert response.status_code == status.HTTP_200_OK
        assert len(response.data) == 1
        assert response.data[0]['id'] == role.id

    def test_list_roles_filter_industry(self, auth_client, role):
        # Role is created with industry "Tech"
        response = auth_client.get(ROLES_URL + '?industry=Tech')
        assert response.status_code == status.HTTP_200_OK
        assert len(response.data) == 1

        response2 = auth_client.get(ROLES_URL + '?industry=Healthcare')
        assert response2.status_code == status.HTTP_200_OK
        assert len(response2.data) == 0

    def test_role_detail_includes_skills(self, auth_client, role, skill):
        response = auth_client.get(f"{ROLES_URL}{role.id}/")
        assert response.status_code == status.HTTP_200_OK
        assert response.data['id'] == role.id
        assert len(response.data['role_skills']) == 1
        assert response.data['role_skills'][0]['skill']['id'] == skill.id
        assert response.data['role_skills'][0]['is_mandatory'] is True


class TestUserTargetRole:
    def test_set_target_role_success(self, auth_client, user, role):
        response = auth_client.post(TARGET_ROLE_URL, {'role_id': role.id}, format='json')
        assert response.status_code == status.HTTP_201_CREATED
        assert UserTargetRole.objects.filter(user=user, is_active=True).exists()

    def test_set_inactive_role_fails(self, auth_client, inactive_role):
        response = auth_client.post(TARGET_ROLE_URL, {'role_id': inactive_role.id}, format='json')
        # Serializer rejects inactive or missing roles as bad input (400).
        assert response.status_code == status.HTTP_400_BAD_REQUEST
        assert 'role_id' in response.data

    def test_switching_deactivates_previous(self, auth_client, user, role, skill):
        role_b = Role.objects.create(role_name="Frontend Dev", industry="Tech", is_active=True)
        RoleSkill.objects.create(role=role_b, skill=skill, required_level="BEGINNER")

        # Select role A
        auth_client.post(TARGET_ROLE_URL, {'role_id': role.id}, format='json')
        tr1 = UserTargetRole.objects.get(user=user)
        assert tr1.is_active is True

        # Select role B
        auth_client.post(TARGET_ROLE_URL, {'role_id': role_b.id}, format='json')

        # Check via API
        response = auth_client.get(TARGET_ROLE_URL)
        assert response.status_code == status.HTTP_200_OK
        assert response.data['role']['id'] == role_b.id

        # Check DB
        tr1.refresh_from_db()
        assert tr1.is_active is False
        assert UserTargetRole.objects.filter(user=user, is_active=True).count() == 1
        assert UserTargetRole.objects.get(user=user, is_active=True).role == role_b

    def test_delete_target_role(self, auth_client, user, role):
        auth_client.post(TARGET_ROLE_URL, {'role_id': role.id}, format='json')
        response = auth_client.delete(TARGET_ROLE_URL)
        assert response.status_code == status.HTTP_204_NO_CONTENT

        # Should be inactive in DB
        assert UserTargetRole.objects.filter(user=user, is_active=True).count() == 0

        # GET should return 404
        get_response = auth_client.get(TARGET_ROLE_URL)
        assert get_response.status_code == status.HTTP_404_NOT_FOUND

    def test_one_active_target_per_user_constraint(self, user, role):
        """DB-level guard: the partial unique constraint must reject a
        second active row. App-level upsert in the view avoids this path,
        but direct ORM writes (admin, scripts) must be safe too."""
        from django.db import IntegrityError, transaction
        UserTargetRole.objects.create(user=user, role=role, is_active=True)
        with pytest.raises(IntegrityError):
            with transaction.atomic():
                UserTargetRole.objects.create(user=user, role=role, is_active=True)

    def test_concurrent_deactivation_returns_400(self, auth_client, role):
        """F14: if the role is deactivated AFTER validate_role_id passes but
        before the view's Role.objects.get(), the DoesNotExist must surface as a
        clean 400 keyed on role_id, not an unhandled 500."""
        from unittest.mock import patch
        with patch.object(Role.objects, 'get', side_effect=Role.DoesNotExist):
            response = auth_client.post(
                TARGET_ROLE_URL, {'role_id': role.id}, format='json',
            )
        assert response.status_code == status.HTTP_400_BAD_REQUEST
        assert 'role_id' in response.data


class TestRoleAdminActions:
    _counter = 0

    def _admin_request(self):
        from django.test import RequestFactory
        from django.contrib.messages.storage.fallback import FallbackStorage
        TestRoleAdminActions._counter += 1
        tag = f'admin-role-{TestRoleAdminActions._counter}@x.com'
        req = RequestFactory().post('/admin/')
        setattr(req, 'session', {})
        setattr(req, '_messages', FallbackStorage(req))
        req.user = User.objects.create_superuser(
            username=tag, email=tag, password='pw',
        )
        return req

    def test_duplicate_role_with_skills(self, role, skill):
        from django.contrib.admin.sites import AdminSite
        from apps.roles.admin import RoleAdmin
        admin = RoleAdmin(Role, AdminSite())
        request = self._admin_request()
        admin.duplicate_role_with_skills(request, Role.objects.filter(pk=role.pk))

        copy = Role.objects.get(role_name='Backend Developer (copy)')
        assert copy.pk != role.pk
        assert copy.industry == role.industry
        assert copy.onet_soc_code == role.onet_soc_code
        src_pairs = {(rs.skill_id, rs.required_level, rs.weight, rs.is_mandatory)
                     for rs in role.role_skills.all()}
        copy_pairs = {(rs.skill_id, rs.required_level, rs.weight, rs.is_mandatory)
                      for rs in copy.role_skills.all()}
        assert src_pairs == copy_pairs

    def test_duplicate_role_handles_name_collisions(self, role):
        from django.contrib.admin.sites import AdminSite
        from apps.roles.admin import RoleAdmin
        admin = RoleAdmin(Role, AdminSite())
        request = self._admin_request()
        admin.duplicate_role_with_skills(request, Role.objects.filter(pk=role.pk))
        admin.duplicate_role_with_skills(request, Role.objects.filter(pk=role.pk))

        names = set(Role.objects.values_list('role_name', flat=True))
        assert 'Backend Developer (copy)' in names
        assert 'Backend Developer (copy) 2' in names

    def test_import_skills_from_onet_missing_yaml(self, tmp_path, monkeypatch, role):
        """When the curated YAML is absent, the action emits an error
        message and makes no changes."""
        from django.contrib.admin.sites import AdminSite
        from apps.roles import admin as roles_admin
        admin = roles_admin.RoleAdmin(Role, AdminSite())
        monkeypatch.setattr(roles_admin, 'CURATED_YAML', tmp_path / 'missing.yaml')
        before = role.role_skills.count()
        request = self._admin_request()
        admin.import_skills_from_onet(request, Role.objects.filter(pk=role.pk))
        assert role.role_skills.count() == before

    def test_import_skills_from_onet_populates_from_yaml(
        self, tmp_path, monkeypatch, skill,
    ):
        from django.contrib.admin.sites import AdminSite
        from apps.roles import admin as roles_admin

        # Set up a Role and a second skill referenced by the curated YAML.
        Skill.objects.create(skill_name='Django', category='Framework', difficulty_level='INTERMEDIATE')
        target_role = Role.objects.create(
            role_name='Backend Engineer',
            industry='Tech', is_active=True,
            onet_soc_code='15-1252.00',
        )

        yaml_path = tmp_path / 'curated.yaml'
        yaml_path.write_text(
            "roles:\n"
            "- name: Backend Engineer\n"
            "  industry: Tech\n"
            "  primary_soc: 15-1252.00\n"
            "  all_socs: [15-1252.00]\n"
            "  skills:\n"
            "  - skill_name: Python\n"
            "    required_level: INTERMEDIATE\n"
            "    weight: 1.0\n"
            "    is_mandatory: true\n"
            "  - skill_name: Django\n"
            "    required_level: ADVANCED\n"
            "    weight: 2.0\n"
            "    is_mandatory: false\n",
            encoding='utf-8',
        )
        monkeypatch.setattr(roles_admin, 'CURATED_YAML', yaml_path)

        admin = roles_admin.RoleAdmin(Role, AdminSite())
        request = self._admin_request()
        admin.import_skills_from_onet(request, Role.objects.filter(pk=target_role.pk))

        rs_rows = {rs.skill.skill_name: rs for rs in target_role.role_skills.all()}
        assert set(rs_rows) == {'Python', 'Django'}
        assert rs_rows['Python'].is_mandatory is True
        assert rs_rows['Django'].required_level == 'ADVANCED'
        assert rs_rows['Django'].weight == 2.0

    def test_duplicate_role_writes_audit_log(self, role, skill):
        from django.contrib.admin.models import ADDITION, LogEntry
        from django.contrib.admin.sites import AdminSite
        from apps.roles.admin import RoleAdmin

        admin = RoleAdmin(Role, AdminSite())
        request = self._admin_request()
        admin.duplicate_role_with_skills(request, Role.objects.filter(pk=role.pk))

        copy = Role.objects.get(role_name='Backend Developer (copy)')
        entries = LogEntry.objects.filter(
            object_id=str(copy.pk), action_flag=ADDITION,
        )
        assert entries.exists()
        assert 'Duplicated from' in entries.first().change_message

    def test_import_skills_writes_audit_log(self, tmp_path, monkeypatch, skill):
        from django.contrib.admin.models import CHANGE, LogEntry
        from django.contrib.admin.sites import AdminSite
        from apps.roles import admin as roles_admin

        target_role = Role.objects.create(
            role_name='Audited Role', industry='Tech',
            is_active=True, onet_soc_code='15-1252.00',
        )
        yaml_path = tmp_path / 'curated.yaml'
        yaml_path.write_text(
            "roles:\n"
            "- name: Audited Role\n"
            "  primary_soc: 15-1252.00\n"
            "  all_socs: [15-1252.00]\n"
            "  skills:\n"
            "  - skill_name: Python\n"
            "    required_level: INTERMEDIATE\n"
            "    weight: 1.0\n"
            "    is_mandatory: true\n",
            encoding='utf-8',
        )
        monkeypatch.setattr(roles_admin, 'CURATED_YAML', yaml_path)

        admin_cls = roles_admin.RoleAdmin(Role, AdminSite())
        admin_cls.import_skills_from_onet(
            self._admin_request(), Role.objects.filter(pk=target_role.pk),
        )

        entries = LogEntry.objects.filter(
            object_id=str(target_role.pk), action_flag=CHANGE,
        )
        assert entries.exists()
        assert 'O*NET import' in entries.first().change_message

    def test_import_skills_is_idempotent(self, tmp_path, monkeypatch, skill):
        from django.contrib.admin.sites import AdminSite
        from apps.roles import admin as roles_admin

        target_role = Role.objects.create(
            role_name='Solo Role', industry='Tech',
            is_active=True, onet_soc_code='99-9999.99',
        )
        yaml_path = tmp_path / 'curated.yaml'
        yaml_path.write_text(
            "roles:\n"
            "- name: Solo Role\n"
            "  primary_soc: 99-9999.99\n"
            "  all_socs: [99-9999.99]\n"
            "  skills:\n"
            "  - skill_name: Python\n"
            "    required_level: INTERMEDIATE\n"
            "    weight: 1.0\n"
            "    is_mandatory: true\n",
            encoding='utf-8',
        )
        monkeypatch.setattr(roles_admin, 'CURATED_YAML', yaml_path)

        admin = roles_admin.RoleAdmin(Role, AdminSite())
        admin.import_skills_from_onet(self._admin_request(), Role.objects.filter(pk=target_role.pk))
        admin.import_skills_from_onet(self._admin_request(), Role.objects.filter(pk=target_role.pk))
        assert target_role.role_skills.count() == 1