feat: Add pending linkages management API and user interface for tag approvals
This commit introduces a new API for managing pending tag linkages, allowing admins to review and approve or deny user-suggested tags. The frontend has been updated with a new User Tagged Photos page for displaying pending linkages, including options for filtering and submitting decisions. Additionally, the Layout component has been modified to include navigation to the new page. Documentation has been updated to reflect these changes.
This commit is contained in:
tanyar09
@@ -0,0 +1,264 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Generator
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from src.web.app import app
|
||||
from src.web.db import models
|
||||
from src.web.db.models import Photo, PhotoTagLinkage, Tag, User
|
||||
from src.web.db.session import get_auth_db, get_db
|
||||
from src.web.constants.roles import DEFAULT_ADMIN_ROLE
|
||||
from src.web.api.auth import get_current_user
|
||||
|
||||
|
||||
# Create isolated in-memory databases for main and auth stores.
|
||||
main_engine = create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
auth_engine = create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
MainSessionLocal = sessionmaker(
|
||||
bind=main_engine, autoflush=False, autocommit=False, future=True
|
||||
)
|
||||
AuthSessionLocal = sessionmaker(
|
||||
bind=auth_engine, autoflush=False, autocommit=False, future=True
|
||||
)
|
||||
|
||||
models.Base.metadata.create_all(bind=main_engine)
|
||||
|
||||
with auth_engine.begin() as connection:
|
||||
connection.execute(
|
||||
text(
|
||||
"""
|
||||
CREATE TABLE users (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT,
|
||||
email TEXT
|
||||
)
|
||||
"""
|
||||
)
|
||||
)
|
||||
connection.execute(
|
||||
text(
|
||||
"""
|
||||
CREATE TABLE pending_linkages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
photo_id INTEGER NOT NULL,
|
||||
tag_id INTEGER,
|
||||
tag_name VARCHAR(255),
|
||||
user_id INTEGER NOT NULL,
|
||||
status VARCHAR(50) DEFAULT 'pending',
|
||||
notes TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
"""
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def override_get_db() -> Generator[Session, None, None]:
|
||||
db = MainSessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def override_get_auth_db() -> Generator[Session, None, None]:
|
||||
db = AuthSessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def override_get_current_user() -> dict[str, str]:
|
||||
return {"username": "admin"}
|
||||
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
app.dependency_overrides[get_auth_db] = override_get_auth_db
|
||||
app.dependency_overrides[get_current_user] = override_get_current_user
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def _ensure_admin_user() -> None:
|
||||
with MainSessionLocal() as session:
|
||||
existing = session.query(User).filter(User.username == "admin").first()
|
||||
if existing:
|
||||
existing.is_admin = True
|
||||
existing.role = DEFAULT_ADMIN_ROLE
|
||||
session.commit()
|
||||
return
|
||||
|
||||
admin_user = User(
|
||||
username="admin",
|
||||
password_hash="test",
|
||||
email="admin@example.com",
|
||||
full_name="Admin",
|
||||
is_active=True,
|
||||
is_admin=True,
|
||||
role=DEFAULT_ADMIN_ROLE,
|
||||
)
|
||||
session.add(admin_user)
|
||||
session.commit()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean_databases() -> Generator[None, None, None]:
|
||||
with MainSessionLocal() as session:
|
||||
session.query(PhotoTagLinkage).delete()
|
||||
session.query(Tag).delete()
|
||||
session.query(Photo).delete()
|
||||
session.query(User).filter(User.username != "admin").delete()
|
||||
session.commit()
|
||||
|
||||
with AuthSessionLocal() as session:
|
||||
session.execute(text("DELETE FROM pending_linkages"))
|
||||
session.execute(text("DELETE FROM users"))
|
||||
session.commit()
|
||||
|
||||
_ensure_admin_user()
|
||||
yield
|
||||
|
||||
|
||||
def _insert_auth_user(user_id: int = 1) -> None:
|
||||
with auth_engine.begin() as connection:
|
||||
connection.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO users (id, name, email)
|
||||
VALUES (:id, :name, :email)
|
||||
"""
|
||||
),
|
||||
{"id": user_id, "name": "Tester", "email": "tester@example.com"},
|
||||
)
|
||||
|
||||
|
||||
def _insert_pending_linkage(
|
||||
photo_id: int,
|
||||
*,
|
||||
tag_id: int | None = None,
|
||||
tag_name: str | None = None,
|
||||
status: str = "pending",
|
||||
user_id: int = 1,
|
||||
) -> int:
|
||||
with auth_engine.begin() as connection:
|
||||
result = connection.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO pending_linkages (
|
||||
photo_id, tag_id, tag_name, user_id, status, notes
|
||||
)
|
||||
VALUES (:photo_id, :tag_id, :tag_name, :user_id, :status, 'note')
|
||||
"""
|
||||
),
|
||||
{
|
||||
"photo_id": photo_id,
|
||||
"tag_id": tag_id,
|
||||
"tag_name": tag_name,
|
||||
"user_id": user_id,
|
||||
"status": status,
|
||||
},
|
||||
)
|
||||
return int(result.lastrowid)
|
||||
|
||||
|
||||
def _create_photo(path: str, filename: str, file_hash: str) -> int:
|
||||
with MainSessionLocal() as session:
|
||||
photo = Photo(path=path, filename=filename, file_hash=file_hash)
|
||||
session.add(photo)
|
||||
session.commit()
|
||||
session.refresh(photo)
|
||||
return photo.id
|
||||
|
||||
|
||||
def test_list_pending_linkages_returns_existing_rows():
|
||||
_ensure_admin_user()
|
||||
photo_id = _create_photo("/tmp/photo1.jpg", "photo1.jpg", "hash1")
|
||||
_insert_auth_user()
|
||||
linkage_id = _insert_pending_linkage(photo_id, tag_name="Beach Day")
|
||||
|
||||
response = client.get("/api/v1/pending-linkages")
|
||||
assert response.status_code == 200
|
||||
|
||||
payload = response.json()
|
||||
assert payload["total"] == 1
|
||||
item = payload["items"][0]
|
||||
assert item["photo_id"] == photo_id
|
||||
assert item["proposed_tag_name"] == "Beach Day"
|
||||
assert item["status"] == "pending"
|
||||
|
||||
|
||||
def test_review_pending_linkages_creates_tag_and_linkage():
|
||||
_ensure_admin_user()
|
||||
photo_id = _create_photo("/tmp/photo2.jpg", "photo2.jpg", "hash2")
|
||||
_insert_auth_user()
|
||||
linkage_id = _insert_pending_linkage(photo_id, tag_name="Sunset Crew")
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/pending-linkages/review",
|
||||
json={"decisions": [{"id": linkage_id, "decision": "approve"}]},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
payload = response.json()
|
||||
assert payload["approved"] == 1
|
||||
assert payload["denied"] == 0
|
||||
assert payload["tags_created"] == 1
|
||||
assert payload["linkages_created"] == 1
|
||||
|
||||
with MainSessionLocal() as session:
|
||||
tags = session.query(Tag).all()
|
||||
assert len(tags) == 1
|
||||
assert tags[0].tag_name == "Sunset Crew"
|
||||
linkage = session.query(PhotoTagLinkage).first()
|
||||
assert linkage is not None
|
||||
assert linkage.photo_id == photo_id
|
||||
assert linkage.tag_id == tags[0].id
|
||||
|
||||
with AuthSessionLocal() as session:
|
||||
statuses = session.execute(
|
||||
text("SELECT status FROM pending_linkages WHERE id = :id"),
|
||||
{"id": linkage_id},
|
||||
).fetchone()
|
||||
assert statuses is not None
|
||||
assert statuses[0] == "approved"
|
||||
|
||||
|
||||
def test_cleanup_pending_linkages_deletes_approved_and_denied():
|
||||
_ensure_admin_user()
|
||||
photo_id = _create_photo("/tmp/photo3.jpg", "photo3.jpg", "hash3")
|
||||
_insert_auth_user()
|
||||
|
||||
approved_id = _insert_pending_linkage(photo_id, tag_name="Approved Tag", status="approved")
|
||||
denied_id = _insert_pending_linkage(photo_id, tag_name="Denied Tag", status="denied")
|
||||
pending_id = _insert_pending_linkage(photo_id, tag_name="Pending Tag", status="pending")
|
||||
|
||||
response = client.post("/api/v1/pending-linkages/cleanup")
|
||||
assert response.status_code == 200
|
||||
|
||||
payload = response.json()
|
||||
assert payload["deleted_records"] == 2
|
||||
|
||||
with AuthSessionLocal() as session:
|
||||
remaining = session.execute(
|
||||
text("SELECT id, status FROM pending_linkages ORDER BY id")
|
||||
).fetchall()
|
||||
assert len(remaining) == 1
|
||||
assert remaining[0][0] == pending_id
|
||||
assert remaining[0][1] == "pending"
|
||||
|
||||
Reference in New Issue
Block a user