mirror of
https://github.com/dartdavros/chatballs.git
synced 2026-10-15 21:49:02 +03:00
Andrey
b9f0f13c81
Деплой падал на seed_catalog → Price.objects.update_or_create(... version=1) реактивировал архивную версию цены, нарушая констрейнт uniq_active_price_offer. Причина: цену оффера меняли через UI (создана новая версия), а seed при каждом деплое пытался вернуть version=1 с is_active=True. На проде seed выполнялся 13 раз — команда跑了 на развёрнутой установке. seed_hub_initial_data теперь no-op, если уже есть OWNER (маркер первичного bootstrap). На новой БД сеет как раньше; на развёрнутой — skip, не трогая данные, изменённые через UI/API (каталог, цены, AI-контент). Тест: test_seed_skipped_when_owner_exists покрывает guard.
640 lines
28 KiB
Python
640 lines
28 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
from django.contrib.auth.tokens import default_token_generator
|
|
from django.core import mail
|
|
from django.test import Client, TestCase, override_settings
|
|
from django.contrib.sessions.backends.db import SessionStore
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.http import urlsafe_base64_encode
|
|
from rest_framework.test import APIClient
|
|
from rest_framework.throttling import ScopedRateThrottle
|
|
|
|
_LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}
|
|
|
|
from hub_platform.identity.bootstrap import bootstrap_edevs_owner
|
|
from hub_platform.identity.auth.totp_utils import _totp_code
|
|
from hub_platform.identity.models import (
|
|
AuditEvent,
|
|
Department,
|
|
EmployeeProfile,
|
|
EmployeeRole,
|
|
HumanUser,
|
|
Organization,
|
|
)
|
|
from hub_platform.identity.permissions import can_access_global_settings, can_access_sales_workspace
|
|
from hub_platform.products.models import Product
|
|
|
|
|
|
class BootstrapOwnerTests(TestCase):
|
|
def test_bootstrap_creates_edevs_foundation(self) -> None:
|
|
result = bootstrap_edevs_owner(
|
|
email="owner@edevs.tech",
|
|
password="temporary-password",
|
|
full_name="Owner",
|
|
)
|
|
|
|
self.assertTrue(result.created_owner)
|
|
self.assertEqual(Organization.objects.get().slug, "edevs")
|
|
# Seed создаёт отделы продаж и поддержки (SPEC-HUB-0010 §4.1).
|
|
self.assertEqual(
|
|
set(Department.objects.values_list("code", flat=True)), {"sales", "support"}
|
|
)
|
|
self.assertEqual(result.support_department.code, "support")
|
|
self.assertEqual(set(Product.objects.values_list("code", flat=True)), {"firepage", "foxray"})
|
|
self.assertEqual(result.owner.employee_profile.role, EmployeeRole.OWNER)
|
|
# TOTP выключен по умолчанию (намеренно, локальная разработка).
|
|
self.assertFalse(result.owner.employee_profile.totp_required)
|
|
self.assertTrue(result.owner.is_staff)
|
|
self.assertTrue(result.owner.is_superuser)
|
|
operator = HumanUser.objects.get(email="a.kotova@edevs.tech")
|
|
self.assertEqual(operator.full_name, "Анна Котова")
|
|
self.assertEqual(operator.employee_profile.role, EmployeeRole.OPERATOR)
|
|
self.assertEqual(operator.employee_profile.phone, "+7 916 245 14 02")
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.owner_bootstrapped").exists())
|
|
|
|
def test_bootstrap_is_idempotent_for_owner(self) -> None:
|
|
first = bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
second = bootstrap_edevs_owner(email="owner@edevs.tech", password="another-password")
|
|
|
|
self.assertTrue(first.created_owner)
|
|
self.assertFalse(second.created_owner)
|
|
self.assertEqual(HumanUser.objects.count(), 2)
|
|
self.assertEqual(Organization.objects.count(), 1)
|
|
self.assertEqual(Product.objects.count(), 2)
|
|
|
|
def test_bootstrap_promotes_existing_owner_to_django_admin_access(self) -> None:
|
|
user = HumanUser.objects.create_user(email="owner@edevs.tech", password="temporary-password")
|
|
|
|
result = bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
|
|
user.refresh_from_db()
|
|
self.assertFalse(result.created_owner)
|
|
self.assertTrue(user.is_staff)
|
|
self.assertTrue(user.is_superuser)
|
|
|
|
|
|
class PermissionTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
self.organization = Organization.objects.get(slug="edevs")
|
|
self.sales = Department.objects.get(code="sales")
|
|
|
|
def test_owner_can_access_global_settings_and_sales_workspace(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
|
|
self.assertTrue(can_access_global_settings(owner))
|
|
self.assertTrue(can_access_sales_workspace(owner))
|
|
|
|
def test_operator_cannot_access_global_settings(self) -> None:
|
|
operator = HumanUser.objects.create_user(email="operator@edevs.tech", password="temporary-password")
|
|
EmployeeProfile.objects.create(
|
|
user=operator,
|
|
organization=self.organization,
|
|
role=EmployeeRole.OPERATOR,
|
|
department=self.sales,
|
|
must_change_password=True,
|
|
)
|
|
|
|
self.assertFalse(can_access_global_settings(operator))
|
|
self.assertTrue(can_access_sales_workspace(operator))
|
|
|
|
|
|
class AuthEndpointTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
self.client = Client()
|
|
|
|
def test_login_returns_session_payload(self) -> None:
|
|
response = self.client.post(
|
|
"/api/v1/auth/login/",
|
|
data=json.dumps({"email": "owner@edevs.tech", "password": "temporary-password"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertTrue(payload["authenticated"])
|
|
self.assertEqual(payload["user"]["organizationName"], "Edevs")
|
|
self.assertEqual(payload["user"]["role"], EmployeeRole.OWNER)
|
|
|
|
def test_session_sets_csrf_cookie_for_spa(self) -> None:
|
|
response = self.client.get("/api/v1/auth/session/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("csrftoken", response.cookies)
|
|
|
|
def test_login_rejects_invalid_password(self) -> None:
|
|
response = self.client.post(
|
|
"/api/v1/auth/login/",
|
|
data=json.dumps({"email": "owner@edevs.tech", "password": "wrong-password"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.login_failed").exists())
|
|
|
|
def test_password_reset_request_enqueues_outbox_event(self) -> None:
|
|
from hub_platform.events.models import OutboxEvent
|
|
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
response = self.client.post(
|
|
"/api/v1/auth/password-reset/request/",
|
|
data=json.dumps({"email": "owner@edevs.tech"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()["ok"])
|
|
# Письмо отправляется воркером асинхронно, не в запросе.
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
event = OutboxEvent.objects.get(event_type="identity.password_reset_requested")
|
|
self.assertEqual(event.payload, {"userId": owner.id})
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.password_reset_requested").exists())
|
|
|
|
def test_password_reset_request_does_not_reveal_unknown_email(self) -> None:
|
|
from hub_platform.events.models import OutboxEvent
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/password-reset/request/",
|
|
data=json.dumps({"email": "nobody@edevs.tech"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()["ok"])
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
self.assertFalse(OutboxEvent.objects.filter(event_type="identity.password_reset_requested").exists())
|
|
|
|
def test_password_reset_handler_sends_email(self) -> None:
|
|
from hub_platform.events.handlers import dispatch
|
|
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
dispatch("identity.password_reset_requested", {"userId": owner.id})
|
|
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
self.assertEqual(mail.outbox[0].to, ["owner@edevs.tech"])
|
|
self.assertIn("reset-password", mail.outbox[0].body)
|
|
|
|
def _reset_link(self, email: str) -> tuple[str, str]:
|
|
user = HumanUser.objects.get(email=email)
|
|
return urlsafe_base64_encode(force_bytes(user.pk)), default_token_generator.make_token(user)
|
|
|
|
def test_password_reset_confirm_sets_new_password(self) -> None:
|
|
uid, token = self._reset_link("owner@edevs.tech")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/password-reset/confirm/",
|
|
data=json.dumps({"uid": uid, "token": token, "newPassword": "Fresh-Pass-99"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()["ok"])
|
|
self.assertTrue(HumanUser.objects.get(email="owner@edevs.tech").check_password("Fresh-Pass-99"))
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.password_reset_completed").exists())
|
|
|
|
def test_password_reset_confirm_rejects_invalid_token(self) -> None:
|
|
uid, _ = self._reset_link("owner@edevs.tech")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/password-reset/confirm/",
|
|
data=json.dumps({"uid": uid, "token": "bad-token", "newPassword": "Fresh-Pass-99"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertFalse(HumanUser.objects.get(email="owner@edevs.tech").check_password("Fresh-Pass-99"))
|
|
|
|
def test_password_reset_validate_reflects_token_state(self) -> None:
|
|
uid, token = self._reset_link("owner@edevs.tech")
|
|
|
|
valid = self.client.get(f"/api/v1/auth/password-reset/validate/?uid={uid}&token={token}")
|
|
invalid = self.client.get(f"/api/v1/auth/password-reset/validate/?uid={uid}&token=bad-token")
|
|
|
|
self.assertTrue(valid.json()["valid"])
|
|
self.assertFalse(invalid.json()["valid"])
|
|
|
|
def test_change_temporary_password_clears_profile_flag(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.must_change_password = True
|
|
profile.save(update_fields=["must_change_password"])
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/change-temporary-password/",
|
|
data=json.dumps(
|
|
{
|
|
"currentPassword": "temporary-password",
|
|
"newPassword": "New-Temporary-99",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
profile.refresh_from_db()
|
|
self.assertFalse(profile.must_change_password)
|
|
|
|
def test_change_temporary_password_rejects_weak_password(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
owner.employee_profile.must_change_password = True
|
|
owner.employee_profile.save(update_fields=["must_change_password"])
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/change-temporary-password/",
|
|
data=json.dumps({"currentPassword": "temporary-password", "newPassword": "onlyletters"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
owner.employee_profile.refresh_from_db()
|
|
self.assertTrue(owner.employee_profile.must_change_password)
|
|
|
|
def test_profile_update_changes_current_user_identity(self) -> None:
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/profile/update/",
|
|
data=json.dumps({"fullName": "Иван Петров", "email": "ivan@edevs.tech"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
owner = HumanUser.objects.get(id=response.json()["user"]["id"])
|
|
self.assertEqual(owner.full_name, "Иван Петров")
|
|
self.assertEqual(owner.email, "ivan@edevs.tech")
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.profile_updated").exists())
|
|
|
|
def test_profile_password_changes_current_user_password(self) -> None:
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/profile/password/",
|
|
data=json.dumps({"currentPassword": "temporary-password", "newPassword": "New-Profile-99"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
self.assertTrue(owner.check_password("New-Profile-99"))
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.profile_password_changed").exists())
|
|
|
|
def test_profile_totp_start_marks_setup_required(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.totp_required = False
|
|
profile.totp_enabled = False
|
|
profile.totp_secret = "JBSWY3DPEHPK3PXP"
|
|
profile.save(update_fields=["totp_required", "totp_enabled", "totp_secret"])
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
response = self.client.post("/api/v1/auth/profile/totp/start/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
profile.refresh_from_db()
|
|
self.assertTrue(profile.totp_required)
|
|
self.assertFalse(profile.totp_enabled)
|
|
self.assertEqual(profile.totp_secret, "")
|
|
|
|
def test_profile_totp_disable_requires_password_and_revokes_other_sessions(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.totp_enabled = True
|
|
profile.totp_secret = "JBSWY3DPEHPK3PXP"
|
|
profile.save(update_fields=["totp_enabled", "totp_secret"])
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
other_session = SessionStore()
|
|
other_session["_auth_user_id"] = str(owner.id)
|
|
other_session["_auth_user_backend"] = "django.contrib.auth.backends.ModelBackend"
|
|
other_session["_auth_user_hash"] = owner.get_session_auth_hash()
|
|
other_session.save()
|
|
|
|
response = self.client.post(
|
|
"/api/v1/auth/profile/totp/disable/",
|
|
data=json.dumps({"currentPassword": "temporary-password"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["revoked"], 1)
|
|
profile.refresh_from_db()
|
|
self.assertFalse(profile.totp_required)
|
|
self.assertFalse(profile.totp_enabled)
|
|
self.assertEqual(profile.totp_secret, "")
|
|
|
|
def test_totp_setup_and_confirm_enables_profile_totp(self) -> None:
|
|
# TOTP по умолчанию не требуется; включаем требование, чтобы пройти setup→confirm.
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.totp_required = True
|
|
profile.save(update_fields=["totp_required"])
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
setup_response = self.client.get("/api/v1/auth/totp/setup/")
|
|
|
|
self.assertEqual(setup_response.status_code, 200)
|
|
secret = setup_response.json()["secret"]
|
|
self.assertTrue(secret)
|
|
self.assertIn("otpauth://totp/", setup_response.json()["otpauthUrl"])
|
|
|
|
confirm_response = self.client.post(
|
|
"/api/v1/auth/totp/confirm/",
|
|
data=json.dumps({"code": _totp_code(secret)}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(confirm_response.status_code, 200)
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
self.assertTrue(owner.employee_profile.totp_enabled)
|
|
|
|
def test_enabled_totp_requires_second_factor_before_session(self) -> None:
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.totp_secret = "JBSWY3DPEHPK3PXP"
|
|
profile.totp_enabled = True
|
|
profile.save(update_fields=["totp_secret", "totp_enabled"])
|
|
|
|
login_response = self.client.post(
|
|
"/api/v1/auth/login/",
|
|
data=json.dumps({"email": "owner@edevs.tech", "password": "temporary-password"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(login_response.status_code, 200)
|
|
self.assertFalse(login_response.json()["authenticated"])
|
|
self.assertTrue(login_response.json()["totpRequired"])
|
|
|
|
verify_response = self.client.post(
|
|
"/api/v1/auth/totp/verify/",
|
|
data=json.dumps({"code": _totp_code(profile.totp_secret)}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(verify_response.status_code, 200)
|
|
self.assertTrue(verify_response.json()["authenticated"])
|
|
|
|
|
|
class DjangoAdminTests(TestCase):
|
|
def test_bootstrapped_owner_can_access_django_admin(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
client = Client()
|
|
self.assertTrue(client.login(username="owner@edevs.tech", password="temporary-password"))
|
|
|
|
response = client.get("/admin/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
|
|
class EmployeeEndpointTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
self.organization = Organization.objects.get(slug="edevs")
|
|
self.sales = Department.objects.get(code="sales")
|
|
self.client = APIClient()
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
def test_owner_creates_operator_with_temporary_password(self) -> None:
|
|
response = self.client.post(
|
|
"/api/v1/employees/operators/",
|
|
data=json.dumps(
|
|
{
|
|
"email": "operator@edevs.tech",
|
|
"fullName": "Operator",
|
|
"temporaryPassword": "operator-password",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
operator = HumanUser.objects.get(email="operator@edevs.tech")
|
|
self.assertTrue(operator.check_password("operator-password"))
|
|
self.assertEqual(operator.employee_profile.role, EmployeeRole.OPERATOR)
|
|
self.assertTrue(operator.employee_profile.must_change_password)
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.operator_created").exists())
|
|
|
|
def test_operator_cannot_create_operator(self) -> None:
|
|
operator = HumanUser.objects.create_user(email="operator@edevs.tech", password="operator-password")
|
|
EmployeeProfile.objects.create(
|
|
user=operator,
|
|
organization=self.organization,
|
|
role=EmployeeRole.OPERATOR,
|
|
department=self.sales,
|
|
)
|
|
self.client.logout()
|
|
self.client.login(username="operator@edevs.tech", password="operator-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/employees/operators/",
|
|
data=json.dumps(
|
|
{
|
|
"email": "another@edevs.tech",
|
|
"temporaryPassword": "operator-password",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.owner_permission_denied").exists())
|
|
|
|
def test_owner_blocks_operator(self) -> None:
|
|
operator = HumanUser.objects.create_user(email="operator@edevs.tech", password="operator-password")
|
|
EmployeeProfile.objects.create(
|
|
user=operator,
|
|
organization=self.organization,
|
|
role=EmployeeRole.OPERATOR,
|
|
department=self.sales,
|
|
)
|
|
|
|
response = self.client.post(f"/api/v1/employees/{operator.id}/block/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
operator.refresh_from_db()
|
|
self.assertFalse(operator.is_active)
|
|
self.assertTrue(operator.employee_profile.is_blocked)
|
|
self.assertTrue(AuditEvent.objects.filter(action="identity.operator_blocked").exists())
|
|
|
|
def test_owner_updates_operator_card_fields(self) -> None:
|
|
operator = HumanUser.objects.get(email="a.kotova@edevs.tech")
|
|
|
|
response = self.client.post(
|
|
f"/api/v1/employees/{operator.id}/update/",
|
|
data=json.dumps(
|
|
{
|
|
"fullName": "Анна Котова",
|
|
"email": "anna.kotova@edevs.tech",
|
|
"phone": "+7 916 245 14 03",
|
|
"role": EmployeeRole.OPERATOR,
|
|
"department": "sales",
|
|
"totpEnabled": True,
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
operator.refresh_from_db()
|
|
operator.employee_profile.refresh_from_db()
|
|
self.assertEqual(operator.email, "anna.kotova@edevs.tech")
|
|
self.assertEqual(operator.employee_profile.phone, "+7 916 245 14 03")
|
|
self.assertTrue(operator.employee_profile.totp_enabled)
|
|
|
|
def test_owner_resets_operator_password(self) -> None:
|
|
operator = HumanUser.objects.get(email="a.kotova@edevs.tech")
|
|
|
|
response = self.client.post(f"/api/v1/employees/{operator.id}/reset-password/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
operator.refresh_from_db()
|
|
operator.employee_profile.refresh_from_db()
|
|
self.assertTrue(operator.check_password(payload["temporaryPassword"]))
|
|
self.assertTrue(operator.employee_profile.must_change_password)
|
|
|
|
def test_owner_unblocks_operator(self) -> None:
|
|
operator = HumanUser.objects.get(email="a.kotova@edevs.tech")
|
|
operator.employee_profile.block()
|
|
|
|
response = self.client.post(f"/api/v1/employees/{operator.id}/unblock/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
operator.refresh_from_db()
|
|
operator.employee_profile.refresh_from_db()
|
|
self.assertTrue(operator.is_active)
|
|
self.assertFalse(operator.employee_profile.is_blocked)
|
|
|
|
|
|
class CompanyEndpointTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
self.organization = Organization.objects.get(slug="edevs")
|
|
self.sales = Department.objects.get(code="sales")
|
|
self.client = APIClient()
|
|
self.client.login(username="owner@edevs.tech", password="temporary-password")
|
|
|
|
def test_owner_reads_departments_and_products(self) -> None:
|
|
departments_response = self.client.get("/api/v1/company/departments/")
|
|
products_response = self.client.get("/api/v1/company/products/")
|
|
|
|
self.assertEqual(departments_response.status_code, 200)
|
|
self.assertEqual(products_response.status_code, 200)
|
|
self.assertEqual(
|
|
set(d["code"] for d in departments_response.json()["items"]),
|
|
{"sales", "support"},
|
|
)
|
|
self.assertEqual(
|
|
set(product["code"] for product in products_response.json()["items"]),
|
|
{"firepage", "foxray"},
|
|
)
|
|
|
|
def test_owner_creates_and_deactivates_product(self) -> None:
|
|
create_response = self.client.post(
|
|
"/api/v1/company/products/create/",
|
|
data=json.dumps({"code": "academy", "name": "Academy"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(create_response.status_code, 201)
|
|
product_id = create_response.json()["product"]["id"]
|
|
|
|
deactivate_response = self.client.post(f"/api/v1/company/products/{product_id}/deactivate/")
|
|
|
|
self.assertEqual(deactivate_response.status_code, 200)
|
|
self.assertEqual(deactivate_response.json()["product"]["status"], "DISABLED")
|
|
self.assertTrue(AuditEvent.objects.filter(action="products.product_created").exists())
|
|
self.assertTrue(AuditEvent.objects.filter(action="products.product_disabled").exists())
|
|
|
|
def test_operator_cannot_create_product(self) -> None:
|
|
operator = HumanUser.objects.create_user(email="operator@edevs.tech", password="operator-password")
|
|
EmployeeProfile.objects.create(
|
|
user=operator,
|
|
organization=self.organization,
|
|
role=EmployeeRole.OPERATOR,
|
|
department=self.sales,
|
|
)
|
|
self.client.logout()
|
|
self.client.login(username="operator@edevs.tech", password="operator-password")
|
|
|
|
response = self.client.post(
|
|
"/api/v1/company/products/create/",
|
|
data=json.dumps({"code": "academy", "name": "Academy"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
|
|
class TotpSecretEncryptionTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
|
|
def test_totp_secret_is_encrypted_at_rest_and_decrypted_on_load(self) -> None:
|
|
from django.db import connection
|
|
|
|
from hub_platform.identity.crypto import decrypt_secret
|
|
|
|
owner = HumanUser.objects.get(email="owner@edevs.tech")
|
|
profile = owner.employee_profile
|
|
profile.totp_secret = "JBSWY3DPEHPK3PXP"
|
|
profile.save(update_fields=["totp_secret"])
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT totp_secret FROM identity_employeeprofile WHERE user_id = %s", [owner.id])
|
|
stored = cursor.fetchone()[0]
|
|
|
|
self.assertNotEqual(stored, "JBSWY3DPEHPK3PXP")
|
|
self.assertEqual(decrypt_secret(stored), "JBSWY3DPEHPK3PXP")
|
|
|
|
profile.refresh_from_db()
|
|
self.assertEqual(profile.totp_secret, "JBSWY3DPEHPK3PXP")
|
|
|
|
|
|
@override_settings(CACHES=_LOCMEM_CACHE)
|
|
class ThrottlingTests(TestCase):
|
|
def setUp(self) -> None:
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
self.client = Client()
|
|
|
|
def _login(self):
|
|
return self.client.post(
|
|
"/api/v1/auth/login/",
|
|
data=json.dumps({"email": "owner@edevs.tech", "password": "wrong"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
def test_login_endpoint_is_rate_limited(self) -> None:
|
|
# DRF биндит THROTTLE_RATES на импорте, поэтому ставим лимит напрямую.
|
|
with mock.patch.dict(ScopedRateThrottle.THROTTLE_RATES, {"login": "1/min"}):
|
|
first = self._login()
|
|
second = self._login()
|
|
|
|
self.assertEqual(first.status_code, 401)
|
|
self.assertEqual(second.status_code, 429)
|
|
|
|
|
|
class SeedIdempotencyTests(TestCase):
|
|
"""seed_hub_initial_data — посев только при первичной установке: на уже
|
|
развёрнутой установке (OWNER существует) команда должна быть no-op, чтобы
|
|
не затирать данные, изменённые через UI/API (см. seed_catalog и цены офферов).
|
|
"""
|
|
|
|
def test_seed_skipped_when_owner_exists(self) -> None:
|
|
from django.core.management import call_command
|
|
from io import StringIO
|
|
|
|
from hub_platform.products.models import Offer
|
|
|
|
bootstrap_edevs_owner(email="owner@edevs.tech", password="temporary-password")
|
|
offers_before = Offer.objects.count()
|
|
|
|
out = StringIO()
|
|
call_command("seed_hub_initial_data", "--owner-email=owner@edevs.tech", "--owner-password=temporary-password", stdout=out)
|
|
|
|
self.assertIn("seed skipped", out.getvalue())
|
|
# Ни новые офферы, ни цены не создаются и не изменяются на развёрнутой установке.
|
|
self.assertEqual(Offer.objects.count(), offers_before)
|