mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-04 20:13:04 +00:00
Alex
574f96341e
The backend import package is now docsgpt, the name it will carry on PyPI; application was far too generic to install into anyone's site-packages. git mv plus a mechanical rewrite of every import, dotted string and path reference: 734 Python files, the compose files, Dockerfile, workflows, docs, setup scripts, devcontainer, k8s manifests, vscode config, pytest and coverage config, .gitignore. Behaviour is unchanged. Kept for one release: - A top-level application package whose meta-path finder resolves application.x.y to the already-imported docsgpt.x.y object, so old imports and entry points (celery -A application.app.celery, uvicorn application.asgi:asgi_app) keep working with a FutureWarning. - Celery registers every application.* task name as an alias of its docsgpt.* task on start-up, so messages queued by the previous release still run. The redbeat key prefix moves to redbeat:docsgpt:v2: so schedule entries the previous release wrote are left unread instead of firing twice. The backend image builds from the repository root (docker build -f docsgpt/Dockerfile .) so it can ship the alias package; a root .dockerignore allow-lists docsgpt/ and application/ and keeps caches, local data, .env files, the sample index files and the Dockerfile out. Compose and the image workflows point at the new context.
571 lines
21 KiB
Python
571 lines
21 KiB
Python
"""Endpoint authorization tests for the teams API.
|
|
|
|
These drive the real app.py chokepoint + team_authz plane (only handle_auth /
|
|
resolve_roles and the leaf repos are mocked), so they catch regressions in the
|
|
authorization wiring: who can read/manage a team, that team_id comes from the
|
|
URL path (never the body), and that sharing requires resource ownership. Data
|
|
correctness lives in tests/storage/db/repositories/test_teams.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from contextlib import contextmanager
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
from docsgpt.app import app as flask_app
|
|
|
|
flask_app.config["TESTING"] = True
|
|
return flask_app.test_client()
|
|
|
|
|
|
@contextmanager
|
|
def _cm(value):
|
|
yield value
|
|
|
|
|
|
def _auth(sub="u1", roles=("user",), team_role=None):
|
|
"""Patch the chokepoint auth + the team_authz membership resolution.
|
|
|
|
``team_role`` is what TeamMembersRepository.role_for returns inside
|
|
team_authz (None = not a member).
|
|
"""
|
|
members = Mock()
|
|
members.role_for.return_value = team_role
|
|
return [
|
|
patch("docsgpt.app.handle_auth", return_value={"sub": sub}),
|
|
patch("docsgpt.app.resolve_roles", return_value=list(roles)),
|
|
patch("docsgpt.api.user.team_authz.db_readonly", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.team_authz.TeamMembersRepository", return_value=members),
|
|
]
|
|
|
|
|
|
def _apply(patches):
|
|
for p in patches:
|
|
p.start()
|
|
|
|
|
|
def _stop(patches):
|
|
for p in reversed(patches):
|
|
p.stop()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestTeamCreation:
|
|
def test_unauthenticated_401(self, client):
|
|
with patch("docsgpt.app.handle_auth", return_value=None):
|
|
resp = client.post("/api/teams", json={"name": "Acme"})
|
|
assert resp.status_code == 401
|
|
|
|
def test_self_serve_create_returns_201(self, client):
|
|
teams_repo = Mock()
|
|
teams_repo.slug_exists.return_value = False
|
|
teams_repo.create.return_value = {"id": "t1", "name": "Acme", "slug": "acme"}
|
|
members_repo = Mock()
|
|
patches = _auth(sub="alice") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post("/api/teams", json={"name": "Acme"})
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 201
|
|
# Creator is added as team_admin in the same flow.
|
|
args, kwargs = members_repo.add_member.call_args
|
|
assert kwargs.get("role") == "team_admin" or "team_admin" in args
|
|
|
|
def test_missing_name_400(self, client):
|
|
patches = _auth(sub="alice")
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post("/api/teams", json={})
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestTeamAccessControl:
|
|
def test_detail_requires_membership(self, client):
|
|
patches = _auth(sub="stranger", team_role=None)
|
|
_apply(patches)
|
|
try:
|
|
resp = client.get("/api/teams/team-1")
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 403
|
|
|
|
def test_member_can_read_detail(self, client):
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "name": "Acme", "owner_id": "alice"}
|
|
members_repo = Mock()
|
|
members_repo.list_members.return_value = []
|
|
members_repo.role_for.return_value = "team_member"
|
|
patches = _auth(sub="bob", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.get("/api/teams/team-1")
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
|
|
def test_team_id_taken_from_path_not_body(self, client):
|
|
# Caller is NOT a member of team-1; a forged team_id in the body must be
|
|
# ignored — the decorator reads team_id from the URL path only.
|
|
patches = _auth(sub="stranger", team_role=None)
|
|
_apply(patches)
|
|
try:
|
|
resp = client.get("/api/teams/team-1", json={"team_id": "team-i-own"})
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 403
|
|
|
|
def test_add_member_requires_team_admin(self, client):
|
|
patches = _auth(sub="bob", team_role="team_member")
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members", json={"user_id": "x", "role": "team_member"}
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 403
|
|
|
|
def test_team_admin_can_add_member(self, client):
|
|
members_repo = Mock()
|
|
patches = _auth(sub="alice", team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members", json={"user_id": "carol", "role": "team_member"}
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
members_repo.set_manual_role.assert_called_once()
|
|
|
|
def test_add_member_by_email_resolves_to_sub(self, client):
|
|
members_repo = Mock()
|
|
users_repo = Mock()
|
|
users_repo.find_by_email.return_value = {"user_id": "resolved-sub"}
|
|
patches = _auth(sub="alice", team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.UsersRepository",
|
|
return_value=users_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members",
|
|
json={"email": "bob@example.com", "role": "team_member"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
args, kwargs = members_repo.set_manual_role.call_args
|
|
assert "resolved-sub" in args # resolved sub used, not the email
|
|
|
|
def test_add_member_email_not_found(self, client):
|
|
users_repo = Mock()
|
|
users_repo.find_by_email.return_value = None
|
|
patches = _auth(sub="alice", team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.UsersRepository",
|
|
return_value=users_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members",
|
|
json={"email": "ghost@example.com", "role": "team_member"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 404
|
|
|
|
def test_delete_team_non_owner_forbidden(self, client):
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "owner_id": "alice"}
|
|
patches = _auth(sub="bob", roles=("user",), team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.delete("/api/teams/team-1")
|
|
finally:
|
|
_stop(patches)
|
|
# team_admin is not the owner — only the owner (or a global admin) deletes.
|
|
assert resp.status_code == 403
|
|
teams_repo.delete.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSharingAuthz:
|
|
def test_share_requires_ownership(self, client):
|
|
patches = _auth(sub="bob", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=False),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "11111111-1111-1111-1111-111111111111",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 403
|
|
|
|
def test_owner_can_share(self, client):
|
|
grants_repo = Mock()
|
|
grants_repo.grant.return_value = {"id": "g1", "access_level": "viewer"}
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamResourceGrantsRepository",
|
|
return_value=grants_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "22222222-2222-2222-2222-222222222222",
|
|
"access_level": "editor",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 201
|
|
_, kwargs = grants_repo.grant.call_args
|
|
assert kwargs.get("access_level") == "editor"
|
|
|
|
def test_invalid_resource_type_rejected(self, client):
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={"resource_type": "secret_thing", "resource_id": "a1"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 400
|
|
|
|
def test_share_with_member_validates_membership(self, client):
|
|
# target_user_id must be a member of the team.
|
|
members_repo = Mock()
|
|
members_repo.is_member.return_value = False
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "22222222-2222-2222-2222-222222222222",
|
|
"target_user_id": "stranger",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 400
|
|
|
|
def test_share_with_valid_member(self, client):
|
|
members_repo = Mock()
|
|
members_repo.is_member.return_value = True
|
|
grants_repo = Mock()
|
|
grants_repo.grant.return_value = {"id": "g1", "target_user_id": "bob"}
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamResourceGrantsRepository",
|
|
return_value=grants_repo,
|
|
),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "22222222-2222-2222-2222-222222222222",
|
|
"target_user_id": "bob",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 201
|
|
_, kwargs = grants_repo.grant.call_args
|
|
assert kwargs.get("target_user_id") == "bob"
|
|
|
|
def test_non_uuid_resource_id_rejected(self, client):
|
|
# Team grants are UUID-only (post-cutover); a legacy/non-UUID id must be
|
|
# rejected cleanly, not cast-and-poison the txn into a generic error.
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={"resource_type": "agent", "resource_id": "legacy-mongo-id"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAdminOversight:
|
|
def test_non_admin_forbidden(self, client):
|
|
patches = _auth(sub="bob", roles=("user",))
|
|
_apply(patches)
|
|
try:
|
|
resp = client.get("/api/admin/teams")
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 403
|
|
|
|
def test_global_admin_lists_all(self, client):
|
|
teams_repo = Mock()
|
|
teams_repo.list_all.return_value = [{"id": "t1", "member_count": 3}]
|
|
patches = _auth(sub="root", roles=("admin", "user")) + [
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.get("/api/admin/teams")
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
assert json.loads(resp.data)["teams"][0]["member_count"] == 3
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestTeamNotifications:
|
|
"""Best-effort, post-commit toasts via publish_user_event."""
|
|
|
|
def test_add_member_notifies_new_member(self, client):
|
|
members_repo = Mock()
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "name": "Acme"}
|
|
publish = Mock()
|
|
patches = _auth(sub="alice", team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamsRepository",
|
|
return_value=teams_repo,
|
|
),
|
|
patch("docsgpt.api.user.teams.routes.publish_user_event", publish),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members",
|
|
json={"user_id": "carol", "role": "team_member"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
publish.assert_called_once()
|
|
args, _ = publish.call_args
|
|
assert args[0] == "carol"
|
|
assert args[1] == "team.member_added"
|
|
assert args[2]["team_name"] == "Acme"
|
|
assert args[2]["role"] == "team_member"
|
|
assert args[2]["added_by"] == "alice"
|
|
|
|
def test_add_member_skips_self(self, client):
|
|
members_repo = Mock()
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "name": "Acme"}
|
|
publish = Mock()
|
|
patches = _auth(sub="alice", team_role="team_admin") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamsRepository",
|
|
return_value=teams_repo,
|
|
),
|
|
patch("docsgpt.api.user.teams.routes.publish_user_event", publish),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/members",
|
|
json={"user_id": "alice", "role": "team_admin"},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 200
|
|
publish.assert_not_called()
|
|
|
|
def test_whole_team_share_notifies_all_but_sharer(self, client):
|
|
grants_repo = Mock()
|
|
grants_repo.grant.return_value = {"id": "g1", "access_level": "editor"}
|
|
members_repo = Mock()
|
|
members_repo.list_members.return_value = [
|
|
{"user_id": "alice"},
|
|
{"user_id": "bob"},
|
|
{"user_id": "carol"},
|
|
]
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "name": "Acme"}
|
|
publish = Mock()
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamResourceGrantsRepository",
|
|
return_value=grants_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamsRepository",
|
|
return_value=teams_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes._resource_display_name",
|
|
return_value="My Agent",
|
|
),
|
|
patch("docsgpt.api.user.teams.routes.publish_user_event", publish),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "22222222-2222-2222-2222-222222222222",
|
|
"access_level": "editor",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 201
|
|
recipients = {call.args[0] for call in publish.call_args_list}
|
|
assert recipients == {"bob", "carol"} # never the sharer (alice)
|
|
first = publish.call_args_list[0].args
|
|
assert first[1] == "resource.shared"
|
|
assert first[2]["resource_name"] == "My Agent"
|
|
assert first[2]["access_level"] == "editor"
|
|
assert first[2]["shared_by"] == "alice"
|
|
|
|
def test_per_member_share_notifies_only_target(self, client):
|
|
grants_repo = Mock()
|
|
grants_repo.grant.return_value = {"id": "g1", "access_level": "viewer"}
|
|
members_repo = Mock()
|
|
members_repo.is_member.return_value = True
|
|
teams_repo = Mock()
|
|
teams_repo.get.return_value = {"id": "team-1", "name": "Acme"}
|
|
publish = Mock()
|
|
patches = _auth(sub="alice", team_role="team_member") + [
|
|
patch("docsgpt.api.user.teams.routes.db_session", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
|
|
patch("docsgpt.api.user.teams.routes.owns_resource", return_value=True),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamResourceGrantsRepository",
|
|
return_value=grants_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamMembersRepository",
|
|
return_value=members_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes.TeamsRepository",
|
|
return_value=teams_repo,
|
|
),
|
|
patch(
|
|
"docsgpt.api.user.teams.routes._resource_display_name",
|
|
return_value="My Agent",
|
|
),
|
|
patch("docsgpt.api.user.teams.routes.publish_user_event", publish),
|
|
]
|
|
_apply(patches)
|
|
try:
|
|
resp = client.post(
|
|
"/api/teams/team-1/grants",
|
|
json={
|
|
"resource_type": "agent",
|
|
"resource_id": "22222222-2222-2222-2222-222222222222",
|
|
"access_level": "viewer",
|
|
"target_user_id": "bob",
|
|
},
|
|
)
|
|
finally:
|
|
_stop(patches)
|
|
assert resp.status_code == 201
|
|
publish.assert_called_once()
|
|
args, _ = publish.call_args
|
|
assert args[0] == "bob"
|
|
assert args[1] == "resource.shared"
|