Spaces:
Running
Running
GitHub Action commited on
Commit ·
c1cf583
1
Parent(s): 3bd7688
deploy from github actions
Browse files- app/routers/alabanza.py +6 -0
- app/routers/lyrics.py +76 -21
- app/routers/songs.py +43 -8
- tests/test_group_permissions.py +355 -0
- tests/verify_group_permissions.py +205 -0
app/routers/alabanza.py
CHANGED
|
@@ -604,6 +604,8 @@ def get_grupo_pistas(grupo_id: int, db: Session = Depends(get_db), current_user:
|
|
| 604 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 605 |
if not ug:
|
| 606 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
|
|
|
|
|
|
| 607 |
|
| 608 |
tasks = db.query(Task).filter(Task.groups_shared.any(Group.id == grupo_id)).all()
|
| 609 |
return [
|
|
@@ -630,6 +632,8 @@ def get_grupo_letras(grupo_id: int, db: Session = Depends(get_db), current_user:
|
|
| 630 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 631 |
if not ug:
|
| 632 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
|
|
|
|
|
|
| 633 |
|
| 634 |
lyrics = db.query(Lyric).filter(Lyric.groups_shared.any(Group.id == grupo_id)).all()
|
| 635 |
return [
|
|
@@ -651,6 +655,8 @@ def get_grupo_repertorios(grupo_id: int, db: Session = Depends(get_db), current_
|
|
| 651 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 652 |
if not ug:
|
| 653 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
|
|
|
|
|
|
| 654 |
|
| 655 |
repertorios = db.query(Repertorio).filter(Repertorio.group_id == grupo_id).all()
|
| 656 |
|
|
|
|
| 604 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 605 |
if not ug:
|
| 606 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
| 607 |
+
if ug.role != "admin" and not ug.perm_ver_pistas:
|
| 608 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver las pistas de este grupo")
|
| 609 |
|
| 610 |
tasks = db.query(Task).filter(Task.groups_shared.any(Group.id == grupo_id)).all()
|
| 611 |
return [
|
|
|
|
| 632 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 633 |
if not ug:
|
| 634 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
| 635 |
+
if ug.role != "admin" and not ug.perm_ver_letras:
|
| 636 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver las letras de este grupo")
|
| 637 |
|
| 638 |
lyrics = db.query(Lyric).filter(Lyric.groups_shared.any(Group.id == grupo_id)).all()
|
| 639 |
return [
|
|
|
|
| 655 |
ug = db.query(UserGroup).filter(UserGroup.user_id == current_user.id, UserGroup.group_id == grupo_id, UserGroup.estado == "activo").first()
|
| 656 |
if not ug:
|
| 657 |
raise HTTPException(status_code=403, detail="No eres miembro de este grupo")
|
| 658 |
+
if ug.role != "admin" and not ug.perm_ver_repertorios:
|
| 659 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver los repertorios de este grupo")
|
| 660 |
|
| 661 |
repertorios = db.query(Repertorio).filter(Repertorio.group_id == grupo_id).all()
|
| 662 |
|
app/routers/lyrics.py
CHANGED
|
@@ -99,10 +99,15 @@ def get_lyric(lyric_id: int, db: Session = Depends(get_db), current_user: User =
|
|
| 99 |
db_lyric = db.query(Lyric).filter(Lyric.id == lyric_id).first()
|
| 100 |
if not db_lyric:
|
| 101 |
raise HTTPException(status_code=404, detail="Lyric not found")
|
| 102 |
-
|
| 103 |
-
|
| 104 |
-
|
| 105 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 106 |
return db_lyric
|
| 107 |
|
| 108 |
@router.put("/{lyric_id}", response_model=LyricOut)
|
|
@@ -212,14 +217,24 @@ def create_repertoire(rep: RepertorioCreate, db: Session = Depends(get_db), curr
|
|
| 212 |
|
| 213 |
@router.put("/repertorios/{id}", response_model=RepertorioOut)
|
| 214 |
def update_repertoire(id: int, rep: RepertorioCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 215 |
-
db_rep = db.query(Repertorio).filter(Repertorio.id == id
|
| 216 |
if not db_rep:
|
| 217 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 218 |
|
| 219 |
-
|
| 220 |
-
|
| 221 |
-
|
| 222 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 223 |
|
| 224 |
db_rep.name = rep.name
|
| 225 |
db_rep.group_id = rep.group_id
|
|
@@ -247,14 +262,24 @@ def update_repertoire(id: int, rep: RepertorioCreate, db: Session = Depends(get_
|
|
| 247 |
|
| 248 |
@router.delete("/repertorios/{id}")
|
| 249 |
def delete_repertoire(id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 250 |
-
db_rep = db.query(Repertorio).filter(Repertorio.id == id
|
| 251 |
if not db_rep:
|
| 252 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 253 |
|
| 254 |
-
|
| 255 |
-
|
| 256 |
-
|
| 257 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 258 |
|
| 259 |
group_id = db_rep.group_id
|
| 260 |
db.query(RepertorioLyric).filter(RepertorioLyric.repertorio_id == id).delete()
|
|
@@ -270,14 +295,24 @@ def delete_repertoire(id: int, db: Session = Depends(get_db), current_user: User
|
|
| 270 |
|
| 271 |
@router.post("/repertorios/{id}/lyrics")
|
| 272 |
def add_lyric_to_repertoire(id: int, lyric_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 273 |
-
db_rep = db.query(Repertorio).filter(Repertorio.id == id
|
| 274 |
if not db_rep:
|
| 275 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 276 |
|
| 277 |
-
|
| 278 |
-
|
| 279 |
-
|
| 280 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 281 |
|
| 282 |
exists = db.query(RepertorioLyric).filter_by(repertorio_id=id, lyric_id=lyric_id).first()
|
| 283 |
if exists:
|
|
@@ -303,6 +338,20 @@ def get_repertoire(id: int, db: Session = Depends(get_db), current_user: User =
|
|
| 303 |
if not rep:
|
| 304 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 305 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 306 |
# Ordenar letras por el campo 'order'
|
| 307 |
sorted_lyrics = sorted(rep.lyrics, key=lambda x: x.order if x.order is not None else 0)
|
| 308 |
lyrics = [rl.lyric for rl in sorted_lyrics]
|
|
@@ -444,8 +493,14 @@ def share_lyric(
|
|
| 444 |
is_owner = lyric.created_by == current_user.id
|
| 445 |
is_group_admin = ug.role == "admin"
|
| 446 |
|
| 447 |
-
|
| 448 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 449 |
|
| 450 |
group = db.query(Group).filter(Group.id == data.group_id).first()
|
| 451 |
if not group:
|
|
|
|
| 99 |
db_lyric = db.query(Lyric).filter(Lyric.id == lyric_id).first()
|
| 100 |
if not db_lyric:
|
| 101 |
raise HTTPException(status_code=404, detail="Lyric not found")
|
| 102 |
+
if db_lyric.created_by != current_user.id:
|
| 103 |
+
user_groups_with_perm = db.query(UserGroup).filter(
|
| 104 |
+
UserGroup.user_id == current_user.id,
|
| 105 |
+
UserGroup.estado == "activo",
|
| 106 |
+
UserGroup.group_id.in_(db_lyric.group_ids),
|
| 107 |
+
(UserGroup.role == "admin") | (UserGroup.perm_ver_letras == True)
|
| 108 |
+
).first()
|
| 109 |
+
if not user_groups_with_perm:
|
| 110 |
+
raise HTTPException(status_code=403, detail="Not authorized to view this lyric")
|
| 111 |
return db_lyric
|
| 112 |
|
| 113 |
@router.put("/{lyric_id}", response_model=LyricOut)
|
|
|
|
| 217 |
|
| 218 |
@router.put("/repertorios/{id}", response_model=RepertorioOut)
|
| 219 |
def update_repertoire(id: int, rep: RepertorioCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 220 |
+
db_rep = db.query(Repertorio).filter(Repertorio.id == id).first()
|
| 221 |
if not db_rep:
|
| 222 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 223 |
|
| 224 |
+
is_authorized = False
|
| 225 |
+
if db_rep.created_by == current_user.id:
|
| 226 |
+
is_authorized = True
|
| 227 |
+
elif db_rep.group_id is not None:
|
| 228 |
+
ug = db.query(UserGroup).filter(
|
| 229 |
+
UserGroup.user_id == current_user.id,
|
| 230 |
+
UserGroup.group_id == db_rep.group_id,
|
| 231 |
+
UserGroup.estado == "activo"
|
| 232 |
+
).first()
|
| 233 |
+
if ug and (ug.role == "admin" or ug.perm_crud_repertorios):
|
| 234 |
+
is_authorized = True
|
| 235 |
+
|
| 236 |
+
if not is_authorized:
|
| 237 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para editar este repertorio")
|
| 238 |
|
| 239 |
db_rep.name = rep.name
|
| 240 |
db_rep.group_id = rep.group_id
|
|
|
|
| 262 |
|
| 263 |
@router.delete("/repertorios/{id}")
|
| 264 |
def delete_repertoire(id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 265 |
+
db_rep = db.query(Repertorio).filter(Repertorio.id == id).first()
|
| 266 |
if not db_rep:
|
| 267 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 268 |
|
| 269 |
+
is_authorized = False
|
| 270 |
+
if db_rep.created_by == current_user.id:
|
| 271 |
+
is_authorized = True
|
| 272 |
+
elif db_rep.group_id is not None:
|
| 273 |
+
ug = db.query(UserGroup).filter(
|
| 274 |
+
UserGroup.user_id == current_user.id,
|
| 275 |
+
UserGroup.group_id == db_rep.group_id,
|
| 276 |
+
UserGroup.estado == "activo"
|
| 277 |
+
).first()
|
| 278 |
+
if ug and (ug.role == "admin" or ug.perm_crud_repertorios):
|
| 279 |
+
is_authorized = True
|
| 280 |
+
|
| 281 |
+
if not is_authorized:
|
| 282 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para eliminar este repertorio")
|
| 283 |
|
| 284 |
group_id = db_rep.group_id
|
| 285 |
db.query(RepertorioLyric).filter(RepertorioLyric.repertorio_id == id).delete()
|
|
|
|
| 295 |
|
| 296 |
@router.post("/repertorios/{id}/lyrics")
|
| 297 |
def add_lyric_to_repertoire(id: int, lyric_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
| 298 |
+
db_rep = db.query(Repertorio).filter(Repertorio.id == id).first()
|
| 299 |
if not db_rep:
|
| 300 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 301 |
|
| 302 |
+
is_authorized = False
|
| 303 |
+
if db_rep.created_by == current_user.id:
|
| 304 |
+
is_authorized = True
|
| 305 |
+
elif db_rep.group_id is not None:
|
| 306 |
+
ug = db.query(UserGroup).filter(
|
| 307 |
+
UserGroup.user_id == current_user.id,
|
| 308 |
+
UserGroup.group_id == db_rep.group_id,
|
| 309 |
+
UserGroup.estado == "activo"
|
| 310 |
+
).first()
|
| 311 |
+
if ug and (ug.role == "admin" or ug.perm_crud_repertorios):
|
| 312 |
+
is_authorized = True
|
| 313 |
+
|
| 314 |
+
if not is_authorized:
|
| 315 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para modificar este repertorio")
|
| 316 |
|
| 317 |
exists = db.query(RepertorioLyric).filter_by(repertorio_id=id, lyric_id=lyric_id).first()
|
| 318 |
if exists:
|
|
|
|
| 338 |
if not rep:
|
| 339 |
raise HTTPException(status_code=404, detail="Repertoire not found")
|
| 340 |
|
| 341 |
+
if rep.group_id is not None:
|
| 342 |
+
ug = db.query(UserGroup).filter(
|
| 343 |
+
UserGroup.user_id == current_user.id,
|
| 344 |
+
UserGroup.group_id == rep.group_id,
|
| 345 |
+
UserGroup.estado == "activo"
|
| 346 |
+
).first()
|
| 347 |
+
if not ug:
|
| 348 |
+
raise HTTPException(status_code=403, detail="No tienes acceso a este repertorio")
|
| 349 |
+
if ug.role != "admin" and not ug.perm_ver_repertorios and rep.created_by != current_user.id:
|
| 350 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver este repertorio")
|
| 351 |
+
else:
|
| 352 |
+
if rep.created_by != current_user.id:
|
| 353 |
+
raise HTTPException(status_code=403, detail="No tienes acceso a este repertorio")
|
| 354 |
+
|
| 355 |
# Ordenar letras por el campo 'order'
|
| 356 |
sorted_lyrics = sorted(rep.lyrics, key=lambda x: x.order if x.order is not None else 0)
|
| 357 |
lyrics = [rl.lyric for rl in sorted_lyrics]
|
|
|
|
| 493 |
is_owner = lyric.created_by == current_user.id
|
| 494 |
is_group_admin = ug.role == "admin"
|
| 495 |
|
| 496 |
+
# Para "share": creador, admin, o miembro con perm_compartir_letras
|
| 497 |
+
# Para "unshare": sólo creador o admin (para no desvincular letras ajenas)
|
| 498 |
+
if data.action == "share":
|
| 499 |
+
if not is_owner and not is_group_admin and not ug.perm_compartir_letras:
|
| 500 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para compartir letras con este grupo")
|
| 501 |
+
elif data.action == "unshare":
|
| 502 |
+
if not is_owner and not is_group_admin:
|
| 503 |
+
raise HTTPException(status_code=403, detail="Solo el creador o un administrador puede desvincular esta letra del grupo")
|
| 504 |
|
| 505 |
group = db.query(Group).filter(Group.id == data.group_id).first()
|
| 506 |
if not group:
|
app/routers/songs.py
CHANGED
|
@@ -859,8 +859,14 @@ def share_song(
|
|
| 859 |
is_owner = task.user_id == current_user.id
|
| 860 |
is_group_admin = ug.role == "admin"
|
| 861 |
|
| 862 |
-
|
| 863 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 864 |
|
| 865 |
group = db.query(Group).filter(Group.id == data.group_id).first()
|
| 866 |
if not group:
|
|
@@ -969,6 +975,16 @@ def get_task_by_id(
|
|
| 969 |
if not task:
|
| 970 |
raise HTTPException(status_code=404, detail="Tarea no encontrada o sin acceso")
|
| 971 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 972 |
import json as _json
|
| 973 |
return {
|
| 974 |
"id": task.id,
|
|
@@ -1004,9 +1020,19 @@ async def get_stems(
|
|
| 1004 |
task_record = db.query(Task).filter(Task.task_id == task_id).filter(
|
| 1005 |
(Task.user_id == current_user.id) | (Task.groups_shared.any(Group.id.in_(user_group_ids)))
|
| 1006 |
).first()
|
| 1007 |
-
|
| 1008 |
if not task_record:
|
| 1009 |
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1010 |
|
| 1011 |
# Dynamic expiration healing: if files expired, clean up DB state and return 404
|
| 1012 |
from datetime import datetime
|
|
@@ -1248,13 +1274,22 @@ async def harmony_status(
|
|
| 1248 |
"""
|
| 1249 |
Consulta el estado de generación de armonías vocales de coro.
|
| 1250 |
"""
|
| 1251 |
-
|
| 1252 |
-
task_record = db.query(Task).filter(
|
| 1253 |
-
Task.
|
| 1254 |
-
Task.user_id == current_user.id
|
| 1255 |
).first()
|
| 1256 |
if not task_record:
|
| 1257 |
-
raise HTTPException(status_code=404, detail="Tarea no encontrada
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1258 |
|
| 1259 |
# 1. Comprobar si ya existen en la base de datos (B2/Tebi URLs)
|
| 1260 |
import json as _json
|
|
|
|
| 859 |
is_owner = task.user_id == current_user.id
|
| 860 |
is_group_admin = ug.role == "admin"
|
| 861 |
|
| 862 |
+
# Para "share": propietario, admin, o miembro con perm_compartir_pistas
|
| 863 |
+
# Para "unshare": sólo propietario o admin (para no quitar pistas ajenas)
|
| 864 |
+
if data.action == "share":
|
| 865 |
+
if not is_owner and not is_group_admin and not ug.perm_compartir_pistas:
|
| 866 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para compartir pistas con este grupo")
|
| 867 |
+
elif data.action == "unshare":
|
| 868 |
+
if not is_owner and not is_group_admin:
|
| 869 |
+
raise HTTPException(status_code=403, detail="Solo el propietario o un administrador puede desvincular esta pista del grupo")
|
| 870 |
|
| 871 |
group = db.query(Group).filter(Group.id == data.group_id).first()
|
| 872 |
if not group:
|
|
|
|
| 975 |
if not task:
|
| 976 |
raise HTTPException(status_code=404, detail="Tarea no encontrada o sin acceso")
|
| 977 |
|
| 978 |
+
if task.user_id != current_user.id:
|
| 979 |
+
user_groups_with_perm = db.query(UserGroup).filter(
|
| 980 |
+
UserGroup.user_id == current_user.id,
|
| 981 |
+
UserGroup.estado == "activo",
|
| 982 |
+
UserGroup.group_id.in_([g.id for g in task.groups_shared]),
|
| 983 |
+
(UserGroup.role == "admin") | (UserGroup.perm_ver_pistas == True)
|
| 984 |
+
).first()
|
| 985 |
+
if not user_groups_with_perm:
|
| 986 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver esta pista")
|
| 987 |
+
|
| 988 |
import json as _json
|
| 989 |
return {
|
| 990 |
"id": task.id,
|
|
|
|
| 1020 |
task_record = db.query(Task).filter(Task.task_id == task_id).filter(
|
| 1021 |
(Task.user_id == current_user.id) | (Task.groups_shared.any(Group.id.in_(user_group_ids)))
|
| 1022 |
).first()
|
| 1023 |
+
|
| 1024 |
if not task_record:
|
| 1025 |
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
| 1026 |
+
|
| 1027 |
+
if task_record.user_id != current_user.id:
|
| 1028 |
+
user_groups_with_perm = db.query(UserGroup).filter(
|
| 1029 |
+
UserGroup.user_id == current_user.id,
|
| 1030 |
+
UserGroup.estado == "activo",
|
| 1031 |
+
UserGroup.group_id.in_([g.id for g in task_record.groups_shared]),
|
| 1032 |
+
(UserGroup.role == "admin") | (UserGroup.perm_ver_pistas == True)
|
| 1033 |
+
).first()
|
| 1034 |
+
if not user_groups_with_perm:
|
| 1035 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para acceder a los stems de esta pista")
|
| 1036 |
|
| 1037 |
# Dynamic expiration healing: if files expired, clean up DB state and return 404
|
| 1038 |
from datetime import datetime
|
|
|
|
| 1274 |
"""
|
| 1275 |
Consulta el estado de generación de armonías vocales de coro.
|
| 1276 |
"""
|
| 1277 |
+
user_group_ids = [ug.group_id for ug in current_user.groups if ug.estado == 'activo']
|
| 1278 |
+
task_record = db.query(Task).filter(Task.task_id == task_id).filter(
|
| 1279 |
+
(Task.user_id == current_user.id) | (Task.groups_shared.any(Group.id.in_(user_group_ids)))
|
|
|
|
| 1280 |
).first()
|
| 1281 |
if not task_record:
|
| 1282 |
+
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
| 1283 |
+
|
| 1284 |
+
if task_record.user_id != current_user.id:
|
| 1285 |
+
user_groups_with_perm = db.query(UserGroup).filter(
|
| 1286 |
+
UserGroup.user_id == current_user.id,
|
| 1287 |
+
UserGroup.estado == "activo",
|
| 1288 |
+
UserGroup.group_id.in_([g.id for g in task_record.groups_shared]),
|
| 1289 |
+
(UserGroup.role == "admin") | (UserGroup.perm_ver_pistas == True)
|
| 1290 |
+
).first()
|
| 1291 |
+
if not user_groups_with_perm:
|
| 1292 |
+
raise HTTPException(status_code=403, detail="No tienes permisos para ver esta pista")
|
| 1293 |
|
| 1294 |
# 1. Comprobar si ya existen en la base de datos (B2/Tebi URLs)
|
| 1295 |
import json as _json
|
tests/test_group_permissions.py
ADDED
|
@@ -0,0 +1,355 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
import sys
|
| 2 |
+
import os
|
| 3 |
+
import unittest
|
| 4 |
+
from unittest.mock import MagicMock
|
| 5 |
+
|
| 6 |
+
# Force environment variables for tests
|
| 7 |
+
os.environ["SUPABASE_JWT_SECRET"] = "supersecretkey_change_me_in_production"
|
| 8 |
+
os.environ["SECRET_KEY"] = "supersecretkey_change_me_in_production"
|
| 9 |
+
os.environ["REDIS_URL"] = "rpc://"
|
| 10 |
+
|
| 11 |
+
# Mock heavy dependencies
|
| 12 |
+
sys.modules['librosa'] = MagicMock()
|
| 13 |
+
sys.modules['torch'] = MagicMock()
|
| 14 |
+
sys.modules['yt_dlp'] = MagicMock()
|
| 15 |
+
sys.modules['pretty_midi'] = MagicMock()
|
| 16 |
+
sys.modules['numpy'] = MagicMock()
|
| 17 |
+
sys.modules['demucs'] = MagicMock()
|
| 18 |
+
sys.modules['demucs.separate'] = MagicMock()
|
| 19 |
+
|
| 20 |
+
# Include MelodixAPI in python path
|
| 21 |
+
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
| 22 |
+
|
| 23 |
+
from fastapi.testclient import TestClient
|
| 24 |
+
from sqlalchemy import create_engine
|
| 25 |
+
from sqlalchemy.orm import sessionmaker
|
| 26 |
+
|
| 27 |
+
from app.database import Base, get_db
|
| 28 |
+
from app.models.user import User, Device
|
| 29 |
+
from app.models.lyrics import Group, UserGroup, Lyric, Repertorio
|
| 30 |
+
from app.models.process import Task
|
| 31 |
+
from main import app
|
| 32 |
+
|
| 33 |
+
TEST_DATABASE_URL = "sqlite:///./test_group_perms.db"
|
| 34 |
+
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
|
| 35 |
+
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
| 36 |
+
|
| 37 |
+
def override_get_db():
|
| 38 |
+
db = TestingSessionLocal()
|
| 39 |
+
try:
|
| 40 |
+
yield db
|
| 41 |
+
finally:
|
| 42 |
+
db.close()
|
| 43 |
+
|
| 44 |
+
app.dependency_overrides[get_db] = override_get_db
|
| 45 |
+
|
| 46 |
+
class TestGroupPermissions(unittest.TestCase):
|
| 47 |
+
|
| 48 |
+
@classmethod
|
| 49 |
+
def setUpClass(cls):
|
| 50 |
+
Base.metadata.create_all(bind=engine)
|
| 51 |
+
cls.client = TestClient(app)
|
| 52 |
+
|
| 53 |
+
@classmethod
|
| 54 |
+
def tearDownClass(cls):
|
| 55 |
+
Base.metadata.drop_all(bind=engine)
|
| 56 |
+
if os.path.exists("./test_group_perms.db"):
|
| 57 |
+
try:
|
| 58 |
+
os.remove("./test_group_perms.db")
|
| 59 |
+
except Exception:
|
| 60 |
+
pass
|
| 61 |
+
|
| 62 |
+
def setUp(self):
|
| 63 |
+
db = TestingSessionLocal()
|
| 64 |
+
db.query(Task).delete()
|
| 65 |
+
db.query(UserGroup).delete()
|
| 66 |
+
db.query(Lyric).delete()
|
| 67 |
+
db.query(Repertorio).delete()
|
| 68 |
+
db.query(Group).delete()
|
| 69 |
+
db.query(Device).delete()
|
| 70 |
+
db.query(User).delete()
|
| 71 |
+
db.commit()
|
| 72 |
+
db.close()
|
| 73 |
+
|
| 74 |
+
def _create_user_and_token(self, username, email):
|
| 75 |
+
payload = {"username": username, "email": email, "password": "password123"}
|
| 76 |
+
self.client.post("/auth/register", json=payload)
|
| 77 |
+
|
| 78 |
+
login_res = self.client.post("/auth/token", data={
|
| 79 |
+
"username": email,
|
| 80 |
+
"password": "password123",
|
| 81 |
+
"device_id": f"dev_{username}",
|
| 82 |
+
"device_name": "Test Device"
|
| 83 |
+
})
|
| 84 |
+
token = login_res.json()["access_token"]
|
| 85 |
+
user_id = login_res.json()["user"]["id"]
|
| 86 |
+
return {"Authorization": f"Bearer {token}"}, user_id
|
| 87 |
+
|
| 88 |
+
def test_lyrics_sharing_permissions(self):
|
| 89 |
+
"""Test sharing lyrics based on perm_compartir_letras granular permission"""
|
| 90 |
+
headers_a, id_a = self._create_user_and_token("user_a", "a@example.com")
|
| 91 |
+
headers_b, id_b = self._create_user_and_token("user_b", "b@example.com")
|
| 92 |
+
|
| 93 |
+
# User A creates group
|
| 94 |
+
g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a)
|
| 95 |
+
self.assertEqual(g_res.status_code, 200)
|
| 96 |
+
group_id = g_res.json()["id"]
|
| 97 |
+
|
| 98 |
+
# Generate invitation and join B
|
| 99 |
+
inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 100 |
+
code = inv_res.json()["code"]
|
| 101 |
+
self.client.post(f"/lyrics/join/{code}", headers=headers_b)
|
| 102 |
+
|
| 103 |
+
# User B creates a lyric
|
| 104 |
+
l_res = self.client.post("/lyrics", json={
|
| 105 |
+
"title": "Lyric B", "artist": "Artist B", "content": "Content B", "group_id": None
|
| 106 |
+
}, headers=headers_b)
|
| 107 |
+
self.assertEqual(l_res.status_code, 200)
|
| 108 |
+
lyric_id = l_res.json()["id"]
|
| 109 |
+
|
| 110 |
+
# By default B can share
|
| 111 |
+
share_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "share"}, headers=headers_b)
|
| 112 |
+
self.assertEqual(share_res.status_code, 200)
|
| 113 |
+
|
| 114 |
+
# Unshare
|
| 115 |
+
unshare_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "unshare"}, headers=headers_b)
|
| 116 |
+
self.assertEqual(unshare_res.status_code, 200)
|
| 117 |
+
|
| 118 |
+
# Disable perm_compartir_letras for User B
|
| 119 |
+
requests_json = {
|
| 120 |
+
"perm_compartir_pistas": True,
|
| 121 |
+
"perm_compartir_letras": False, # Disabled
|
| 122 |
+
"perm_crud_repertorios": False,
|
| 123 |
+
"perm_crud_letras": True,
|
| 124 |
+
"perm_ver_pistas": True,
|
| 125 |
+
"perm_ver_letras": True,
|
| 126 |
+
"perm_ver_repertorios": True,
|
| 127 |
+
"perm_gestionar_solicitudes": False
|
| 128 |
+
}
|
| 129 |
+
update_perm = self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a)
|
| 130 |
+
self.assertEqual(update_perm.status_code, 200)
|
| 131 |
+
|
| 132 |
+
# User B tries to share -> 403 Forbidden
|
| 133 |
+
share_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "share"}, headers=headers_b)
|
| 134 |
+
self.assertEqual(share_res.status_code, 403)
|
| 135 |
+
|
| 136 |
+
def test_lyrics_view_permissions(self):
|
| 137 |
+
"""Test viewing lyrics in group and get_lyric based on perm_ver_letras"""
|
| 138 |
+
headers_a, id_a = self._create_user_and_token("user_a", "a@example.com")
|
| 139 |
+
headers_b, id_b = self._create_user_and_token("user_b", "b@example.com")
|
| 140 |
+
|
| 141 |
+
# User A creates group
|
| 142 |
+
g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a)
|
| 143 |
+
group_id = g_res.json()["id"]
|
| 144 |
+
|
| 145 |
+
# Join B
|
| 146 |
+
inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 147 |
+
code = inv_res.json()["code"]
|
| 148 |
+
self.client.post(f"/lyrics/join/{code}", headers=headers_b)
|
| 149 |
+
|
| 150 |
+
# User A creates and shares a lyric
|
| 151 |
+
l_res = self.client.post("/lyrics", json={
|
| 152 |
+
"title": "Lyric A", "artist": "Artist A", "content": "Content A", "group_id": group_id
|
| 153 |
+
}, headers=headers_a)
|
| 154 |
+
lyric_id = l_res.json()["id"]
|
| 155 |
+
|
| 156 |
+
# User B can view by default
|
| 157 |
+
get_res = self.client.get(f"/lyrics/{lyric_id}", headers=headers_b)
|
| 158 |
+
self.assertEqual(get_res.status_code, 200)
|
| 159 |
+
|
| 160 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/letras", headers=headers_b)
|
| 161 |
+
self.assertEqual(list_res.status_code, 200)
|
| 162 |
+
self.assertEqual(len(list_res.json()), 1)
|
| 163 |
+
|
| 164 |
+
# Disable perm_ver_letras
|
| 165 |
+
requests_json = {
|
| 166 |
+
"perm_compartir_pistas": True,
|
| 167 |
+
"perm_compartir_letras": True,
|
| 168 |
+
"perm_crud_repertorios": False,
|
| 169 |
+
"perm_crud_letras": True,
|
| 170 |
+
"perm_ver_pistas": True,
|
| 171 |
+
"perm_ver_letras": False, # Disabled
|
| 172 |
+
"perm_ver_repertorios": True,
|
| 173 |
+
"perm_gestionar_solicitudes": False
|
| 174 |
+
}
|
| 175 |
+
update_perm = self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a)
|
| 176 |
+
self.assertEqual(update_perm.status_code, 200)
|
| 177 |
+
|
| 178 |
+
# User B tries to view group lyrics -> 403
|
| 179 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/letras", headers=headers_b)
|
| 180 |
+
self.assertEqual(list_res.status_code, 403)
|
| 181 |
+
|
| 182 |
+
# User B tries to view individual lyric -> 403
|
| 183 |
+
get_res = self.client.get(f"/lyrics/{lyric_id}", headers=headers_b)
|
| 184 |
+
self.assertEqual(get_res.status_code, 403)
|
| 185 |
+
|
| 186 |
+
def test_repertoire_crud_permissions(self):
|
| 187 |
+
"""Test creating, editing, and deleting repertoires based on perm_crud_repertorios"""
|
| 188 |
+
headers_a, id_a = self._create_user_and_token("user_a", "a@example.com")
|
| 189 |
+
headers_b, id_b = self._create_user_and_token("user_b", "b@example.com")
|
| 190 |
+
|
| 191 |
+
# User A creates group
|
| 192 |
+
g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a)
|
| 193 |
+
group_id = g_res.json()["id"]
|
| 194 |
+
|
| 195 |
+
# Join B
|
| 196 |
+
inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 197 |
+
code = inv_res.json()["code"]
|
| 198 |
+
self.client.post(f"/lyrics/join/{code}", headers=headers_b)
|
| 199 |
+
|
| 200 |
+
# User A creates a repertoire
|
| 201 |
+
rep_res = self.client.post("/lyrics/repertorios", json={
|
| 202 |
+
"name": "Rep Admin", "group_id": group_id, "lyrics_ids": []
|
| 203 |
+
}, headers=headers_a)
|
| 204 |
+
rep_id = rep_res.json()["id"]
|
| 205 |
+
|
| 206 |
+
# User B has perm_crud_repertorios = False by default -> should get 403 on editing
|
| 207 |
+
edit_res = self.client.put(f"/lyrics/repertorios/{rep_id}", json={
|
| 208 |
+
"name": "Rep Mod", "group_id": group_id, "lyrics_ids": []
|
| 209 |
+
}, headers=headers_b)
|
| 210 |
+
self.assertEqual(edit_res.status_code, 403)
|
| 211 |
+
|
| 212 |
+
# Enable perm_crud_repertorios
|
| 213 |
+
requests_json = {
|
| 214 |
+
"perm_compartir_pistas": True,
|
| 215 |
+
"perm_compartir_letras": True,
|
| 216 |
+
"perm_crud_repertorios": True, # Enabled!
|
| 217 |
+
"perm_crud_letras": True,
|
| 218 |
+
"perm_ver_pistas": True,
|
| 219 |
+
"perm_ver_letras": True,
|
| 220 |
+
"perm_ver_repertorios": True,
|
| 221 |
+
"perm_gestionar_solicitudes": False
|
| 222 |
+
}
|
| 223 |
+
self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a)
|
| 224 |
+
|
| 225 |
+
# User B edits -> 200 OK
|
| 226 |
+
edit_res = self.client.put(f"/lyrics/repertorios/{rep_id}", json={
|
| 227 |
+
"name": "Rep Mod", "group_id": group_id, "lyrics_ids": []
|
| 228 |
+
}, headers=headers_b)
|
| 229 |
+
self.assertEqual(edit_res.status_code, 200)
|
| 230 |
+
|
| 231 |
+
def test_repertoire_view_permissions(self):
|
| 232 |
+
"""Test viewing repertoires based on perm_ver_repertorios"""
|
| 233 |
+
headers_a, id_a = self._create_user_and_token("user_a", "a@example.com")
|
| 234 |
+
headers_b, id_b = self._create_user_and_token("user_b", "b@example.com")
|
| 235 |
+
|
| 236 |
+
# User A creates group
|
| 237 |
+
g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a)
|
| 238 |
+
group_id = g_res.json()["id"]
|
| 239 |
+
|
| 240 |
+
# Join B
|
| 241 |
+
inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 242 |
+
code = inv_res.json()["code"]
|
| 243 |
+
self.client.post(f"/lyrics/join/{code}", headers=headers_b)
|
| 244 |
+
|
| 245 |
+
# User A creates a repertoire
|
| 246 |
+
rep_res = self.client.post("/lyrics/repertorios", json={
|
| 247 |
+
"name": "Rep Admin", "group_id": group_id, "lyrics_ids": []
|
| 248 |
+
}, headers=headers_a)
|
| 249 |
+
rep_id = rep_res.json()["id"]
|
| 250 |
+
|
| 251 |
+
# User B can view by default
|
| 252 |
+
get_res = self.client.get(f"/lyrics/repertorios/{rep_id}", headers=headers_b)
|
| 253 |
+
self.assertEqual(get_res.status_code, 200)
|
| 254 |
+
|
| 255 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/repertorios", headers=headers_b)
|
| 256 |
+
self.assertEqual(list_res.status_code, 200)
|
| 257 |
+
|
| 258 |
+
# Disable perm_ver_repertorios
|
| 259 |
+
requests_json = {
|
| 260 |
+
"perm_compartir_pistas": True,
|
| 261 |
+
"perm_compartir_letras": True,
|
| 262 |
+
"perm_crud_repertorios": True,
|
| 263 |
+
"perm_crud_letras": True,
|
| 264 |
+
"perm_ver_pistas": True,
|
| 265 |
+
"perm_ver_letras": True,
|
| 266 |
+
"perm_ver_repertorios": False, # Disabled
|
| 267 |
+
"perm_gestionar_solicitudes": False
|
| 268 |
+
}
|
| 269 |
+
self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a)
|
| 270 |
+
|
| 271 |
+
# User B tries to view group repertoires -> 403
|
| 272 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/repertorios", headers=headers_b)
|
| 273 |
+
self.assertEqual(list_res.status_code, 403)
|
| 274 |
+
|
| 275 |
+
# User B tries to view individual repertoire -> 403
|
| 276 |
+
get_res = self.client.get(f"/lyrics/repertorios/{rep_id}", headers=headers_b)
|
| 277 |
+
self.assertEqual(get_res.status_code, 403)
|
| 278 |
+
|
| 279 |
+
def test_songs_view_permissions(self):
|
| 280 |
+
"""Test viewing tracks/songs and get_stems/harmony_status based on perm_ver_pistas"""
|
| 281 |
+
headers_a, id_a = self._create_user_and_token("user_a", "a@example.com")
|
| 282 |
+
headers_b, id_b = self._create_user_and_token("user_b", "b@example.com")
|
| 283 |
+
|
| 284 |
+
# User A creates group
|
| 285 |
+
g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a)
|
| 286 |
+
group_id = g_res.json()["id"]
|
| 287 |
+
|
| 288 |
+
# Join B
|
| 289 |
+
inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 290 |
+
code = inv_res.json()["code"]
|
| 291 |
+
self.client.post(f"/lyrics/join/{code}", headers=headers_b)
|
| 292 |
+
|
| 293 |
+
# Seed a Task in DB directly
|
| 294 |
+
db = TestingSessionLocal()
|
| 295 |
+
task = Task(
|
| 296 |
+
task_id="task_test_123",
|
| 297 |
+
user_id=id_a,
|
| 298 |
+
song_name="Song A",
|
| 299 |
+
status="success",
|
| 300 |
+
telegram_file_ids='{"vocals": "stems/task_test_123/vocals.flac"}'
|
| 301 |
+
)
|
| 302 |
+
db.add(task)
|
| 303 |
+
db.commit()
|
| 304 |
+
|
| 305 |
+
# Share it
|
| 306 |
+
db_group = db.query(Group).filter(Group.id == group_id).first()
|
| 307 |
+
task.groups_shared.append(db_group)
|
| 308 |
+
db.commit()
|
| 309 |
+
db.close()
|
| 310 |
+
|
| 311 |
+
# User B can view by default
|
| 312 |
+
get_res = self.client.get("/songs/tasks/task_test_123", headers=headers_b)
|
| 313 |
+
self.assertEqual(get_res.status_code, 200)
|
| 314 |
+
|
| 315 |
+
stems_res = self.client.get("/songs/stems/task_test_123", headers=headers_b)
|
| 316 |
+
self.assertEqual(stems_res.status_code, 200)
|
| 317 |
+
|
| 318 |
+
harm_res = self.client.get("/songs/harmony-status/task_test_123", headers=headers_b)
|
| 319 |
+
self.assertEqual(harm_res.status_code, 200)
|
| 320 |
+
|
| 321 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/pistas", headers=headers_b)
|
| 322 |
+
self.assertEqual(list_res.status_code, 200)
|
| 323 |
+
|
| 324 |
+
# Disable perm_ver_pistas
|
| 325 |
+
requests_json = {
|
| 326 |
+
"perm_compartir_pistas": True,
|
| 327 |
+
"perm_compartir_letras": True,
|
| 328 |
+
"perm_crud_repertorios": True,
|
| 329 |
+
"perm_crud_letras": True,
|
| 330 |
+
"perm_ver_pistas": False, # Disabled
|
| 331 |
+
"perm_ver_letras": True,
|
| 332 |
+
"perm_ver_repertorios": True,
|
| 333 |
+
"perm_gestionar_solicitudes": False
|
| 334 |
+
}
|
| 335 |
+
self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a)
|
| 336 |
+
|
| 337 |
+
# User B tries to view group tracks -> 403
|
| 338 |
+
list_res = self.client.get(f"/alabanza/grupos/{group_id}/pistas", headers=headers_b)
|
| 339 |
+
self.assertEqual(list_res.status_code, 403)
|
| 340 |
+
|
| 341 |
+
# User B tries to view individual track -> 403
|
| 342 |
+
get_res = self.client.get("/songs/tasks/task_test_123", headers=headers_b)
|
| 343 |
+
self.assertEqual(get_res.status_code, 403)
|
| 344 |
+
|
| 345 |
+
# User B tries to view stems -> 403
|
| 346 |
+
stems_res = self.client.get("/songs/stems/task_test_123", headers=headers_b)
|
| 347 |
+
self.assertEqual(stems_res.status_code, 403)
|
| 348 |
+
|
| 349 |
+
# User B tries to check harmony status -> 403
|
| 350 |
+
harm_res = self.client.get("/songs/harmony-status/task_test_123", headers=headers_b)
|
| 351 |
+
self.assertEqual(harm_res.status_code, 403)
|
| 352 |
+
|
| 353 |
+
|
| 354 |
+
if __name__ == "__main__":
|
| 355 |
+
unittest.main()
|
tests/verify_group_permissions.py
ADDED
|
@@ -0,0 +1,205 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
import requests
|
| 2 |
+
import uuid
|
| 3 |
+
|
| 4 |
+
BASE_URL = "http://127.0.0.1:8000"
|
| 5 |
+
|
| 6 |
+
def run_tests():
|
| 7 |
+
print("=" * 70)
|
| 8 |
+
print("INICIANDO PRUEBAS DE VERIFICACIÓN DE PERMISOS GRANULARES DE GRUPO")
|
| 9 |
+
print("=" * 70)
|
| 10 |
+
|
| 11 |
+
# 1. Registrar y autenticar usuarios de prueba
|
| 12 |
+
suffix = str(uuid.uuid4())[:8]
|
| 13 |
+
user_a = f"admin_perm_{suffix}"
|
| 14 |
+
user_b = f"member_perm_{suffix}"
|
| 15 |
+
password = "password123"
|
| 16 |
+
|
| 17 |
+
print("\n[1] Registrando y logueando usuarios...")
|
| 18 |
+
for u in [user_a, user_b]:
|
| 19 |
+
requests.post(f"{BASE_URL}/auth/register", json={
|
| 20 |
+
"username": u,
|
| 21 |
+
"email": f"{u}@example.com",
|
| 22 |
+
"password": password
|
| 23 |
+
})
|
| 24 |
+
|
| 25 |
+
# Login User A (Admin del grupo)
|
| 26 |
+
res_a = requests.post(f"{BASE_URL}/auth/token", data={
|
| 27 |
+
"username": f"{user_a}@example.com",
|
| 28 |
+
"password": password,
|
| 29 |
+
"device_id": f"dev_a_{suffix}",
|
| 30 |
+
"device_name": "Device A"
|
| 31 |
+
})
|
| 32 |
+
token_a = res_a.json()["access_token"]
|
| 33 |
+
headers_a = {"Authorization": f"Bearer {token_a}"}
|
| 34 |
+
|
| 35 |
+
# Login User B (Miembro del grupo)
|
| 36 |
+
res_b = requests.post(f"{BASE_URL}/auth/token", data={
|
| 37 |
+
"username": f"{user_b}@example.com",
|
| 38 |
+
"password": password,
|
| 39 |
+
"device_id": f"dev_b_{suffix}",
|
| 40 |
+
"device_name": "Device B"
|
| 41 |
+
})
|
| 42 |
+
token_b = res_b.json()["access_token"]
|
| 43 |
+
headers_b = {"Authorization": f"Bearer {token_b}"}
|
| 44 |
+
member_user_id = res_b.json()["user"]["id"]
|
| 45 |
+
|
| 46 |
+
print(" [OK] Usuarios listos.")
|
| 47 |
+
|
| 48 |
+
# 2. User A crea un grupo
|
| 49 |
+
print("\n[2] User A creando grupo...")
|
| 50 |
+
group_res = requests.post(f"{BASE_URL}/lyrics/groups", json={"name": f"Grupo Perm {suffix}"}, headers=headers_a)
|
| 51 |
+
group_id = group_res.json()["id"]
|
| 52 |
+
print(f" [OK] Grupo creado con ID {group_id}")
|
| 53 |
+
|
| 54 |
+
# 3. User B se une al grupo
|
| 55 |
+
print("\n[3] User B uniéndose...")
|
| 56 |
+
invite_res = requests.post(f"{BASE_URL}/lyrics/groups/{group_id}/invite", headers=headers_a)
|
| 57 |
+
code = invite_res.json()["code"]
|
| 58 |
+
requests.post(f"{BASE_URL}/lyrics/join/{code}", headers=headers_b)
|
| 59 |
+
print(" [OK] User B unido al grupo.")
|
| 60 |
+
|
| 61 |
+
# 4. User A (Admin) crea una letra y una pista (task) de prueba y las comparte con el grupo
|
| 62 |
+
# Primero creamos letra
|
| 63 |
+
lyric_res = requests.post(f"{BASE_URL}/lyrics", json={
|
| 64 |
+
"title": "Letra de Admin",
|
| 65 |
+
"artist": "Artista Admin",
|
| 66 |
+
"content": "Contenido de Admin",
|
| 67 |
+
"group_id": group_id
|
| 68 |
+
}, headers=headers_a)
|
| 69 |
+
lyric_id = lyric_res.json()["id"]
|
| 70 |
+
|
| 71 |
+
# Simular una pista en DB (para propósitos del test no necesitamos procesarla, pero necesitamos el registro)
|
| 72 |
+
# Como crear una pista requiere subir archivo o youtube, subiremos un pequeño archivo de prueba o compartiremos uno existente.
|
| 73 |
+
# En Melodix, el endpoint para crear tarea es `/songs` o `/songs/upload-youtube`.
|
| 74 |
+
# Vamos a subir una pista de prueba simulada si es posible, o registrarla.
|
| 75 |
+
# Para no complicar con subida de archivos binarios reales, podemos usar /songs/upload-youtube con un enlace dummy de youtube
|
| 76 |
+
# Pero eso consume créditos y podría tardar/fallar si no hay internet o celery no corre.
|
| 77 |
+
# Alternativamente, podemos probar las letras y repertorios primero.
|
| 78 |
+
# Probemos letras y repertorios.
|
| 79 |
+
|
| 80 |
+
# 5. User A crea un repertorio
|
| 81 |
+
print("\n[4] User A creando repertorio grupal...")
|
| 82 |
+
rep_res = requests.post(f"{BASE_URL}/lyrics/repertorios", json={
|
| 83 |
+
"name": "Repertorio de Prueba",
|
| 84 |
+
"group_id": group_id,
|
| 85 |
+
"lyrics_ids": [lyric_id]
|
| 86 |
+
}, headers=headers_a)
|
| 87 |
+
rep_id = rep_res.json()["id"]
|
| 88 |
+
print(f" [OK] Repertorio creado con ID {rep_id}")
|
| 89 |
+
|
| 90 |
+
# 6. Probemos permisos por defecto (miembros tienen permisos de lectura habilitados por defecto, pero no CRUD repertorio)
|
| 91 |
+
print("\n[5] Probando permisos por defecto de User B...")
|
| 92 |
+
# Ver letras del grupo
|
| 93 |
+
res = requests.get(f"{BASE_URL}/alabanza/grupos/{group_id}/letras", headers=headers_b)
|
| 94 |
+
assert res.status_code == 200, f"Error obteniendo letras del grupo: {res.text}"
|
| 95 |
+
|
| 96 |
+
# Ver letra individual
|
| 97 |
+
res = requests.get(f"{BASE_URL}/lyrics/{lyric_id}", headers=headers_b)
|
| 98 |
+
assert res.status_code == 200, f"Error obteniendo letra: {res.text}"
|
| 99 |
+
|
| 100 |
+
# Ver repertorios del grupo
|
| 101 |
+
res = requests.get(f"{BASE_URL}/alabanza/grupos/{group_id}/repertorios", headers=headers_b)
|
| 102 |
+
assert res.status_code == 200, f"Error obteniendo repertorios del grupo: {res.text}"
|
| 103 |
+
|
| 104 |
+
# Ver repertorio individual
|
| 105 |
+
res = requests.get(f"{BASE_URL}/lyrics/repertorios/{rep_id}", headers=headers_b)
|
| 106 |
+
assert res.status_code == 200, f"Error obteniendo repertorio: {res.text}"
|
| 107 |
+
|
| 108 |
+
# Intentar modificar repertorio (esperado 403 por defecto ya que perm_crud_repertorios es False)
|
| 109 |
+
res = requests.put(f"{BASE_URL}/lyrics/repertorios/{rep_id}", json={
|
| 110 |
+
"name": "Repertorio Modificado por B",
|
| 111 |
+
"group_id": group_id,
|
| 112 |
+
"lyrics_ids": [lyric_id]
|
| 113 |
+
}, headers=headers_b)
|
| 114 |
+
assert res.status_code == 403, f"Se esperaba 403 al editar repertorio sin permiso, se obtuvo {res.status_code}"
|
| 115 |
+
print(" [OK] Restricción de CRUD repertorio por defecto funciona (403).")
|
| 116 |
+
|
| 117 |
+
# 7. Cambiar permisos: Habilitar perm_crud_repertorios, deshabilitar perm_ver_letras y perm_ver_repertorios
|
| 118 |
+
print("\n[6] User A actualiza permisos de User B...")
|
| 119 |
+
perm_res = requests.put(f"{BASE_URL}/alabanza/grupos/{group_id}/miembros/{member_user_id}/permisos", json={
|
| 120 |
+
"perm_compartir_pistas": True,
|
| 121 |
+
"perm_compartir_letras": False, # Deshabilitado
|
| 122 |
+
"perm_crud_repertorios": True, # Habilitado
|
| 123 |
+
"perm_crud_letras": True,
|
| 124 |
+
"perm_ver_pistas": True,
|
| 125 |
+
"perm_ver_letras": False, # Deshabilitado
|
| 126 |
+
"perm_ver_repertorios": False, # Deshabilitado
|
| 127 |
+
"perm_gestionar_solicitudes": False
|
| 128 |
+
}, headers=headers_a)
|
| 129 |
+
assert perm_res.status_code == 200, f"Error actualizando permisos: {perm_res.text}"
|
| 130 |
+
print(" [OK] Permisos actualizados en el servidor.")
|
| 131 |
+
|
| 132 |
+
# 8. Validar que deshabilitar perm_ver_letras y perm_ver_repertorios funcione
|
| 133 |
+
print("\n[7] Probando permisos de visualización deshabilitados para User B...")
|
| 134 |
+
# Obtener letras del grupo (esperado 403)
|
| 135 |
+
res = requests.get(f"{BASE_URL}/alabanza/grupos/{group_id}/letras", headers=headers_b)
|
| 136 |
+
assert res.status_code == 403, f"Se esperaba 403 al listar letras, se obtuvo {res.status_code}"
|
| 137 |
+
|
| 138 |
+
# Obtener letra individual (esperado 403)
|
| 139 |
+
res = requests.get(f"{BASE_URL}/lyrics/{lyric_id}", headers=headers_b)
|
| 140 |
+
assert res.status_code == 403, f"Se esperaba 403 al obtener letra, se obtuvo {res.status_code}"
|
| 141 |
+
|
| 142 |
+
# Obtener repertorios del grupo (esperado 403)
|
| 143 |
+
res = requests.get(f"{BASE_URL}/alabanza/grupos/{group_id}/repertorios", headers=headers_b)
|
| 144 |
+
assert res.status_code == 403, f"Se esperaba 403 al listar repertorios, se obtuvo {res.status_code}"
|
| 145 |
+
|
| 146 |
+
# Obtener repertorio individual (esperado 403)
|
| 147 |
+
res = requests.get(f"{BASE_URL}/lyrics/repertorios/{rep_id}", headers=headers_b)
|
| 148 |
+
assert res.status_code == 403, f"Se esperaba 403 al obtener repertorio, se obtuvo {res.status_code}"
|
| 149 |
+
print(" [OK] Restricciones de visualización funcionan (403).")
|
| 150 |
+
|
| 151 |
+
# 9. Validar que perm_crud_repertorios habilitado funcione
|
| 152 |
+
print("\n[8] Probando perm_crud_repertorios habilitado para User B...")
|
| 153 |
+
res = requests.put(f"{BASE_URL}/lyrics/repertorios/{rep_id}", json={
|
| 154 |
+
"name": "Repertorio Modificado por B con Permiso",
|
| 155 |
+
"group_id": group_id,
|
| 156 |
+
"lyrics_ids": [lyric_id]
|
| 157 |
+
}, headers=headers_b)
|
| 158 |
+
assert res.status_code == 200, f"Error al editar repertorio teniendo permiso: {res.text}"
|
| 159 |
+
print(" [OK] Edición de repertorio por miembro autorizado exitosa.")
|
| 160 |
+
|
| 161 |
+
# 10. Validar compartir letra con perm_compartir_letras deshabilitado
|
| 162 |
+
# Primero creamos una letra propia de User B (que no está compartida con el grupo aún)
|
| 163 |
+
lyric_b_res = requests.post(f"{BASE_URL}/lyrics", json={
|
| 164 |
+
"title": "Letra de B",
|
| 165 |
+
"artist": "Artista B",
|
| 166 |
+
"content": "Contenido de B",
|
| 167 |
+
"group_id": None
|
| 168 |
+
}, headers=headers_b)
|
| 169 |
+
lyric_b_id = lyric_b_res.json()["id"]
|
| 170 |
+
|
| 171 |
+
print("\n[9] Probando compartir letra con perm_compartir_letras deshabilitado...")
|
| 172 |
+
res = requests.post(f"{BASE_URL}/lyrics/{lyric_b_id}/share", json={
|
| 173 |
+
"group_id": group_id,
|
| 174 |
+
"action": "share"
|
| 175 |
+
}, headers=headers_b)
|
| 176 |
+
assert res.status_code == 403, f"Se esperaba 403 al compartir letra sin permiso, se obtuvo {res.status_code}"
|
| 177 |
+
print(" [OK] Compartir letra bloqueado correctamente (403).")
|
| 178 |
+
|
| 179 |
+
# Habilitar perm_compartir_letras y volver a intentar
|
| 180 |
+
print("\n[10] User A habilita perm_compartir_letras a User B...")
|
| 181 |
+
perm_res = requests.put(f"{BASE_URL}/alabanza/grupos/{group_id}/miembros/{member_user_id}/permisos", json={
|
| 182 |
+
"perm_compartir_pistas": True,
|
| 183 |
+
"perm_compartir_letras": True, # Habilitado ahora
|
| 184 |
+
"perm_crud_repertorios": True,
|
| 185 |
+
"perm_crud_letras": True,
|
| 186 |
+
"perm_ver_pistas": True,
|
| 187 |
+
"perm_ver_letras": True,
|
| 188 |
+
"perm_ver_repertorios": True,
|
| 189 |
+
"perm_gestionar_solicitudes": False
|
| 190 |
+
}, headers=headers_a)
|
| 191 |
+
|
| 192 |
+
print("Intentando compartir letra con permiso habilitado...")
|
| 193 |
+
res = requests.post(f"{BASE_URL}/lyrics/{lyric_b_id}/share", json={
|
| 194 |
+
"group_id": group_id,
|
| 195 |
+
"action": "share"
|
| 196 |
+
}, headers=headers_b)
|
| 197 |
+
assert res.status_code == 200, f"Error al compartir letra teniendo permiso: {res.text}"
|
| 198 |
+
print(" [OK] Compartir letra por miembro autorizado exitosa.")
|
| 199 |
+
|
| 200 |
+
print("\n" + "=" * 70)
|
| 201 |
+
print(" ¡TODAS LAS PRUEBAS DE PERMISOS DE GRUPO PASARON CON EXITO!")
|
| 202 |
+
print("=" * 70 + "\n")
|
| 203 |
+
|
| 204 |
+
if __name__ == "__main__":
|
| 205 |
+
run_tests()
|