mirror of
https://github.com/TronoSfera/Law.git
synced 2026-05-18 18:13:46 +03:00
276 lines
11 KiB
Python
276 lines
11 KiB
Python
import os
|
||
import unittest
|
||
from datetime import timedelta
|
||
from unittest.mock import patch
|
||
from uuid import UUID
|
||
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import create_engine, delete
|
||
from sqlalchemy.orm import sessionmaker
|
||
from sqlalchemy.pool import StaticPool
|
||
|
||
# Ensure settings can be initialized in test environments
|
||
os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:")
|
||
os.environ.setdefault("REDIS_URL", "redis://localhost:6379/0")
|
||
os.environ.setdefault("S3_ENDPOINT", "http://localhost:9000")
|
||
os.environ.setdefault("S3_ACCESS_KEY", "test")
|
||
os.environ.setdefault("S3_SECRET_KEY", "test")
|
||
os.environ.setdefault("S3_BUCKET", "test")
|
||
|
||
from app.main import app
|
||
from app.core.config import settings
|
||
from app.core.security import create_jwt, decode_jwt
|
||
from app.db.session import get_db
|
||
from app.models.notification import Notification
|
||
from app.models.otp_session import OtpSession
|
||
from app.models.request import Request
|
||
from app.models.topic_required_field import TopicRequiredField
|
||
|
||
|
||
class PublicRequestCreateTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.engine = create_engine(
|
||
"sqlite+pysqlite:///:memory:",
|
||
connect_args={"check_same_thread": False},
|
||
poolclass=StaticPool,
|
||
)
|
||
cls.SessionLocal = sessionmaker(bind=cls.engine, autocommit=False, autoflush=False)
|
||
Request.__table__.create(bind=cls.engine)
|
||
Notification.__table__.create(bind=cls.engine)
|
||
OtpSession.__table__.create(bind=cls.engine)
|
||
TopicRequiredField.__table__.create(bind=cls.engine)
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
Notification.__table__.drop(bind=cls.engine)
|
||
OtpSession.__table__.drop(bind=cls.engine)
|
||
TopicRequiredField.__table__.drop(bind=cls.engine)
|
||
Request.__table__.drop(bind=cls.engine)
|
||
cls.engine.dispose()
|
||
|
||
def setUp(self):
|
||
with self.SessionLocal() as db:
|
||
db.execute(delete(Notification))
|
||
db.execute(delete(OtpSession))
|
||
db.execute(delete(TopicRequiredField))
|
||
db.execute(delete(Request))
|
||
db.commit()
|
||
|
||
def override_get_db():
|
||
db = self.SessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
app.dependency_overrides[get_db] = override_get_db
|
||
self.client = TestClient(app)
|
||
|
||
def tearDown(self):
|
||
self.client.close()
|
||
app.dependency_overrides.clear()
|
||
|
||
def _send_and_verify_create_otp(self, phone: str) -> None:
|
||
with patch("app.api.public.otp._generate_code", return_value="123456"):
|
||
sent = self.client.post(
|
||
"/api/public/otp/send",
|
||
json={"purpose": "CREATE_REQUEST", "client_phone": phone},
|
||
)
|
||
self.assertEqual(sent.status_code, 200)
|
||
body = sent.json()
|
||
self.assertEqual(body["status"], "sent")
|
||
self.assertEqual(body["sms_response"]["provider"], "mock_sms")
|
||
|
||
verified = self.client.post(
|
||
"/api/public/otp/verify",
|
||
json={"purpose": "CREATE_REQUEST", "client_phone": phone, "code": "123456"},
|
||
)
|
||
self.assertEqual(verified.status_code, 200)
|
||
self.assertEqual(verified.json()["status"], "verified")
|
||
|
||
def test_create_request_requires_verified_otp_cookie(self):
|
||
payload = {
|
||
"client_name": "ООО Ромашка",
|
||
"client_phone": "+79990000001",
|
||
"topic_code": "consulting",
|
||
"description": "Тестируем создание заявки",
|
||
"extra_fields": {"referral_name": "Партнер"},
|
||
}
|
||
response = self.client.post("/api/public/requests", json=payload)
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
self._send_and_verify_create_otp(payload["client_phone"])
|
||
|
||
response = self.client.post("/api/public/requests", json=payload)
|
||
self.assertEqual(response.status_code, 201)
|
||
body = response.json()
|
||
self.assertTrue(body["track_number"].startswith("TRK-"))
|
||
self.assertFalse(body["otp_required"])
|
||
request_id = UUID(body["request_id"])
|
||
|
||
with self.SessionLocal() as db:
|
||
created = db.get(Request, request_id)
|
||
self.assertIsNotNone(created)
|
||
self.assertEqual(created.client_name, payload["client_name"])
|
||
self.assertEqual(created.client_phone, payload["client_phone"])
|
||
self.assertEqual(created.topic_code, payload["topic_code"])
|
||
self.assertEqual(created.description, payload["description"])
|
||
self.assertEqual(created.extra_fields, payload["extra_fields"])
|
||
self.assertEqual(created.status_code, "NEW")
|
||
self.assertEqual(created.track_number, body["track_number"])
|
||
self.assertEqual(created.responsible, "Клиент")
|
||
|
||
# After creation, cookie is switched to VIEW_REQUEST for this track.
|
||
read = self.client.get(f"/api/public/requests/{body['track_number']}")
|
||
self.assertEqual(read.status_code, 200)
|
||
self.assertEqual(read.json()["track_number"], body["track_number"])
|
||
|
||
def test_view_request_requires_view_otp_and_uses_track_cookie(self):
|
||
with self.SessionLocal() as db:
|
||
row = Request(
|
||
track_number="TRK-VIEW-OTP",
|
||
client_name="Клиент",
|
||
client_phone="+79991112233",
|
||
topic_code="consulting",
|
||
status_code="NEW",
|
||
description="Проверка просмотра",
|
||
extra_fields={},
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
|
||
no_session = self.client.get("/api/public/requests/TRK-VIEW-OTP")
|
||
self.assertEqual(no_session.status_code, 401)
|
||
|
||
with patch("app.api.public.otp._generate_code", return_value="654321"):
|
||
sent = self.client.post(
|
||
"/api/public/otp/send",
|
||
json={"purpose": "VIEW_REQUEST", "track_number": "TRK-VIEW-OTP"},
|
||
)
|
||
self.assertEqual(sent.status_code, 200)
|
||
self.assertEqual(sent.json()["status"], "sent")
|
||
|
||
wrong_code = self.client.post(
|
||
"/api/public/otp/verify",
|
||
json={"purpose": "VIEW_REQUEST", "track_number": "TRK-VIEW-OTP", "code": "000000"},
|
||
)
|
||
self.assertEqual(wrong_code.status_code, 400)
|
||
|
||
verified = self.client.post(
|
||
"/api/public/otp/verify",
|
||
json={"purpose": "VIEW_REQUEST", "track_number": "TRK-VIEW-OTP", "code": "654321"},
|
||
)
|
||
self.assertEqual(verified.status_code, 200)
|
||
|
||
ok = self.client.get("/api/public/requests/TRK-VIEW-OTP")
|
||
self.assertEqual(ok.status_code, 200)
|
||
self.assertEqual(ok.json()["track_number"], "TRK-VIEW-OTP")
|
||
|
||
denied_other_track = self.client.get("/api/public/requests/TRK-OTHER")
|
||
self.assertEqual(denied_other_track.status_code, 403)
|
||
|
||
def test_open_request_marks_client_updates_as_read(self):
|
||
with self.SessionLocal() as db:
|
||
row = Request(
|
||
track_number="TRK-READ-1",
|
||
client_name="Клиент",
|
||
client_phone="+79995550011",
|
||
topic_code="consulting",
|
||
status_code="IN_PROGRESS",
|
||
description="Проверка чтения",
|
||
extra_fields={},
|
||
client_has_unread_updates=True,
|
||
client_unread_event_type="STATUS",
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
request_id = row.id
|
||
|
||
public_token = create_jwt({"sub": "TRK-READ-1", "purpose": "VIEW_REQUEST"}, settings.PUBLIC_JWT_SECRET, timedelta(days=1))
|
||
cookies = {settings.PUBLIC_COOKIE_NAME: public_token}
|
||
|
||
opened = self.client.get("/api/public/requests/TRK-READ-1", cookies=cookies)
|
||
self.assertEqual(opened.status_code, 200)
|
||
body = opened.json()
|
||
self.assertFalse(body["client_has_unread_updates"])
|
||
self.assertIsNone(body["client_unread_event_type"])
|
||
|
||
with self.SessionLocal() as db:
|
||
refreshed = db.get(Request, request_id)
|
||
self.assertIsNotNone(refreshed)
|
||
self.assertFalse(refreshed.client_has_unread_updates)
|
||
self.assertIsNone(refreshed.client_unread_event_type)
|
||
|
||
def test_create_request_checks_required_topic_fields(self):
|
||
phone = "+79990000005"
|
||
self._send_and_verify_create_otp(phone)
|
||
|
||
with self.SessionLocal() as db:
|
||
db.add(
|
||
TopicRequiredField(
|
||
topic_code="consulting",
|
||
field_key="passport_series",
|
||
required=True,
|
||
enabled=True,
|
||
sort_order=1,
|
||
responsible="root@example.com",
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
missing = self.client.post(
|
||
"/api/public/requests",
|
||
json={
|
||
"client_name": "ООО Поле",
|
||
"client_phone": phone,
|
||
"topic_code": "consulting",
|
||
"description": "Проверка обязательного поля",
|
||
"extra_fields": {},
|
||
},
|
||
)
|
||
self.assertEqual(missing.status_code, 400)
|
||
self.assertIn("passport_series", missing.json().get("detail", ""))
|
||
|
||
created = self.client.post(
|
||
"/api/public/requests",
|
||
json={
|
||
"client_name": "ООО Поле",
|
||
"client_phone": phone,
|
||
"topic_code": "consulting",
|
||
"description": "Проверка обязательного поля",
|
||
"extra_fields": {"passport_series": "1234"},
|
||
},
|
||
)
|
||
self.assertEqual(created.status_code, 201)
|
||
self.assertTrue(created.json()["track_number"].startswith("TRK-"))
|
||
|
||
def test_verify_otp_sets_public_cookie_for_configured_ttl(self):
|
||
phone = "+79990001234"
|
||
with patch("app.api.public.otp._generate_code", return_value="777777"):
|
||
sent = self.client.post(
|
||
"/api/public/otp/send",
|
||
json={"purpose": "CREATE_REQUEST", "client_phone": phone},
|
||
)
|
||
self.assertEqual(sent.status_code, 200)
|
||
|
||
verified = self.client.post(
|
||
"/api/public/otp/verify",
|
||
json={"purpose": "CREATE_REQUEST", "client_phone": phone, "code": "777777"},
|
||
)
|
||
self.assertEqual(verified.status_code, 200)
|
||
|
||
token = verified.cookies.get(settings.PUBLIC_COOKIE_NAME)
|
||
self.assertTrue(token)
|
||
payload = decode_jwt(token, settings.PUBLIC_JWT_SECRET)
|
||
self.assertEqual(payload.get("sub"), phone)
|
||
self.assertEqual(payload.get("purpose"), "CREATE_REQUEST")
|
||
self.assertEqual(
|
||
int(payload.get("exp") or 0) - int(payload.get("iat") or 0),
|
||
settings.PUBLIC_JWT_TTL_DAYS * 24 * 3600,
|
||
)
|
||
|
||
cookie_header = str(verified.headers.get("set-cookie") or "")
|
||
self.assertIn(f"{settings.PUBLIC_COOKIE_NAME}=", cookie_header)
|
||
self.assertIn(f"Max-Age={settings.PUBLIC_JWT_TTL_DAYS * 24 * 3600}", cookie_header)
|
||
self.assertIn("httponly", cookie_header.lower())
|