block user in django admin

tests/test_customer_login_access.py

@@ -0,0 +1,215 @@
+from unittest.mock import patch
+
+import pytest
+from django.contrib.auth import SESSION_KEY
+from django.test import Client
+from django.urls import reverse
+
+from apps.accounts.models import OTPChallenge, User
+from apps.customers.models import CustomerProfile
+from apps.staff.models import StaffProfile
+
+
+pytestmark = pytest.mark.django_db
+PASSWORD = "Customer-access-2026!"
+REASON = "به علت بدهی معوق، ورود شما تا بررسی واحد مالی مسدود شده است."
+
+
+@pytest.fixture
+def customer():
+    user = User.objects.create_user(username="blocked-customer", mobile="09121234566", password=PASSWORD)
+    CustomerProfile.objects.create(user=user)
+    return user
+
+
+def block(user, reason=REASON):
+    profile = CustomerProfile.objects.get(user=user)
+    profile.login_blocked = True
+    profile.login_block_reason = reason
+    profile.save(update_fields=["login_blocked", "login_block_reason"])
+    return profile
+
+
+@pytest.mark.parametrize("stakeholder", CustomerProfile.StakeholderType.values)
+def test_blocked_customers_and_contacts_cannot_request_sms(client, customer, stakeholder):
+    profile = block(customer)
+    profile.stakeholder_type = stakeholder
+    profile.save(update_fields=["stakeholder_type"])
+    with patch("apps.accounts.views.get_sms_provider") as provider:
+        response = client.post(reverse("accounts:login"), {"mobile": customer.local_mobile})
+    assert response.status_code == 200
+    assert REASON in response.content.decode()
+    assert 'role="alert"' in response.content.decode()
+    assert SESSION_KEY not in client.session
+    assert not OTPChallenge.objects.exists()
+    provider.assert_not_called()
+
+
+def test_password_login_is_blocked_and_reason_is_escaped(client, customer):
+    block(customer, "علت مسدودی\n<script>alert('unsafe')</script>")
+    response = client.post(reverse("accounts:password_login"), {"mobile": customer.local_mobile, "password": PASSWORD})
+    content = response.content.decode()
+    assert response.status_code == 200
+    assert "علت مسدودی" in content
+    assert "&lt;script&gt;" in content
+    assert "<script>alert('unsafe')</script>" not in content
+    assert SESSION_KEY not in client.session
+
+
+@pytest.mark.parametrize("method", ["get", "post"])
+def test_blocking_after_otp_issue_prevents_verification(client, customer, method):
+    challenge, code = OTPChallenge.issue(customer.mobile)
+    session = client.session
+    session["otp_challenge_id"] = challenge.pk
+    session.save()
+    block(customer)
+    response = getattr(client, method)(reverse("accounts:otp"), {"code": code}, follow=True)
+    assert response.status_code == 200
+    assert REASON in response.content.decode()
+    assert SESSION_KEY not in client.session
+    assert "otp_challenge_id" not in client.session
+
+
+@pytest.mark.parametrize("endpoint", ["accounts:profile", "tickets:list", "payments:home", "accounts:login", "accounts:login_access"])
+def test_blocking_logs_out_all_existing_sessions(client, customer, endpoint):
+    other_browser = Client()
+    client.force_login(customer)
+    other_browser.force_login(customer)
+    block(customer)
+    for browser in (client, other_browser):
+        response = browser.get(reverse(endpoint))
+        if endpoint == "accounts:login_access":
+            assert response.json()["authenticated"] is False
+            assert "no-store" in response["Cache-Control"]
+        else:
+            assert response.status_code == 302
+            assert response.url == reverse("accounts:login")
+        assert SESSION_KEY not in browser.session
+        assert REASON in browser.get(reverse("accounts:login")).content.decode()
+
+
+def test_blocked_session_cannot_write_customer_profile(client, customer):
+    client.force_login(customer)
+    block(customer)
+    response = client.post(reverse("accounts:profile"), {"first_name": "تغییر غیرمجاز"})
+    assert response.status_code == 302
+    customer.customer_profile.refresh_from_db()
+    assert customer.customer_profile.first_name == ""
+
+
+def test_unblocking_does_not_restore_an_old_session_but_allows_new_login(client, customer):
+    client.force_login(customer)
+    profile = block(customer)
+    profile.login_blocked = False
+    profile.save(update_fields=["login_blocked"])
+    response = client.get(reverse("accounts:profile"))
+    assert response.url == reverse("accounts:login")
+    assert SESSION_KEY not in client.session
+    login_page = client.get(response.url).content.decode()
+    assert "نشست قبلی شما باطل شده است." in login_page
+    assert "ورود به حساب شما مسدود است" not in login_page
+    response = client.post(reverse("accounts:password_login"), {"mobile": customer.local_mobile, "password": PASSWORD})
+    assert response.url == reverse("payments:home")
+    assert SESSION_KEY in client.session
+    assert client.get(reverse("accounts:profile")).status_code == 200
+
+
+def test_unblocked_customer_can_still_login_with_otp(client, customer):
+    challenge, code = OTPChallenge.issue(customer.mobile)
+    session = client.session
+    session["otp_challenge_id"] = challenge.pk
+    session.save()
+    response = client.post(reverse("accounts:otp"), {"code": code})
+    assert response.url == reverse("payments:home")
+    assert SESSION_KEY in client.session
+
+
+@pytest.mark.parametrize("role", ["staff", "django_staff", "superuser"])
+def test_staff_and_admin_sessions_are_exempt_even_with_blocked_customer_profile(client, role):
+    user = User.objects.create_user(
+        username=role, password=PASSWORD, mobile="09129998877",
+        account_type="STAFF" if role == "staff" else "CUSTOMER",
+        is_staff=role == "django_staff", is_superuser=role == "superuser",
+    )
+    CustomerProfile.objects.create(user=user, login_blocked=True, login_block_reason=REASON)
+    if role == "staff":
+        StaffProfile.objects.create(user=user, display_name="کارشناس")
+        response = client.post(reverse("accounts:staff_login"), {"username": user.username, "password": PASSWORD})
+        assert response.url == reverse("staff_tickets:list")
+        workspace = client.get(reverse("staff_tickets:list"))
+        assert workspace.status_code == 200
+        assert "js/customer-login-access.js" not in workspace.content.decode()
+    else:
+        client.force_login(user)
+    assert client.get(reverse("accounts:login_access")).json()["authenticated"] is True
+    assert SESSION_KEY in client.session
+    response = client.get(reverse("accounts:login"))
+    assert response.status_code == 302
+    assert SESSION_KEY in client.session
+
+
+def test_only_admin_can_change_login_block_settings(client, customer):
+    client.force_login(customer)
+    client.post(reverse("accounts:profile"), {
+        "first_name": "مشتری", "last_name": "آزمایشی", "customer_type": "PERSON",
+        "login_blocked": "on", "login_block_reason": REASON,
+    })
+    profile = CustomerProfile.objects.get(user=customer)
+    assert not profile.login_blocked
+    assert profile.login_block_reason == ""
+
+
+def test_admin_requires_a_reason_and_can_block_and_unblock(client, customer):
+    admin = User.objects.create_superuser(username="access-admin", password=PASSWORD)
+    client.force_login(admin)
+    profile = customer.customer_profile
+    url = reverse("admin:customers_customerprofile_change", args=[profile.pk])
+    data = {"user": customer.pk, "stakeholder_type": "CUSTOMER", "login_blocked": "on", "login_block_reason": "   "}
+    response = client.post(url, data)
+    assert response.status_code == 200
+    assert "برای مسدود کردن ورود، دلیل مسدودی را بنویسید." in response.content.decode()
+    profile.refresh_from_db()
+    assert not profile.login_blocked
+    data["login_block_reason"] = REASON
+    assert client.post(url, data).status_code == 302
+    profile.refresh_from_db()
+    assert profile.login_blocked and profile.login_block_reason == REASON
+    assert profile.login_access_version == 1
+    data.pop("login_blocked")
+    assert client.post(url, data).status_code == 302
+    profile.refresh_from_db()
+    assert not profile.login_blocked
+    assert profile.login_access_version == 1
+
+
+def test_legacy_session_without_access_version_is_revoked(client, customer):
+    client.force_login(customer)
+    session = client.session
+    session.pop("customer_login_access_version")
+    session.save()
+    block(customer)
+    assert client.get(reverse("accounts:profile")).url == reverse("accounts:login")
+    assert SESSION_KEY not in client.session
+
+
+def test_missing_profile_does_not_prevent_customer_login(client):
+    user = User.objects.create_user(username="no-profile", mobile="09120000111", password=PASSWORD)
+    response = client.post(reverse("accounts:password_login"), {"mobile": user.local_mobile, "password": PASSWORD})
+    assert response.url == reverse("payments:home")
+    assert client.get(reverse("accounts:login_access")).json()["authenticated"] is True
+
+
+def test_login_block_does_not_change_scoped_magic_link_access(client, customer):
+    from apps.accounts.models import TicketMagicLink
+    from apps.tickets.models import TicketCategory
+    from apps.tickets.services import create_ticket
+
+    category = TicketCategory.objects.create(title="فروش", slug="login-access-sale")
+    ticket = create_ticket(customer=customer, category=category, subject="دسترسی لینک", body="متن")
+    token = TicketMagicLink.issue(ticket)
+    block(customer)
+    response = client.get(reverse("accounts:magic", args=[token]), follow=True)
+    assert response.status_code == 200
+    assert "فقط به همین تیکت دسترسی دارید" in response.content.decode()
+    assert SESSION_KEY not in client.session
+    assert client.get(reverse("tickets:list")).status_code == 302