346 lines
12 KiB
Python
346 lines
12 KiB
Python
import re
|
|
from datetime import date
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base, get_db
|
|
from app.main import app
|
|
from app.models import User
|
|
from app.security import hash_password
|
|
|
|
CSRF_RE = re.compile(r'name="csrf_token" value="([^"]+)"')
|
|
|
|
|
|
@pytest.fixture()
|
|
def db():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, expire_on_commit=False)()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(db):
|
|
def override_get_db():
|
|
yield db
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
app.dependency_overrides.pop(get_db, None)
|
|
|
|
|
|
@pytest.fixture()
|
|
def admin(db):
|
|
user = User(
|
|
name="Admin",
|
|
username="admin",
|
|
password_hash=hash_password("admin-password"),
|
|
is_admin=True,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
def csrf(client: TestClient, path: str) -> str:
|
|
match = CSRF_RE.search(client.get(path).text)
|
|
assert match, f"no csrf token on {path}"
|
|
return match.group(1)
|
|
|
|
|
|
def login(client: TestClient, username: str, password: str) -> str:
|
|
token = csrf(client, "/login")
|
|
return client.post(
|
|
"/login",
|
|
data={"username": username, "password": password, "csrf_token": token},
|
|
follow_redirects=True,
|
|
).text
|
|
|
|
|
|
def add_person(client: TestClient, **fields) -> str:
|
|
data = {"name": "Person", "csrf_token": csrf(client, "/admin/users")}
|
|
data.update(fields)
|
|
return client.post("/admin/users", data=data, follow_redirects=True).text
|
|
|
|
|
|
def update_person(client: TestClient, user_id: int, **fields) -> str:
|
|
data = {"name": "Person", "is_active": "on", "notify_email": "on", "csrf_token": csrf(client, "/admin/users")}
|
|
data.update(fields)
|
|
return client.post(f"/admin/users/{user_id}/update", data=data, follow_redirects=True).text
|
|
|
|
|
|
def usernames(db) -> list[str | None]:
|
|
return list(db.scalars(select(User.username).order_by(User.name)).all())
|
|
|
|
|
|
def test_person_can_be_created_without_login(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
page = add_person(client, name="Rota Only")
|
|
assert "Rota Only" in page
|
|
assert "sem login" in page
|
|
assert usernames(db) == ["admin", None]
|
|
assert db.scalar(select(User).where(User.name == "Rota Only")).can_login is False
|
|
|
|
|
|
@pytest.mark.parametrize("fields", [{"username": "newcomer"}, {"password": "a-secure-password"}])
|
|
def test_username_and_password_must_be_set_together(client, db, admin, fields):
|
|
login(client, "admin", "admin-password")
|
|
page = add_person(client, name="Halfway", **fields)
|
|
assert "Defina o nome de utilizador e a palavra-passe" in page
|
|
assert usernames(db) == ["admin"]
|
|
assert db.scalar(select(User).where(User.name == "Halfway")) is None
|
|
|
|
|
|
def test_person_without_credentials_cannot_sign_in(client, db, admin):
|
|
db.add(
|
|
User(
|
|
name="Rota Only",
|
|
username=None,
|
|
password_hash=hash_password("a-secure-password"),
|
|
is_active=True,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
assert "Nome de utilizador ou palavra-passe inválidos" in login(client, " ", "a-secure-password")
|
|
assert client.get("/swaps", follow_redirects=False).status_code == 401
|
|
|
|
|
|
def test_username_without_password_cannot_sign_in(client, db, admin):
|
|
db.add(User(name="Rota Only", username="rotarota", password_hash=None, is_active=True))
|
|
db.commit()
|
|
assert "Nome de utilizador ou palavra-passe inválidos" in login(client, "rotarota", "a-secure-password")
|
|
|
|
|
|
def test_admin_can_give_and_remove_a_login(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
add_person(client, name="Rota Only")
|
|
person = db.scalar(select(User).where(User.name == "Rota Only"))
|
|
|
|
page = update_person(client, person.id, name="Rota Only", username="rotarota")
|
|
assert "Defina uma palavra-passe para esta pessoa poder iniciar sessão." in page
|
|
assert db.get(User, person.id).can_login is False
|
|
|
|
update_person(
|
|
client,
|
|
person.id,
|
|
name="Rota Only",
|
|
username="rotarota",
|
|
new_password="a-secure-password",
|
|
)
|
|
assert db.get(User, person.id).can_login is True
|
|
|
|
client.post("/logout", data={"csrf_token": csrf(client, "/")})
|
|
assert "Bem-vindo(a), Rota Only." in login(client, "rotarota", "a-secure-password")
|
|
|
|
login(client, "admin", "admin-password")
|
|
page = update_person(client, person.id, name="Rota Only")
|
|
assert "Pessoa atualizada." in page
|
|
assert db.get(User, person.id).can_login is False
|
|
|
|
|
|
def test_username_must_stay_unique(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
add_person(client, name="Rota Only")
|
|
person = db.scalar(select(User).where(User.name == "Rota Only"))
|
|
|
|
page = update_person(
|
|
client, person.id, name="Rota Only", username="admin", new_password="a-secure-password"
|
|
)
|
|
assert "Esse nome de utilizador já está a ser usado." in page
|
|
assert db.get(User, person.id).can_login is False
|
|
|
|
|
|
def test_admin_cannot_remove_their_own_login(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
page = update_person(client, admin.id, name="Admin", new_password="another-password")
|
|
assert "Defina um nome de utilizador antes de definir a palavra-passe da sua conta." in page
|
|
assert db.get(User, admin.id).can_login is True
|
|
|
|
|
|
def test_signed_in_home_is_the_rota_and_nav_has_no_rota_link(client, db, admin):
|
|
landing = login(client, "admin", "admin-password")
|
|
assert '<a class="brand" href="/"' in landing
|
|
|
|
nav = client.get("/admin/users").text
|
|
assert '<a href="/">Rota</a>' not in nav
|
|
assert '<a href="/schedules">Escalas</a>' in nav
|
|
assert '<a href="/swaps">Trocas</a>' in nav
|
|
assert 'href="/"' in nav # the brand stays the home link for signed-in members
|
|
|
|
|
|
def delete_person(client: TestClient, user_id: int) -> str:
|
|
return client.get(f"/admin/users/{user_id}/delete").text
|
|
|
|
|
|
def test_removing_a_person_clears_every_reference(client, db, admin):
|
|
from app.models import (
|
|
Assignment,
|
|
MonthlyPattern,
|
|
NotificationLog,
|
|
RotationMember,
|
|
Schedule,
|
|
SwapRequest,
|
|
)
|
|
|
|
login(client, "admin", "admin-password")
|
|
page = add_person(client, name="Rota Only")
|
|
assert f'href="/admin/users/{db.scalar(select(User.id).where(User.name == "Rota Only"))}/delete"' in page
|
|
person = db.scalar(select(User).where(User.name == "Rota Only"))
|
|
schedule = Schedule(
|
|
name="Almoço - Sábados",
|
|
slug="saturday-lunch",
|
|
schedule_type="rotation",
|
|
weekdays="5",
|
|
requires_manager_approval=True,
|
|
is_active=True,
|
|
rotation_anchor_date=date(2026, 1, 3),
|
|
)
|
|
db.add(schedule)
|
|
db.flush()
|
|
assignment = Assignment(schedule_id=schedule.id, date=date(2026, 10, 3), user_id=person.id)
|
|
pattern = MonthlyPattern(schedule_id=schedule.id, day_of_month=1, user_id=person.id)
|
|
member = RotationMember(schedule_id=schedule.id, user_id=person.id, position=0)
|
|
log = NotificationLog(user_id=person.id, event_type="swap_requested", subject="x", body="x", status="skipped")
|
|
db.add_all([assignment, pattern, member, log])
|
|
db.flush()
|
|
db.add(SwapRequest(assignment_id=assignment.id, requester_id=person.id, status="open"))
|
|
db.commit()
|
|
|
|
# Someone who still holds a shift cannot be removed, not even from the past.
|
|
token = csrf(client, "/admin/users")
|
|
assert client.get(f"/admin/users/{person.id}/delete").status_code == 400
|
|
assert client.post(
|
|
f"/admin/users/{person.id}/delete", data={"csrf_token": token}
|
|
).status_code == 400
|
|
assert db.get(User, person.id) is not None
|
|
assert db.get(Assignment, assignment.id).user_id == person.id
|
|
assert f'href="/admin/users/{person.id}/delete"' not in client.get("/admin/users").text
|
|
|
|
# Freeing the shift makes the removal possible again.
|
|
assignment.user_id = None
|
|
db.commit()
|
|
assert f'href="/admin/users/{person.id}/delete"' in client.get("/admin/users").text
|
|
|
|
confirm = delete_person(client, person.id)
|
|
assert "Remover definitivamente" in confirm
|
|
assert "1 posição(ões) saem das rotações" in confirm
|
|
assert "1 dia(s) do padrão" in confirm
|
|
assert "1 pedido(s) de troca em aberto" in confirm
|
|
|
|
response = client.post(
|
|
f"/admin/users/{person.id}/delete",
|
|
data={"csrf_token": csrf(client, "/admin/users")},
|
|
follow_redirects=True,
|
|
)
|
|
|
|
assert "foi removida" not in response.text
|
|
assert "Rota Only" not in response.text
|
|
assert db.get(User, person.id) is None
|
|
db.refresh(log)
|
|
assert log.user_id is None
|
|
assert db.get(MonthlyPattern, pattern.id).user_id is None
|
|
assert db.get(RotationMember, member.id) is None
|
|
assert db.scalars(
|
|
select(SwapRequest).where(SwapRequest.requester_id == person.id)
|
|
).all() == []
|
|
|
|
|
|
def test_removing_a_person_keeps_other_peoples_requests_intact(client, db, admin):
|
|
from app.models import Assignment, Schedule, SwapRequest
|
|
|
|
login(client, "admin", "admin-password")
|
|
add_person(client, name="Rota Only")
|
|
person = db.scalar(select(User).where(User.name == "Rota Only"))
|
|
schedule = Schedule(
|
|
name="Almoço - Sábados",
|
|
slug="saturday-lunch",
|
|
schedule_type="rotation",
|
|
weekdays="5",
|
|
requires_manager_approval=True,
|
|
is_active=True,
|
|
rotation_anchor_date=date(2026, 1, 3),
|
|
)
|
|
db.add(schedule)
|
|
db.flush()
|
|
assignment = Assignment(schedule_id=schedule.id, date=date(2026, 10, 3), user_id=admin.id)
|
|
db.add(assignment)
|
|
db.flush()
|
|
request = SwapRequest(
|
|
assignment_id=assignment.id,
|
|
requester_id=admin.id,
|
|
target_user_id=person.id,
|
|
status="open",
|
|
)
|
|
db.add(request)
|
|
db.commit()
|
|
|
|
client.post(
|
|
f"/admin/users/{person.id}/delete",
|
|
data={"csrf_token": csrf(client, "/admin/users")},
|
|
follow_redirects=True,
|
|
)
|
|
|
|
db.refresh(request)
|
|
assert db.get(SwapRequest, request.id) is not None
|
|
assert request.target_user_id is None
|
|
assert request.requester_id == admin.id
|
|
|
|
|
|
def test_a_person_cannot_be_removed_by_a_member_or_without_the_right_token(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
add_person(client, name="Rota Only")
|
|
person = db.scalar(select(User).where(User.name == "Rota Only"))
|
|
url = f"/admin/users/{person.id}/delete"
|
|
token = csrf(client, "/admin/users")
|
|
|
|
assert client.post(url, data={"csrf_token": "invalid"}).status_code == 400
|
|
|
|
add_person(client, name="Member")
|
|
member = db.scalar(select(User).where(User.name == "Member"))
|
|
update_person(client, member.id, name="Member", username="member", new_password="member-password")
|
|
client.post("/logout", data={"csrf_token": csrf(client, "/admin/users")})
|
|
token = csrf(client, "/login")
|
|
client.post(
|
|
"/login",
|
|
data={"username": "member", "password": "member-password", "csrf_token": token},
|
|
)
|
|
token = csrf(client, "/")
|
|
assert client.get(url).status_code == 403
|
|
assert client.post(url, data={"csrf_token": token}).status_code == 403
|
|
|
|
client.post("/logout", data={"csrf_token": token})
|
|
assert client.get(url).status_code == 401
|
|
assert client.post(url, data={"csrf_token": "invalid"}).status_code == 401
|
|
assert db.get(User, person.id) is not None
|
|
|
|
|
|
def test_admins_cannot_remove_themselves_or_the_last_admin(client, db, admin):
|
|
login(client, "admin", "admin-password")
|
|
token = csrf(client, "/admin/users")
|
|
|
|
assert client.get(f"/admin/users/{admin.id}/delete").status_code == 400
|
|
assert client.post(f"/admin/users/{admin.id}/delete", data={"csrf_token": token}).status_code == 400
|
|
assert f'href="/admin/users/{admin.id}/delete"' not in client.get("/admin/users").text
|
|
|
|
add_person(client, name="Second Admin")
|
|
second = db.scalar(select(User).where(User.name == "Second Admin"))
|
|
update_person(client, second.id, name="Second Admin", is_admin="on")
|
|
db.refresh(second)
|
|
assert second.is_admin is True
|
|
assert f'href="/admin/users/{second.id}/delete"' in client.get("/admin/users").text
|
|
|
|
client.post(f"/admin/users/{second.id}/delete", data={"csrf_token": token})
|
|
assert db.get(User, second.id) is None
|
|
assert db.get(User, admin.id) is not None
|