Link a pending portal member to the local user with their email

A portal identity whose email a local user already has is still refused
on arrival. POST /cloud/members/{portal_sub}/link lets a superuser map it
onto that user, never a superuser's, after the portal confirms the
membership. GET /cloud/members marks such members linkable, and Admin →
Users offers "Link portal account" for them, saying that a linked user is
deleted with a removal at the portal or a Disconnect.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/tests/test_cloud.py

@@ -809,6 +809,79 @@ def test_admin_lists_and_revokes_portal_members(
     db.commit()
 
 
+def test_a_superuser_links_a_pending_member_to_their_local_user(
+    client: TestClient,
+    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
+    portal_key: rsa.RSAPrivateKey,
+    superuser_token_headers: dict[str, str],
+    normal_user_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    """Arriving alone they are refused; linked, they arrive as that user."""
+    local = create_random_user(db)
+    boss = create_random_user(db)
+    boss.is_superuser = True
+    db.add(boss)
+    db.commit()
+    sub, boss_sub = str(uuid.uuid4()), str(uuid.uuid4())
+
+    def member(user_id: str, email: str) -> Mock:
+        return Mock(
+            status_code=200,
+            json=Mock(return_value={"user_id": user_id, "email": email}),
+        )
+
+    def link(user_id: str, reply: Mock, headers: dict[str, str]) -> httpx.Response:
+        with patch("fluksio.cloud.enroll.httpx.get", return_value=reply):
+            return client.post(
+                f"{settings.API_V1_STR}/cloud/members/{user_id}/link", headers=headers
+            )
+
+    # Never adopted on their own: the local row might be anyone's.
+    with patch("fluksio.cloud.enroll.httpx.get", return_value=member(sub, local.email)):
+        assert user_from_token(db, portal_token(portal_key, subject=sub)) is None
+
+    listed = Mock(
+        status_code=200,
+        json=Mock(
+            return_value={
+                "data": [
+                    {"user_id": sub, "email": local.email},
+                    {"user_id": boss_sub, "email": boss.email},
+                ]
+            }
+        ),
+    )
+    with patch("fluksio.api.routes.cloud.httpx.get", return_value=listed):
+        members = client.get(
+            f"{settings.API_V1_STR}/cloud/members", headers=superuser_token_headers
+        ).json()
+    assert {row["user_id"]: row["linkable"] for row in members} == {
+        sub: True,
+        boss_sub: False,
+    }
+
+    su = superuser_token_headers
+    assert link(boss_sub, member(boss_sub, boss.email), su).status_code == 403
+    assert link(sub, Mock(status_code=404), su).status_code == 404
+    assert link(sub, member(sub, "nobody-here@example.com"), su).status_code == 404
+    assert (
+        link(sub, member(sub, local.email), normal_user_token_headers).status_code
+        == 403
+    )
+
+    linked = link(sub, member(sub, local.email), su)
+    assert linked.status_code == 200, linked.text
+    assert linked.json()["portal_sub"] == sub
+    assert link(sub, member(sub, local.email), su).status_code == 409
+    db.expire_all()
+    resolved = user_from_token(db, portal_token(portal_key, subject=sub))
+    assert resolved is not None and resolved.id == local.id
+    db.delete(db.get(User, local.id))
+    db.delete(db.get(User, boss.id))
+    db.commit()
+
+
 def test_the_portal_keys_follow_the_tunnel(
     enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
     portal_key: rsa.RSAPrivateKey,