Ebbe BaßandClaude Sonnet 5.5
3fd8f27127
Profile page at /profile with first name, last name, display name, telephone and mobile read from AD and shown with the username and e-mail. Only changed fields are written through the directory adapter in one transaction with the audit entry (field names only); numbers are accepted in any common notation and stored as +country code and digits; names are required. Plain-language errors per field, German text and tests; review findings logged in the plan. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FocPJ75XPma7TcuYEMy8zm
412 lines
15 KiB
Python
412 lines
15 KiB
Python
"""Profile edit: the page, the phone format, the diff against AD, audit and errors (FR-9)."""
|
|
|
|
import logging
|
|
import re
|
|
|
|
import pytest
|
|
from django.test import Client
|
|
|
|
from portal.adapters.directory import DirectoryUnavailable, Forbidden, ValidationFailed
|
|
from portal.audit import services as audit_services
|
|
from portal.audit.models import AuditEntry
|
|
from portal.profile.forms import InvalidNumber, normalize_phone
|
|
from portal.tests.helpers import sign_in
|
|
|
|
pytestmark = pytest.mark.django_db(transaction=True)
|
|
|
|
VALID = {
|
|
"first_name": "Alice",
|
|
"last_name": "Anders",
|
|
"display_name": "Alice Anders",
|
|
"telephone": "+49301234567",
|
|
"mobile": "intern 23",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def people(adapters):
|
|
d = adapters.directory
|
|
d.add_user(
|
|
"alice",
|
|
given_name="Alice",
|
|
surname="Anders",
|
|
display_name="Alice Anders",
|
|
mail="alice@example.test",
|
|
telephone="+49301234567",
|
|
mobile="intern 23", # an old value that is not a number
|
|
)
|
|
d.add_user("bob", given_name="Bob", surname="Brown", display_name="Bob Brown")
|
|
return d
|
|
|
|
|
|
def signed_in(adapters, name="alice"):
|
|
client, _ = sign_in(adapters, name)
|
|
return client
|
|
|
|
|
|
def save(client, **changes):
|
|
return client.post("/profile", {**VALID, **changes})
|
|
|
|
|
|
def profile_entries():
|
|
return AuditEntry.objects.filter(action__startswith="profile.")
|
|
|
|
|
|
def audit_actions():
|
|
return list(profile_entries().values_list("action", flat=True))
|
|
|
|
|
|
def stored(adapters, sam="alice"):
|
|
return adapters.directory.find_user(sam)
|
|
|
|
|
|
class Spy:
|
|
def __init__(self, directory):
|
|
self.writes = []
|
|
original = directory.update_attributes
|
|
|
|
def wrapper(sam, changes):
|
|
self.writes.append((sam, dict(changes)))
|
|
return original(sam, changes)
|
|
|
|
directory.update_attributes = wrapper
|
|
|
|
|
|
# -- the page ------------------------------------------------------------------------------
|
|
|
|
|
|
def test_the_page_shows_the_ad_values_and_plain_text_for_username_and_mail(people, adapters):
|
|
html = signed_in(adapters).get("/profile").content.decode()
|
|
for value in ("Alice", "Anders", "Alice Anders", "+49301234567", "intern 23"):
|
|
assert f'value="{value}"' in html
|
|
assert "alice@example.test" in html
|
|
assert 'name="username"' not in html and 'name="mail"' not in html
|
|
assert re.search(r'<a href="/profile" class="current" aria-current="page">', html)
|
|
|
|
|
|
def test_the_page_speaks_german_in_both_registers(people, adapters):
|
|
client = signed_in(adapters)
|
|
client.cookies["pm_lang"] = "de"
|
|
assert "Diese Angaben sind in Ihrem Konto" in client.get("/profile").content.decode()
|
|
client.cookies["pm_form"] = "du"
|
|
assert "in deinem Konto" in client.get("/profile").content.decode()
|
|
|
|
|
|
def test_the_page_is_unavailable_when_the_directory_is_down(people, adapters):
|
|
client = signed_in(adapters)
|
|
adapters.directory.unavailable = True
|
|
response = client.get("/profile")
|
|
assert response.status_code == 503 and b'name="first_name"' not in response.content
|
|
|
|
|
|
def test_access_rules(people, adapters):
|
|
assert Client().get("/profile").status_code == 302
|
|
assert Client().post("/profile", VALID).status_code == 302
|
|
assert signed_in(adapters).put("/profile").status_code == 405
|
|
client = signed_in(adapters)
|
|
strict = Client(enforce_csrf_checks=True)
|
|
strict.cookies = client.cookies
|
|
assert strict.post("/profile", {**VALID, "first_name": "Eve"}).status_code == 403
|
|
assert stored(adapters).given_name == "Alice"
|
|
|
|
|
|
# -- saving ----------------------------------------------------------------------------------
|
|
|
|
|
|
def test_only_the_changed_fields_are_written_and_audited_without_values(people, adapters):
|
|
spy = Spy(adapters.directory)
|
|
client = signed_in(adapters)
|
|
response = save(client, first_name="Alicia", mobile="0171 / 123456")
|
|
assert response.status_code == 302 and response["Location"] == "/profile"
|
|
assert spy.writes == [("alice", {"givenName": "Alicia", "mobile": "+49171123456"})]
|
|
account = stored(adapters)
|
|
assert (account.given_name, account.mobile) == ("Alicia", "+49171123456")
|
|
assert (account.surname, account.telephone, account.mail) == (
|
|
"Anders",
|
|
"+49301234567",
|
|
"alice@example.test",
|
|
)
|
|
entry = AuditEntry.objects.get(action="profile.updated")
|
|
assert entry.actor == "alice" and entry.ip == "127.0.0.1"
|
|
assert entry.detail == {"fields": ["first_name", "mobile"]}
|
|
assert profile_entries().count() == 1
|
|
|
|
|
|
def test_the_saved_notice_shows_once_with_the_stored_number_form(people, adapters):
|
|
client = signed_in(adapters)
|
|
response = save(client, telephone="0049 (0) 30 1234568")
|
|
page = client.get(response["Location"]).content.decode()
|
|
assert "Your changes were saved." in page and 'value="+49301234568"' in page
|
|
assert "Your changes were saved." not in client.get("/profile").content.decode()
|
|
|
|
|
|
def test_nothing_changed_makes_no_write_and_no_audit(people, adapters):
|
|
spy = Spy(adapters.directory)
|
|
client = signed_in(adapters)
|
|
response = save(client)
|
|
assert response.status_code == 302 and spy.writes == [] and profile_entries().count() == 0
|
|
assert "Nothing to change." in client.get("/profile").content.decode()
|
|
|
|
|
|
def test_a_number_equal_once_normalized_is_not_a_change(people, adapters):
|
|
spy = Spy(adapters.directory)
|
|
save(signed_in(adapters), telephone="+49 30 1234567")
|
|
assert spy.writes == [] and profile_entries().count() == 0
|
|
|
|
|
|
def test_an_empty_number_clears_the_attribute(people, adapters):
|
|
spy = Spy(adapters.directory)
|
|
save(signed_in(adapters), telephone="")
|
|
assert spy.writes == [("alice", {"telephoneNumber": ""})]
|
|
assert stored(adapters).telephone == ""
|
|
|
|
|
|
def test_an_old_value_that_is_not_a_number_never_blocks_another_field(people, adapters):
|
|
save(signed_in(adapters), last_name="Ander")
|
|
account = stored(adapters)
|
|
assert account.surname == "Ander" and account.mobile == "intern 23"
|
|
|
|
|
|
def test_old_values_that_break_the_rules_are_left_alone_but_changed_ones_are_checked(
|
|
people, adapters
|
|
):
|
|
long_name = "x" * 100
|
|
adapters.directory.add_user("carol", given_name=long_name, surname="", display_name="Carol")
|
|
client = signed_in(adapters, "carol")
|
|
spy = Spy(adapters.directory)
|
|
old = {
|
|
"first_name": long_name,
|
|
"last_name": "",
|
|
"display_name": "Carol",
|
|
"telephone": "",
|
|
"mobile": "",
|
|
}
|
|
assert client.post("/profile", {**old, "mobile": "5"}).status_code == 400
|
|
assert client.post("/profile", {**old, "display_name": "Caro"}).status_code == 302
|
|
assert spy.writes == [("carol", {"displayName": "Caro"})]
|
|
|
|
|
|
def test_the_session_name_follows_the_saved_display_name(people, adapters):
|
|
client = signed_in(adapters)
|
|
assert "Alice Anders" in client.get("/").content.decode()
|
|
save(client, display_name="Ally")
|
|
assert "Ally" in client.get("/profile").content.decode()
|
|
assert "Alice Anders" not in client.get("/").content.decode()
|
|
|
|
|
|
def test_the_save_compares_with_a_fresh_read_not_with_the_page(people, adapters):
|
|
spy = Spy(adapters.directory)
|
|
client = signed_in(adapters)
|
|
client.get("/profile")
|
|
adapters.directory.update_attributes("alice", {"sn": "Elsewhere"})
|
|
save(client, display_name="Ally", last_name="Elsewhere") # same as AD now: not written
|
|
assert spy.writes[1:] == [("alice", {"displayName": "Ally"})]
|
|
|
|
|
|
# -- phone numbers ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "result"),
|
|
[
|
|
("0171 / 123456", "+49171123456"),
|
|
("0171123456", "+49171123456"),
|
|
("+49 171 123456", "+49171123456"),
|
|
("+49 (0) 171-123456", "+49171123456"),
|
|
("0049 171 123456", "+49171123456"),
|
|
("0049 (0) 171 123456", "+49171123456"),
|
|
("+43 660 1234567", "+436601234567"),
|
|
("0043 660 1234567", "+436601234567"),
|
|
("(0171) 123.456", "+49171123456"),
|
|
("+49171123456", "+49171123456"),
|
|
],
|
|
)
|
|
def test_phone_inputs_and_their_stored_form(text, result):
|
|
assert normalize_phone(text) == result
|
|
assert normalize_phone(result) == result
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"text",
|
|
[
|
|
"171123456",
|
|
"abc",
|
|
"0171 abc",
|
|
"+49 171 12+3456",
|
|
"+49 1711 23456 7890 1234",
|
|
"0123",
|
|
"+49 12",
|
|
"++49171123456",
|
|
"intern 23",
|
|
"",
|
|
"*)(uid=*",
|
|
],
|
|
)
|
|
def test_phone_inputs_that_are_refused(text):
|
|
with pytest.raises(InvalidNumber):
|
|
normalize_phone(text)
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["telephone", "mobile"])
|
|
@pytest.mark.parametrize("text", ["171123456", "0171 abc", "+49 12", "+49 171 12+3456"])
|
|
def test_a_bad_number_is_refused_under_its_field(people, adapters, field, text):
|
|
spy = Spy(adapters.directory)
|
|
response = save(signed_in(adapters), **{field: text})
|
|
html = response.content.decode()
|
|
assert response.status_code == 400 and spy.writes == []
|
|
assert "must be a number that starts with + and the country code, or with 0" in html
|
|
assert 'class="error"' in html and profile_entries().count() == 0
|
|
|
|
|
|
# -- invalid values ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["first_name", "last_name", "display_name"])
|
|
def test_a_required_name_cannot_be_empty(people, adapters, field):
|
|
spy = Spy(adapters.directory)
|
|
response = save(signed_in(adapters), **{field: " "})
|
|
assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
|
|
assert "is required." in response.content.decode()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("first_name", "x" * 65),
|
|
("last_name", "x" * 65),
|
|
("display_name", "x" * 257),
|
|
("first_name", "A\nB"),
|
|
("last_name", "A\tB"),
|
|
("display_name", "A\x00B"),
|
|
],
|
|
)
|
|
def test_too_long_or_control_characters_are_refused_and_other_input_stays(
|
|
people, adapters, field, value
|
|
):
|
|
spy = Spy(adapters.directory)
|
|
response = save(signed_in(adapters), mobile="0171 5", **{field: value})
|
|
html = response.content.decode()
|
|
assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
|
|
assert "Your changes were not saved." in html
|
|
assert 'value="0171 5"' in html # what else was typed stays in the form
|
|
assert f'<a href="#id_{field}">' in html
|
|
|
|
|
|
def test_the_limits_are_ads_limits(people, adapters):
|
|
save(signed_in(adapters), first_name="x" * 64, display_name="y" * 256)
|
|
account = stored(adapters)
|
|
assert len(account.given_name) == 64 and len(account.display_name) == 256
|
|
|
|
|
|
def test_values_are_trimmed(people, adapters):
|
|
save(signed_in(adapters), first_name=" Alicia ")
|
|
assert stored(adapters).given_name == "Alicia"
|
|
|
|
|
|
@pytest.mark.parametrize("text", ["*)(uid=*", "\\00", "CN=x,DC=y", "a)(|(cn=*"])
|
|
def test_injection_strings_are_plain_text_and_touch_nothing_else(people, adapters, text):
|
|
spy = Spy(adapters.directory)
|
|
before_bob = stored(adapters, "bob")
|
|
assert save(signed_in(adapters), display_name=text).status_code == 302
|
|
assert spy.writes == [("alice", {"displayName": text})]
|
|
assert stored(adapters).display_name == text
|
|
assert stored(adapters, "bob") == before_bob
|
|
|
|
|
|
# -- who is edited -----------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_posted_username_is_ignored_and_another_user_is_untouched(people, adapters):
|
|
before_bob = stored(adapters, "bob")
|
|
response = save(signed_in(adapters), first_name="Alicia", username="bob", sam="bob", dn="x")
|
|
assert response.status_code == 302
|
|
assert stored(adapters).given_name == "Alicia" and stored(adapters, "bob") == before_bob
|
|
assert AuditEntry.objects.get(action="profile.updated").actor == "alice"
|
|
|
|
|
|
# -- failures ------------------------------------------------------------------------------------
|
|
|
|
|
|
def test_ad_refusing_a_value_changes_nothing_and_is_audited_as_failure(people, adapters):
|
|
def refuse(sam, changes):
|
|
raise ValidationFailed("constraint")
|
|
|
|
adapters.directory.update_attributes = refuse
|
|
response = save(signed_in(adapters), first_name="Alicia")
|
|
html = response.content.decode()
|
|
assert response.status_code == 400
|
|
assert "A value was not accepted. Please check your entries." in html
|
|
assert "constraint" not in html and 'value="Alicia"' in html
|
|
assert stored(adapters).given_name == "Alice"
|
|
assert audit_actions() == ["profile.update_failed"]
|
|
assert profile_entries().get().detail == {"reason": "refused"}
|
|
|
|
|
|
def test_directory_down_on_save_is_503_and_audited(people, adapters):
|
|
client = signed_in(adapters)
|
|
adapters.directory.unavailable = True
|
|
response = save(client, first_name="Alicia")
|
|
assert response.status_code == 503
|
|
assert "This is not possible right now. Please try again later." in response.content.decode()
|
|
adapters.directory.unavailable = False
|
|
assert stored(adapters).given_name == "Alice"
|
|
assert audit_actions() == ["profile.update_failed"]
|
|
assert profile_entries().get().detail == {"reason": "unavailable"}
|
|
|
|
|
|
def test_directory_failing_during_the_write_rolls_back_and_audits(people, adapters):
|
|
def down(sam, changes):
|
|
raise DirectoryUnavailable("gone")
|
|
|
|
adapters.directory.update_attributes = down
|
|
assert save(signed_in(adapters), first_name="Alicia").status_code == 503
|
|
assert audit_actions() == ["profile.update_failed"]
|
|
|
|
|
|
def test_a_failure_after_the_ad_call_is_logged_and_the_user_told_it_is_unclear(
|
|
people, adapters, monkeypatch, caplog
|
|
):
|
|
real_record = audit_services.record
|
|
|
|
def broken(*args, **kwargs):
|
|
raise RuntimeError("database gone: Alicia")
|
|
|
|
client = signed_in(adapters)
|
|
monkeypatch.setattr(audit_services, "record", broken)
|
|
with caplog.at_level(logging.ERROR):
|
|
response = save(client, first_name="Alicia", mobile="0171 5555555")
|
|
monkeypatch.setattr(audit_services, "record", real_record)
|
|
assert response.status_code == 500 and "may have been changed" in response.content.decode()
|
|
assert stored(adapters).given_name == "Alicia" # AD did change
|
|
assert audit_actions() == ["profile.update_failed"] # no "updated" entry survived
|
|
log = next(r.getMessage() for r in caplog.records if r.levelno == logging.ERROR)
|
|
assert "user=alice" in log and "fields=first_name,mobile" in log and "step=audit" in log
|
|
assert "Alicia" not in log and "5555555" not in log
|
|
|
|
|
|
def test_a_missing_write_right_is_the_not_possible_page_and_a_failure_entry(
|
|
people, adapters, caplog
|
|
):
|
|
def forbidden(sam, changes):
|
|
raise Forbidden("insufficient access: Alicia")
|
|
|
|
client = signed_in(adapters)
|
|
adapters.directory.update_attributes = forbidden
|
|
with caplog.at_level(logging.ERROR):
|
|
response = save(client, first_name="Alicia")
|
|
assert response.status_code == 503
|
|
assert "This is not possible right now." in response.content.decode()
|
|
assert audit_actions() == ["profile.update_failed"]
|
|
assert profile_entries().get().detail == {"reason": "forbidden"}
|
|
assert not any("Alicia" in r.getMessage() for r in caplog.records)
|
|
|
|
|
|
def test_an_invalid_save_reads_the_directory_once(people, adapters):
|
|
client = signed_in(adapters)
|
|
reads = []
|
|
original = adapters.directory.find_user
|
|
adapters.directory.find_user = lambda sam: reads.append(sam) or original(sam)
|
|
response = save(client, first_name=" ")
|
|
assert response.status_code == 400 and len(reads) == 1
|
|
assert "alice@example.test" in response.content.decode()
|