mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-22 03:49:18 +02:00
fix: firmware upload only for idle printers (dialog, server, re-check after download, queue waits during upload); AMS Reset Slot asks first and is refused while printing; confirms for shopping list Clear all, Git history Clear and SSO unlink (last SSO link of an account without password stays); LDAP/SMTP test use the form's values; emptying a user's email removes it; local backup delete errors are shown; removed the unreachable printer-card Spoolman link/unlink
This commit is contained in:
maziggy
@@ -0,0 +1,188 @@
|
||||
"""The LDAP and SMTP "Test" buttons test what the form holds, not only what was saved.
|
||||
|
||||
A password left empty in the form keeps the saved one, which still only goes
|
||||
to the server it was entered for.
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
LDAP_SAVED = {
|
||||
"ldap_server_url": "ldaps://ldap.example.com",
|
||||
"ldap_bind_dn": "cn=svc,dc=example,dc=com",
|
||||
"ldap_bind_password": "saved-bind-secret",
|
||||
"ldap_search_base": "dc=example,dc=com",
|
||||
"ldap_security": "ldaps",
|
||||
}
|
||||
|
||||
SMTP_SAVED = {
|
||||
"smtp_host": "smtp.example.com",
|
||||
"smtp_port": 587,
|
||||
"smtp_username": "mailer@example.com",
|
||||
"smtp_password": "saved-smtp-secret",
|
||||
"smtp_security": "starttls",
|
||||
"smtp_auth_enabled": True,
|
||||
"smtp_from_email": "noreply@example.com",
|
||||
}
|
||||
|
||||
|
||||
async def _ldap_test(async_client: AsyncClient, body: dict | None):
|
||||
tested = []
|
||||
|
||||
def fake_test(config):
|
||||
tested.append(config)
|
||||
return True, "ok"
|
||||
|
||||
with patch("backend.app.services.ldap_service.test_ldap_connection", side_effect=fake_test):
|
||||
if body is None:
|
||||
response = await async_client.post("/api/v1/auth/ldap/test")
|
||||
else:
|
||||
response = await async_client.post("/api/v1/auth/ldap/test", json=body)
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json(), tested
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def ldap_saved(async_client: AsyncClient):
|
||||
response = await async_client.put("/api/v1/settings/", json=LDAP_SAVED)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
|
||||
class TestLdap:
|
||||
@pytest.mark.asyncio
|
||||
async def test_tests_the_form_values(self, async_client, ldap_saved):
|
||||
result, tested = await _ldap_test(
|
||||
async_client,
|
||||
{
|
||||
**LDAP_SAVED,
|
||||
"ldap_bind_dn": "cn=other,dc=example,dc=com",
|
||||
"ldap_bind_password": "typed-secret",
|
||||
"ldap_search_base": "ou=people,dc=example,dc=com",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert tested[0].bind_dn == "cn=other,dc=example,dc=com"
|
||||
assert tested[0].bind_password == "typed-secret"
|
||||
assert tested[0].search_base == "ou=people,dc=example,dc=com"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_empty_password_keeps_the_saved_one_for_the_same_server(self, async_client, ldap_saved):
|
||||
_, tested = await _ldap_test(async_client, {**LDAP_SAVED, "ldap_bind_password": ""})
|
||||
|
||||
assert tested[0].bind_password == "saved-bind-secret"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"changed",
|
||||
[{"ldap_server_url": "ldaps://elsewhere.example.net"}, {"ldap_security": "starttls"}],
|
||||
)
|
||||
async def test_the_saved_password_never_goes_to_another_server(self, async_client, ldap_saved, changed):
|
||||
result, tested = await _ldap_test(async_client, {**LDAP_SAVED, "ldap_bind_password": "", **changed})
|
||||
|
||||
assert result["success"] is False
|
||||
assert "again" in result["message"]
|
||||
assert tested == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_without_a_body_it_tests_the_saved_settings(self, async_client, ldap_saved):
|
||||
_, tested = await _ldap_test(async_client, None)
|
||||
|
||||
assert tested[0].server_url == "ldaps://ldap.example.com"
|
||||
assert tested[0].bind_password == "saved-bind-secret"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_saved_ca_certificate_is_used_while_ldap_is_off(self, async_client, db_session, ldap_saved):
|
||||
from backend.app.models.settings import Settings
|
||||
|
||||
db_session.add(Settings(key="ldap_ca_cert_path", value="/certs/ca.pem"))
|
||||
await db_session.commit()
|
||||
|
||||
_, tested = await _ldap_test(async_client, {**LDAP_SAVED, "ldap_bind_password": ""})
|
||||
|
||||
assert tested[0].ca_cert_path == "/certs/ca.pem"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ldap_unset_security_is_starttls_as_when_saving(async_client, db_session):
|
||||
"""Settings saved without a security row: the form's StartTLS is no change of destination."""
|
||||
from backend.app.models.settings import Settings
|
||||
|
||||
for key, value in {**LDAP_SAVED, "ldap_server_url": "ldap://ldap.example.com"}.items():
|
||||
if key != "ldap_security":
|
||||
db_session.add(Settings(key=key, value=value))
|
||||
await db_session.commit()
|
||||
|
||||
result, tested = await _ldap_test(
|
||||
async_client,
|
||||
{
|
||||
**LDAP_SAVED,
|
||||
"ldap_server_url": "ldap://ldap.example.com",
|
||||
"ldap_security": "starttls",
|
||||
"ldap_bind_password": "",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is True, result
|
||||
assert tested[0].bind_password == "saved-bind-secret"
|
||||
|
||||
|
||||
async def _smtp_test(async_client: AsyncClient, smtp: dict | None):
|
||||
body: dict = {"test_recipient": "someone@example.com"}
|
||||
if smtp is not None:
|
||||
body["smtp"] = smtp
|
||||
with patch("backend.app.api.routes.auth.send_email") as send:
|
||||
response = await async_client.post("/api/v1/auth/smtp/test", json=body)
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json(), send
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def smtp_saved(async_client: AsyncClient):
|
||||
response = await async_client.post("/api/v1/auth/smtp", json=SMTP_SAVED)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
|
||||
class TestSmtp:
|
||||
@pytest.mark.asyncio
|
||||
async def test_tests_the_form_values(self, async_client, smtp_saved):
|
||||
result, send = await _smtp_test(
|
||||
async_client, {**SMTP_SAVED, "smtp_from_email": "typed@example.com", "smtp_password": "typed-secret"}
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
used = send.call_args.kwargs["smtp_settings"]
|
||||
assert used.smtp_from_email == "typed@example.com"
|
||||
assert used.smtp_password == "typed-secret"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_empty_password_keeps_the_saved_one_for_the_same_server(self, async_client, smtp_saved):
|
||||
_, send = await _smtp_test(async_client, {**SMTP_SAVED, "smtp_password": ""})
|
||||
|
||||
assert send.call_args.kwargs["smtp_settings"].smtp_password == "saved-smtp-secret"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"changed", [{"smtp_host": "mail.elsewhere.example.net"}, {"smtp_port": 2525}, {"smtp_security": "ssl"}]
|
||||
)
|
||||
async def test_the_saved_password_never_goes_to_another_server(self, async_client, smtp_saved, changed):
|
||||
result, send = await _smtp_test(async_client, {**SMTP_SAVED, "smtp_password": "", **changed})
|
||||
|
||||
assert result["success"] is False
|
||||
assert "again" in result["message"]
|
||||
send.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_without_the_form_it_tests_the_saved_settings(self, async_client, smtp_saved):
|
||||
_, send = await _smtp_test(async_client, None)
|
||||
|
||||
assert send.call_args.kwargs["smtp_settings"].smtp_host == "smtp.example.com"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_form_works_before_anything_was_saved(self, async_client):
|
||||
result, send = await _smtp_test(async_client, SMTP_SAVED)
|
||||
|
||||
assert result["success"] is True
|
||||
assert send.call_args.kwargs["smtp_settings"].smtp_password == "saved-smtp-secret"
|
||||
@@ -0,0 +1,177 @@
|
||||
"""Actions that only an idle printer takes: firmware upload and AMS slot reset.
|
||||
|
||||
Firmware goes onto the SD card the printer prints from, and a slot reset
|
||||
clears the filament setting of a tray a running print may be feeding from.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.services.firmware_update import FirmwareUpdateService, get_upload_state, reset_upload_state
|
||||
|
||||
|
||||
def _client(state: str) -> MagicMock:
|
||||
client = MagicMock()
|
||||
client.state = SimpleNamespace(state=state, firmware_version="01.06.00.00", sdcard=True, connected=True)
|
||||
client.reset_ams_slot.return_value = True
|
||||
return client
|
||||
|
||||
|
||||
def _firmware_service() -> MagicMock:
|
||||
service = MagicMock()
|
||||
service.check_for_update = AsyncMock(return_value={"update_available": True, "latest_version": "01.07.00.00"})
|
||||
service.get_firmware_file_info = AsyncMock(return_value={"filename": "fw.bin"})
|
||||
service.download_firmware = AsyncMock()
|
||||
service.get_latest_version = AsyncMock(return_value=None)
|
||||
return service
|
||||
|
||||
|
||||
def _patched(client: MagicMock, service: MagicMock):
|
||||
return (
|
||||
patch("backend.app.services.firmware_update.printer_manager.get_client", return_value=client),
|
||||
patch("backend.app.services.firmware_update.get_firmware_service", return_value=service),
|
||||
patch(
|
||||
"backend.app.services.firmware_update.get_storage_info_async",
|
||||
AsyncMock(return_value={"free_bytes": 10 * 1024**3}),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class TestFirmwareUpload:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("state", ["RUNNING", "PAUSE", "PREPARE", "SLICING", "unknown"])
|
||||
async def test_a_busy_printer_is_refused_in_the_prepare_check(self, async_client, printer_factory, state):
|
||||
printer = await printer_factory()
|
||||
a, b, c = _patched(_client(state), _firmware_service())
|
||||
with a, b, c:
|
||||
response = await async_client.get(f"/api/v1/firmware/updates/{printer.id}/prepare")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["can_proceed"] is False
|
||||
assert any("idle" in error for error in body["errors"])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("state", ["IDLE", "FINISH", "FAILED"])
|
||||
async def test_an_idle_printer_passes_the_prepare_check(self, async_client, printer_factory, state):
|
||||
printer = await printer_factory()
|
||||
a, b, c = _patched(_client(state), _firmware_service())
|
||||
with a, b, c:
|
||||
response = await async_client.get(f"/api/v1/firmware/updates/{printer.id}/prepare")
|
||||
|
||||
assert response.json()["can_proceed"] is True, response.json()["errors"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_upload_route_refuses_a_printing_printer(self, async_client, printer_factory):
|
||||
printer = await printer_factory()
|
||||
a, b, c = _patched(_client("RUNNING"), _firmware_service())
|
||||
with (
|
||||
a,
|
||||
b,
|
||||
c,
|
||||
patch.object(FirmwareUpdateService, "start_upload", AsyncMock(return_value=True)) as start,
|
||||
):
|
||||
response = await async_client.post(f"/api/v1/firmware/updates/{printer.id}/upload")
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "idle" in response.json()["detail"]
|
||||
start.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_print_started_during_the_download_gets_no_firmware(self, tmp_path):
|
||||
"""The download can take minutes; the card is checked again before writing to it."""
|
||||
client = _client("IDLE")
|
||||
service = _firmware_service()
|
||||
firmware = tmp_path / "fw.bin"
|
||||
firmware.write_bytes(b"x")
|
||||
|
||||
async def download(*_args, **_kwargs):
|
||||
client.state.state = "RUNNING"
|
||||
return firmware
|
||||
|
||||
service.download_firmware = AsyncMock(side_effect=download)
|
||||
upload = AsyncMock(return_value=True)
|
||||
reset_upload_state(9101)
|
||||
with (
|
||||
patch("backend.app.services.firmware_update.printer_manager.get_client", return_value=client),
|
||||
patch("backend.app.services.firmware_update.get_firmware_service", return_value=service),
|
||||
patch("backend.app.services.firmware_update.upload_file_async", upload),
|
||||
patch(
|
||||
"backend.app.services.firmware_update.get_ftp_retry_settings",
|
||||
AsyncMock(return_value=(False, 0, 0, 30)),
|
||||
),
|
||||
):
|
||||
await FirmwareUpdateService()._do_upload(9101, "192.0.2.10", "12345678", "X1C", "01.07.00.00")
|
||||
|
||||
upload.assert_not_called()
|
||||
state = get_upload_state(9101)
|
||||
assert state.status.value == "error"
|
||||
assert "idle" in (state.error or "")
|
||||
reset_upload_state(9101)
|
||||
|
||||
|
||||
class TestAmsSlotReset:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("state", ["RUNNING", "PAUSE", "PREPARE", "SLICING"])
|
||||
async def test_refused_while_a_print_is_loaded(self, async_client, printer_factory, state):
|
||||
printer = await printer_factory()
|
||||
client = _client(state)
|
||||
with patch("backend.app.api.routes.printers.printer_manager.get_client", return_value=client):
|
||||
response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/0/tray/1/reset")
|
||||
|
||||
assert response.status_code == 409
|
||||
client.reset_ams_slot.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("state", ["IDLE", "FINISH", "FAILED"])
|
||||
async def test_sent_when_no_print_is_loaded(self, async_client, printer_factory, state):
|
||||
printer = await printer_factory()
|
||||
client = _client(state)
|
||||
with patch("backend.app.api.routes.printers.printer_manager.get_client", return_value=client):
|
||||
response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/0/tray/1/reset")
|
||||
|
||||
assert response.status_code == 200
|
||||
client.reset_ams_slot.assert_called_once_with(ams_id=0, tray_id=1)
|
||||
|
||||
|
||||
class TestQueueDuringFirmwareUpload:
|
||||
"""A queued print does not start while firmware is being put on the printer's card."""
|
||||
|
||||
@pytest.fixture
|
||||
def idle_printer(self):
|
||||
from backend.app.services.printer_manager import printer_manager
|
||||
|
||||
printer_id = 9102
|
||||
client = MagicMock()
|
||||
client.state = SimpleNamespace(state="IDLE", connected=True)
|
||||
printer_manager._clients[printer_id] = client
|
||||
reset_upload_state(printer_id)
|
||||
try:
|
||||
with (
|
||||
patch.object(printer_manager, "is_connected", return_value=True),
|
||||
patch.object(printer_manager, "get_status", return_value=client.state),
|
||||
):
|
||||
yield printer_id
|
||||
finally:
|
||||
printer_manager._clients.pop(printer_id, None)
|
||||
reset_upload_state(printer_id)
|
||||
|
||||
@pytest.mark.parametrize("status", ["preparing", "downloading", "uploading"])
|
||||
def test_not_idle_while_an_upload_runs(self, idle_printer, status):
|
||||
from backend.app.services.firmware_update import FirmwareUploadStatus
|
||||
from backend.app.services.print_scheduler import PrintScheduler
|
||||
|
||||
get_upload_state(idle_printer).status = FirmwareUploadStatus(status)
|
||||
|
||||
assert PrintScheduler()._is_printer_idle(idle_printer, require_plate_clear=False) is False
|
||||
|
||||
@pytest.mark.parametrize("status", ["idle", "complete", "error"])
|
||||
def test_idle_again_once_it_is_over(self, idle_printer, status):
|
||||
from backend.app.services.firmware_update import FirmwareUploadStatus
|
||||
from backend.app.services.print_scheduler import PrintScheduler
|
||||
|
||||
get_upload_state(idle_printer).status = FirmwareUploadStatus(status)
|
||||
|
||||
assert PrintScheduler()._is_printer_idle(idle_printer, require_plate_clear=False) is True
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Removing a linked SSO account never removes a user's last way to sign in.
|
||||
|
||||
A user Bambuddy created from an SSO login has no password, so their last
|
||||
link stays; a user with a password, or with another link, can remove it.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.core.auth import create_access_token, get_password_hash
|
||||
from backend.app.models.oidc_provider import OIDCProvider, UserOIDCLink
|
||||
from backend.app.models.user import User
|
||||
from backend.tests.integration.test_mfa_api import _setup_and_login
|
||||
|
||||
|
||||
async def _provider(db_session, name: str) -> int:
|
||||
provider = OIDCProvider(name=name, issuer_url=f"https://{name}.example.com", client_id="bambuddy")
|
||||
provider.client_secret = "provider-secret"
|
||||
db_session.add(provider)
|
||||
await db_session.commit()
|
||||
return provider.id
|
||||
|
||||
|
||||
async def _user(db_session, *, oidc_only: bool, providers: list[int]) -> tuple[User, dict]:
|
||||
user = User(
|
||||
username="sso-user",
|
||||
password_hash=None if oidc_only else get_password_hash("Whatever1!"),
|
||||
auth_source="oidc" if oidc_only else "local",
|
||||
role="user",
|
||||
is_active=True,
|
||||
)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
for i, provider_id in enumerate(providers):
|
||||
db_session.add(UserOIDCLink(user_id=user.id, provider_id=provider_id, provider_user_id=f"sub-{i}"))
|
||||
await db_session.commit()
|
||||
return user, {"Authorization": f"Bearer {create_access_token(data={'sub': user.username})}"}
|
||||
|
||||
|
||||
async def _links(db_session, user_id: int) -> int:
|
||||
db_session.expire_all()
|
||||
return len((await db_session.execute(select(UserOIDCLink).where(UserOIDCLink.user_id == user_id))).all())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_last_link_of_a_user_without_password_stays(async_client: AsyncClient, db_session):
|
||||
await _setup_and_login(async_client, "ssoadmin", "SsoAdmin12!")
|
||||
provider = await _provider(db_session, "alpha")
|
||||
user, headers = await _user(db_session, oidc_only=True, providers=[provider])
|
||||
|
||||
response = await async_client.delete(f"/api/v1/auth/oidc/links/{provider}", headers=headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert await _links(db_session, user.id) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_user_without_password_can_remove_one_of_two_links(async_client: AsyncClient, db_session):
|
||||
await _setup_and_login(async_client, "ssoadmin", "SsoAdmin12!")
|
||||
first, second = await _provider(db_session, "alpha"), await _provider(db_session, "beta")
|
||||
user, headers = await _user(db_session, oidc_only=True, providers=[first, second])
|
||||
|
||||
response = await async_client.delete(f"/api/v1/auth/oidc/links/{first}", headers=headers)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert await _links(db_session, user.id) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_user_with_a_password_can_remove_the_last_link(async_client: AsyncClient, db_session):
|
||||
await _setup_and_login(async_client, "ssoadmin", "SsoAdmin12!")
|
||||
provider = await _provider(db_session, "alpha")
|
||||
user, headers = await _user(db_session, oidc_only=False, providers=[provider])
|
||||
|
||||
response = await async_client.delete(f"/api/v1/auth/oidc/links/{provider}", headers=headers)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert await _links(db_session, user.id) == 0
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Editing a user can remove their email: an empty email clears it.
|
||||
|
||||
Left out or null, the email stays as it is. With advanced authentication on,
|
||||
a user needs an email (password resets go there), so it can't be removed.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.models.settings import Settings
|
||||
from backend.app.models.user import User
|
||||
|
||||
|
||||
async def _user(db_session, email: str | None = "mia@example.invalid") -> User:
|
||||
from backend.app.core.auth import get_password_hash
|
||||
|
||||
user = User(username="mia", password_hash=get_password_hash("Whatever1!"), email=email, role="user", is_active=True)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
return user
|
||||
|
||||
|
||||
async def _email(db_session, user_id: int) -> str | None:
|
||||
db_session.expire_all()
|
||||
return (await db_session.execute(select(User.email).where(User.id == user_id))).scalar_one()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("email", ["", " "])
|
||||
async def test_an_empty_email_clears_it(async_client: AsyncClient, db_session, email):
|
||||
user = await _user(db_session)
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json={"email": email})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["email"] is None
|
||||
assert await _email(db_session, user.id) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("body", [{}, {"email": None}, {"role": "user"}])
|
||||
async def test_left_out_or_null_keeps_it(async_client: AsyncClient, db_session, body):
|
||||
user = await _user(db_session)
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json=body)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert await _email(db_session, user.id) == "mia@example.invalid"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_two_users_without_email_are_no_duplicate(async_client: AsyncClient, db_session):
|
||||
from backend.app.core.auth import get_password_hash
|
||||
|
||||
db_session.add(User(username="other", password_hash=get_password_hash("Whatever1!"), email=None, role="user"))
|
||||
user = await _user(db_session)
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json={"email": ""})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refused_while_advanced_authentication_is_on(async_client: AsyncClient, db_session):
|
||||
user = await _user(db_session)
|
||||
db_session.add(Settings(key="advanced_auth_enabled", value="true"))
|
||||
await db_session.commit()
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json={"email": ""})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "advanced authentication" in response.json()["detail"]
|
||||
assert await _email(db_session, user.id) == "mia@example.invalid"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_user_without_email_can_still_be_edited_with_advanced_authentication(
|
||||
async_client: AsyncClient, db_session
|
||||
):
|
||||
user = await _user(db_session, email=None)
|
||||
db_session.add(Settings(key="advanced_auth_enabled", value="true"))
|
||||
await db_session.commit()
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json={"email": "", "role": "user"})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert await _email(db_session, user.id) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refused_while_the_user_signs_in_with_an_email_code(async_client: AsyncClient, db_session):
|
||||
"""Email 2FA needs SMTP, not advanced authentication: the code goes to this address."""
|
||||
user = await _user(db_session)
|
||||
db_session.add(Settings(key=f"user_{user.id}_email_2fa_enabled", value="true"))
|
||||
await db_session.commit()
|
||||
|
||||
response = await async_client.patch(f"/api/v1/users/{user.id}", json={"email": ""})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "email" in response.json()["detail"].lower()
|
||||
assert await _email(db_session, user.id) == "mia@example.invalid"
|
||||
Reference in New Issue
Block a user