Alias tests

hushline/model.py

@@ -108,6 +108,9 @@ def extra_fields(self) -> Generator[ExtraField, None, None]:
     def valid_fields(self) -> Sequence[ExtraField]:
         return [x for x in self.extra_fields if x.label and x.value]
 
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} id={self.id} username={self.username}>"
+
 
 class User(Model):
     __tablename__ = "users"

hushline/settings/__init__.py

@@ -171,14 +171,15 @@ def handle_change_username_form(
 
 
 def handle_new_alias_form(user: User, new_alias_form: NewAliasForm, redirect_url: str) -> Response:
-    current_app.logger.debug("Creating alias for {user.primary_username.username}")
+    current_app.logger.debug(f"Creating alias for {user.primary_username.username}")
     # TODO check that users are allowed to add aliases here (is premium, not too many)
     uname = Username(_username=new_alias_form.username.data, user_id=user.id, is_primary=False)
     db.session.add(uname)
     try:
         db.session.commit()
     except IntegrityError as e:
-        if isinstance(e.orig, UniqueViolation) and '"usernames_username_key"' in str(e.orig):
+        db.session.rollback()
+        if isinstance(e.orig, UniqueViolation) and '"uq_usernames_username"' in str(e.orig):
             flash("💔 This username is already taken.")
         else:
             flash("⛔️ Internal server error. Alias not created.")
@@ -216,7 +217,6 @@ async def index() -> str | Response:
         profile_form = ProfileForm()
 
         if request.method == "POST":
-            # Update bio and custom fields
             if "update_bio" in request.form and profile_form.validate_on_submit():
                 return await handle_update_bio(
                     user.primary_username, profile_form, url_for(".index")
@@ -615,7 +615,12 @@ def delete_account() -> Response | str:
 
         user = db.session.get(User, user_id)
         if user:
-            db.session.execute(db.delete(Message).filter_by(user_id=user.id))
+            db.session.execute(
+                db.delete(Message).filter(
+                    Message.username_id.in_(db.select(Username.id).filter_by(user_id=user.id))
+                )
+            )
+            db.session.execute(db.delete(Username).filter_by(user_id=user.id))
             db.session.delete(user)
             db.session.commit()
 

tests/conftest.py

@@ -17,7 +17,7 @@
 from hushline import create_app
 from hushline.crypto import _SCRYPT_PARAMS
 from hushline.db import db
-from hushline.model import User, Username
+from hushline.model import Message, User, Username
 
 CONN_FMT_STR = "postgresql+psycopg://hushline:hushline@postgres:5432/{database}"
 TEMPLATE_DB_NAME = "app_db_template"
@@ -163,3 +163,21 @@ def _pgp_user(client: FlaskClient, user: User) -> None:
     with open("tests/test_pgp_key.txt") as f:
         user.pgp_key = f.read()
     db.session.commit()
+
+
+@pytest.fixture()
+def user_alias(app: Flask, user: User) -> Username:
+    uuid_ish = str(uuid4())[0:12]
+    username = Username(user_id=user.id, _username=f"test-{uuid_ish}", is_primary=False)
+    db.session.add(username)
+    db.session.commit()
+
+    return username
+
+
+@pytest.fixture()
+def message(app: Flask, user: User) -> Message:
+    msg = Message(content=str(uuid4()), username_id=user.primary_username.id)
+    db.session.add(msg)
+    db.session.commit()
+    return msg

tests/test_profile.py

@@ -4,7 +4,7 @@
 from flask.testing import FlaskClient
 
 from hushline.db import db
-from hushline.model import Message, User
+from hushline.model import Message, User, Username
 
 
 def get_captcha_from_session(client: FlaskClient, username: str) -> str:
@@ -46,6 +46,32 @@ def test_profile_submit_message(client: FlaskClient, user: User) -> None:
     assert msg_content in response.text, response.text
 
 
+@pytest.mark.usefixtures("_authenticated_user")
+def test_profile_submit_message_to_alias(
+    client: FlaskClient, user: User, user_alias: Username
+) -> None:
+    msg_content = "This is a test message."
+
+    response = client.post(
+        url_for("profile", username=user_alias.username),
+        data={
+            "content": msg_content,
+            "client_side_encrypted": "false",
+            "captcha_answer": get_captcha_from_session(client, user.primary_username.username),
+        },
+        follow_redirects=True,
+    )
+    assert response.status_code == 200, response.text
+    assert "Message submitted successfully." in response.text
+
+    message = db.session.scalars(db.select(Message).filter_by(username_id=user_alias.id)).one()
+    assert message.content == msg_content
+
+    response = client.get(url_for("inbox", username=user_alias.username), follow_redirects=True)
+    assert response.status_code == 200
+    assert msg_content in response.text, response.text
+
+
 @pytest.mark.usefixtures("_authenticated_user")
 def test_profile_submit_message_with_contact_method(client: FlaskClient, user: User) -> None:
     message_content = "This is a test message."

tests/test_settings.py

@@ -1,11 +1,12 @@
 from unittest.mock import ANY, MagicMock, patch
+from uuid import uuid4
 
 import pytest
 from flask import url_for
 from flask.testing import FlaskClient
 
 from hushline.db import db
-from hushline.model import SMTPEncryption, User, Username
+from hushline.model import Message, SMTPEncryption, User, Username
 
 
 @pytest.mark.usefixtures("_authenticated_user")
@@ -301,3 +302,179 @@ def test_update_smtp_settings_default_forwarding(
     assert updated_user.smtp_password is None
     assert updated_user.smtp_encryption.value == SMTPEncryption.default().value
     assert updated_user.smtp_sender is None
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_add_alias(client: FlaskClient, user: User) -> None:
+    alias_username = str(uuid4())[0:12]
+    response = client.post(
+        url_for("settings.index"),
+        data={
+            "username": alias_username,
+            "new_alias": "",  # html form
+        },
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert "Alias created successfully" in response.text
+
+    alias = db.session.scalars(db.select(Username).filter_by(_username=alias_username)).one()
+    assert not alias.is_primary
+    assert alias.user_id == user.id
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_add_alias_duplicate(client: FlaskClient, user: User) -> None:
+    response = client.post(
+        url_for("settings.index"),
+        data={
+            "username": user.primary_username.username,
+            "new_alias": "",  # html form
+        },
+        follow_redirects=True,
+    )
+    assert "This username is already taken." in response.text
+    assert db.session.scalar(db.func.count(Username.id)) == 1
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_alias_page_loads(client: FlaskClient, user: User, user_alias: Username) -> None:
+    response = client.get(
+        url_for("settings.alias", username_id=user_alias.id), follow_redirects=True
+    )
+    assert response.status_code == 200
+    assert f"Alias: @{user_alias.username}" in response.text
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_delete_account(client: FlaskClient, user: User, message: Message) -> None:
+    # save these because SqlAlchemy is too smart about nullifying them on deletion
+    user_id = user.id
+    username_id = user.primary_username.id
+    msg_id = message.id
+
+    resp = client.post(url_for("settings.delete_account"), follow_redirects=True)
+    assert resp.status_code == 200
+    assert "Your account and all related information have been deleted." in resp.text
+    assert db.session.scalars(db.select(User).filter_by(id=user_id)).one_or_none() is None
+    assert db.session.scalars(db.select(Username).filter_by(id=username_id)).one_or_none() is None
+    assert db.session.scalars(db.select(Message).filter_by(id=msg_id)).one_or_none() is None
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_alias_change_display_name(client: FlaskClient, user: User, user_alias: Username) -> None:
+    new_display_name = (user_alias.display_name or "") + "_NEW"
+
+    response = client.post(
+        url_for("settings.alias", username_id=user_alias.id),
+        data={
+            "display_name": new_display_name,
+            "update_display_name": "",  # html form
+        },
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert "Display name updated successfully" in response.text
+
+    updated_user = db.session.scalars(
+        db.select(Username).filter_by(_username=user_alias.username)
+    ).one()
+    assert updated_user.display_name == new_display_name
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_change_bio(client: FlaskClient, user: User) -> None:
+    data = {
+        "bio": str(uuid4()),
+        "update_bio": "",  # html form
+    }
+
+    for i in range(1, 5):
+        data[f"extra_field_label{i}"] = str(uuid4())
+        data[f"extra_field_value{i}"] = str(uuid4())
+
+    response = client.post(
+        url_for("settings.index"),
+        data=data,
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert "Bio and fields updated successfully" in response.text
+
+    updated_user = db.session.scalars(
+        db.select(Username).filter_by(_username=user.primary_username.username)
+    ).one()
+    assert updated_user.bio == data["bio"]
+
+    for i in range(1, 5):
+        label = f"extra_field_label{i}"
+        value = f"extra_field_value{i}"
+        assert getattr(updated_user, label) == data[label]
+        assert getattr(updated_user, value) == data[value]
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_alias_change_bio(client: FlaskClient, user: User, user_alias: Username) -> None:
+    data = {
+        "bio": str(uuid4()),
+        "update_bio": "",  # html form
+    }
+
+    for i in range(1, 5):
+        data[f"extra_field_label{i}"] = str(uuid4())
+        data[f"extra_field_value{i}"] = str(uuid4())
+
+    response = client.post(
+        url_for("settings.alias", username_id=user_alias.id),
+        data=data,
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert "Bio and fields updated successfully" in response.text
+
+    updated_user = db.session.scalars(
+        db.select(Username).filter_by(_username=user_alias.username)
+    ).one()
+    assert updated_user.bio == data["bio"]
+
+    for i in range(1, 5):
+        label = f"extra_field_label{i}"
+        value = f"extra_field_value{i}"
+        assert getattr(updated_user, label) == data[label]
+        assert getattr(updated_user, value) == data[value]
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_change_directory_visibility(client: FlaskClient, user: User) -> None:
+    original_visibility = user.primary_username.show_in_directory
+    resp = client.post(
+        url_for("settings.index"),
+        data={
+            "show_in_directory": not original_visibility,
+            "update_directory_visibility": "",  # html form
+        },
+        follow_redirects=True,
+    )
+    assert resp.status_code == 200
+    assert "Directory visibility updated successfully" in resp.text
+    assert db.session.scalar(
+        db.select(Username.show_in_directory).filter_by(id=user.primary_username.id)
+    )
+
+
+@pytest.mark.usefixtures("_authenticated_user")
+def test_alias_change_directory_visibility(
+    client: FlaskClient, user: User, user_alias: Username
+) -> None:
+    original_visibility = user_alias.show_in_directory
+    resp = client.post(
+        url_for("settings.alias", username_id=user_alias.id),
+        data={
+            "show_in_directory": not original_visibility,
+            "update_directory_visibility": "",  # html form
+        },
+        follow_redirects=True,
+    )
+    assert resp.status_code == 200
+    assert "Directory visibility updated successfully" in resp.text
+    assert db.session.scalar(db.select(Username.show_in_directory).filter_by(id=user_alias.id))