You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

108 lines
4.2 KiB

"""Кто видит тест: автор + назначенные пользователи (порт `services/testAccessService.js`)."""
from __future__ import annotations
from dataclasses import dataclass
from sqlalchemy import text
from ..db import get_engine
def is_test_author(created_by, user_id) -> bool:
"""`tests.created_by` — UUID. Сравниваем по строковому представлению."""
if created_by is None or user_id is None:
return False
return str(created_by) == str(user_id)
@dataclass
class AccessResult:
ok: bool
is_author: bool
not_found: bool
def user_has_test_access(user_id: str, test_id: str) -> AccessResult:
eng = get_engine()
with eng.connect() as conn:
row = conn.execute(
text('SELECT created_by FROM tests WHERE id = :id'),
{'id': test_id},
).mappings().first()
if not row:
return AccessResult(ok=False, is_author=False, not_found=True)
if is_test_author(row['created_by'], user_id):
return AccessResult(ok=True, is_author=True, not_found=False)
ar = conn.execute(
text(
"""
SELECT 1
FROM test_assignments ta
INNER JOIN test_versions tv_a ON tv_a.id = ta.test_version_id
INNER JOIN test_assignment_targets tat ON tat.assignment_id = ta.id
WHERE tv_a.test_id = :test_id
AND tat.target_type = 'user'
AND tat.target_id = :user_id
LIMIT 1
"""
),
{'test_id': test_id, 'user_id': user_id},
).first()
return AccessResult(ok=ar is not None, is_author=False, not_found=False)
def list_visible_tests(user_id: str) -> list[dict]:
"""Каталог: только активная цепочка + (автор OR назначен)."""
eng = get_engine()
with eng.connect() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT t.id, t.title, t.description, t.is_active AS chain_active,
t.created_at, t.updated_at,
tv.id AS active_version_id, tv.version,
t.created_by, u.full_name AS author_full_name
FROM tests t
INNER JOIN test_versions tv ON tv.test_id = t.id AND tv.is_active = true
INNER JOIN users u ON u.id = t.created_by
WHERE t.is_active = true
AND (
t.created_by = :uid
OR EXISTS (
SELECT 1
FROM test_assignments ta
INNER JOIN test_versions tv2 ON tv2.id = ta.test_version_id
INNER JOIN test_assignment_targets tat ON tat.assignment_id = ta.id
WHERE tv2.test_id = t.id
AND tat.target_type = 'user'
AND tat.target_id = :uid
)
)
ORDER BY t.updated_at DESC NULLS LAST, t.created_at DESC
"""
),
{'uid': user_id},
).mappings().all()
return [dict(r) for r in rows]
def list_hidden_by_author(user_id: str) -> list[dict]:
"""Скрытые автором цепочки (`is_active = false`) — видны только автору."""
eng = get_engine()
with eng.connect() as conn:
rows = conn.execute(
text(
"""
SELECT t.id, t.title, t.description, t.is_active AS chain_active,
t.created_at, t.updated_at, tv.id AS active_version_id, tv.version,
t.created_by, u.full_name AS author_full_name
FROM tests t
INNER JOIN test_versions tv ON tv.test_id = t.id AND tv.is_active = true
INNER JOIN users u ON u.id = t.created_by
WHERE t.is_active = false AND t.created_by = :uid
ORDER BY t.updated_at DESC NULLS LAST, t.created_at DESC
"""
),
{'uid': user_id},
).mappings().all()
return [dict(r) for r in rows]