240 lines
8.9 KiB
Python
240 lines
8.9 KiB
Python
"""Tests for session sharing (create, access, revoke)."""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
class TestSessionSharing:
|
|
"""Test session share creation, access control, and revocation."""
|
|
|
|
async def _create_session(self, client, auth_headers, tree_id):
|
|
"""Helper: start a session for a tree."""
|
|
response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": tree_id},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()
|
|
|
|
async def test_create_public_share(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Create a public share link."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public", "share_name": "Customer link"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
share = response.json()
|
|
assert share["visibility"] == "public"
|
|
assert share["share_name"] == "Customer link"
|
|
assert share["is_active"] is True
|
|
assert share["view_count"] == 0
|
|
assert len(share["share_token"]) > 0
|
|
|
|
async def test_create_account_share(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Create an account-only share link."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "account"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 201
|
|
assert response.json()["visibility"] == "account"
|
|
|
|
async def test_access_public_share(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Access a public share without authentication."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
# Create share
|
|
share_resp = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public"},
|
|
headers=auth_headers
|
|
)
|
|
share_token = share_resp.json()["share_token"]
|
|
|
|
# Access without auth
|
|
response = await client.get(f"/api/v1/share/{share_token}")
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["session_id"] == session["id"]
|
|
assert data["visibility"] == "public"
|
|
assert data["path_taken"] is not None
|
|
|
|
async def test_access_revoked_share_returns_404(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Accessing a revoked share returns 404."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
# Create and revoke share
|
|
share_resp = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public"},
|
|
headers=auth_headers
|
|
)
|
|
share = share_resp.json()
|
|
|
|
await client.delete(
|
|
f"/api/v1/shares/{share['id']}",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Try to access revoked share
|
|
response = await client.get(f"/api/v1/share/{share['share_token']}")
|
|
assert response.status_code == 404
|
|
|
|
async def test_access_nonexistent_share_returns_404(self, client: AsyncClient):
|
|
"""Accessing a nonexistent share token returns 404."""
|
|
response = await client.get("/api/v1/share/nonexistent-token-12345")
|
|
assert response.status_code == 404
|
|
|
|
async def test_list_my_shares(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""List shares created by current user."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
# Create two shares
|
|
await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public", "share_name": "Link 1"},
|
|
headers=auth_headers
|
|
)
|
|
await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "account", "share_name": "Link 2"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
response = await client.get(
|
|
"/api/v1/shares/my-shares",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
shares = response.json()
|
|
assert len(shares) == 2
|
|
|
|
async def test_revoke_share(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Revoke a share link (soft delete)."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
share_resp = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public"},
|
|
headers=auth_headers
|
|
)
|
|
share = share_resp.json()
|
|
|
|
# Revoke
|
|
response = await client.delete(
|
|
f"/api/v1/shares/{share['id']}",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
# Verify it's gone from my-shares
|
|
list_resp = await client.get(
|
|
"/api/v1/shares/my-shares",
|
|
headers=auth_headers
|
|
)
|
|
shares = list_resp.json()
|
|
assert len(shares) == 0
|
|
|
|
async def test_multiple_shares_per_session(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""Multiple shares for the same session work independently."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
# Create public + account shares
|
|
resp1 = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public", "share_name": "For customer"},
|
|
headers=auth_headers
|
|
)
|
|
resp2 = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "account", "share_name": "For team"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert resp1.status_code == 201
|
|
assert resp2.status_code == 201
|
|
|
|
# Both tokens are different
|
|
assert resp1.json()["share_token"] != resp2.json()["share_token"]
|
|
|
|
# Both accessible
|
|
access1 = await client.get(f"/api/v1/share/{resp1.json()['share_token']}")
|
|
assert access1.status_code == 200
|
|
|
|
async def test_share_view_count_increments(self, client: AsyncClient, auth_headers, test_tree):
|
|
"""View count increments on each access."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
share_resp = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public"},
|
|
headers=auth_headers
|
|
)
|
|
token = share_resp.json()["share_token"]
|
|
|
|
# Access three times
|
|
await client.get(f"/api/v1/share/{token}")
|
|
await client.get(f"/api/v1/share/{token}")
|
|
await client.get(f"/api/v1/share/{token}")
|
|
|
|
# Check view count via my-shares
|
|
list_resp = await client.get(
|
|
"/api/v1/shares/my-shares",
|
|
headers=auth_headers
|
|
)
|
|
shares = list_resp.json()
|
|
assert shares[0]["view_count"] == 3
|
|
|
|
async def test_share_requires_session_ownership(self, client: AsyncClient, auth_headers, test_tree, test_db):
|
|
"""Non-owner cannot create a share for someone else's session."""
|
|
session = await self._create_session(client, auth_headers, test_tree["id"])
|
|
|
|
# Register a different user
|
|
await client.post("/api/v1/auth/register", json={
|
|
"email": "other@example.com",
|
|
"password": "OtherPassword123!",
|
|
"name": "Other User"
|
|
})
|
|
login_resp = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "other@example.com",
|
|
"password": "OtherPassword123!"
|
|
})
|
|
other_headers = {"Authorization": f"Bearer {login_resp.json()['access_token']}"}
|
|
|
|
# Try to share other user's session
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session['id']}/shares",
|
|
json={"visibility": "public"},
|
|
headers=other_headers
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_share_nonexistent_session(self, client: AsyncClient, auth_headers):
|
|
"""Creating a share for nonexistent session returns 404."""
|
|
response = await client.post(
|
|
"/api/v1/sessions/00000000-0000-0000-0000-000000000000/shares",
|
|
json={"visibility": "public"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_create_share_requires_auth(self, client: AsyncClient, test_tree):
|
|
"""Creating a share without auth returns 401."""
|
|
response = await client.post(
|
|
"/api/v1/sessions/00000000-0000-0000-0000-000000000000/shares",
|
|
json={"visibility": "public"}
|
|
)
|
|
assert response.status_code == 401
|