215 lines
7.1 KiB
Python
215 lines
7.1 KiB
Python
"""
|
|
Tests for api_service account CRUD endpoints.
|
|
Validates tasks 4.1 and 4.2.
|
|
"""
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from unittest.mock import patch
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from shared.models import get_db
|
|
from tests.conftest import TEST_ENGINE, TestSessionLocal, Base, FakeRedis
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client():
|
|
"""
|
|
Provide an httpx AsyncClient wired to the api_service app,
|
|
with DB overridden to test SQLite and a fake Redis for auth tokens.
|
|
"""
|
|
fake_redis = FakeRedis()
|
|
|
|
async with TEST_ENGINE.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Import apps after DB is ready
|
|
from api_service.app.main import app as api_app
|
|
from auth_service.app.main import app as auth_app
|
|
|
|
async def override_get_db():
|
|
async with TestSessionLocal() as session:
|
|
yield session
|
|
|
|
async def _fake_get_redis():
|
|
return fake_redis
|
|
|
|
api_app.dependency_overrides[get_db] = override_get_db
|
|
auth_app.dependency_overrides[get_db] = override_get_db
|
|
|
|
with patch(
|
|
"auth_service.app.utils.security.get_redis",
|
|
new=_fake_get_redis,
|
|
):
|
|
# We need both clients: auth for getting tokens, api for account ops
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=auth_app), base_url="http://auth"
|
|
) as auth_client, AsyncClient(
|
|
transport=ASGITransport(app=api_app), base_url="http://api"
|
|
) as api_client:
|
|
yield auth_client, api_client
|
|
|
|
api_app.dependency_overrides.clear()
|
|
auth_app.dependency_overrides.clear()
|
|
|
|
|
|
async def _register_and_login(auth_client: AsyncClient, suffix: str = "1") -> str:
|
|
"""Helper: register a user and return an access token."""
|
|
reg = await auth_client.post("/auth/register", json={
|
|
"username": f"acctuser{suffix}",
|
|
"email": f"acct{suffix}@example.com",
|
|
"password": "Str0ng!Pass1",
|
|
})
|
|
assert reg.status_code == 201, f"Register failed: {reg.json()}"
|
|
resp = await auth_client.post("/auth/login", json={
|
|
"email": f"acct{suffix}@example.com",
|
|
"password": "Str0ng!Pass1",
|
|
})
|
|
login_body = resp.json()
|
|
assert resp.status_code == 200, f"Login failed: {login_body}"
|
|
# Handle both wrapped (success_response) and unwrapped token formats
|
|
if "data" in login_body:
|
|
return login_body["data"]["access_token"]
|
|
return login_body["access_token"]
|
|
|
|
|
|
def _auth_header(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ===================== Basic structure tests =====================
|
|
|
|
|
|
class TestAPIServiceBase:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health(self, client):
|
|
_, api = client
|
|
resp = await api.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["success"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root(self, client):
|
|
_, api = client
|
|
resp = await api.get("/")
|
|
assert resp.status_code == 200
|
|
assert "API Service" in resp.json()["data"]["service"]
|
|
|
|
|
|
# ===================== Account CRUD tests =====================
|
|
|
|
|
|
class TestAccountCRUD:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_account(self, client):
|
|
auth, api = client
|
|
token = await _register_and_login(auth)
|
|
|
|
resp = await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": "12345",
|
|
"cookie": "SUB=abc; SUBP=xyz;",
|
|
"remark": "test account",
|
|
}, headers=_auth_header(token))
|
|
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["success"] is True
|
|
assert body["data"]["weibo_user_id"] == "12345"
|
|
assert body["data"]["status"] == "pending"
|
|
assert body["data"]["remark"] == "test account"
|
|
# Cookie plaintext must NOT appear in response
|
|
assert "SUB=abc" not in str(body)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_accounts(self, client):
|
|
auth, api = client
|
|
token = await _register_and_login(auth, "list")
|
|
|
|
# Create two accounts
|
|
for i in range(2):
|
|
await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": f"uid{i}",
|
|
"cookie": f"cookie{i}",
|
|
}, headers=_auth_header(token))
|
|
|
|
resp = await api.get("/api/v1/accounts", headers=_auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert len(data) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_account_detail(self, client):
|
|
auth, api = client
|
|
token = await _register_and_login(auth, "detail")
|
|
|
|
create_resp = await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": "99",
|
|
"cookie": "c=1",
|
|
"remark": "my remark",
|
|
}, headers=_auth_header(token))
|
|
account_id = create_resp.json()["data"]["id"]
|
|
|
|
resp = await api.get(f"/api/v1/accounts/{account_id}", headers=_auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"]["remark"] == "my remark"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_account_remark(self, client):
|
|
auth, api = client
|
|
token = await _register_and_login(auth, "upd")
|
|
|
|
create_resp = await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": "55",
|
|
"cookie": "c=old",
|
|
}, headers=_auth_header(token))
|
|
account_id = create_resp.json()["data"]["id"]
|
|
|
|
resp = await api.put(f"/api/v1/accounts/{account_id}", json={
|
|
"remark": "updated remark",
|
|
}, headers=_auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"]["remark"] == "updated remark"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_account(self, client):
|
|
auth, api = client
|
|
token = await _register_and_login(auth, "del")
|
|
|
|
create_resp = await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": "77",
|
|
"cookie": "c=del",
|
|
}, headers=_auth_header(token))
|
|
account_id = create_resp.json()["data"]["id"]
|
|
|
|
resp = await api.delete(f"/api/v1/accounts/{account_id}", headers=_auth_header(token))
|
|
assert resp.status_code == 200
|
|
|
|
# Verify it's gone
|
|
resp2 = await api.get(f"/api/v1/accounts/{account_id}", headers=_auth_header(token))
|
|
assert resp2.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_access_other_users_account_forbidden(self, client):
|
|
auth, api = client
|
|
token_a = await _register_and_login(auth, "ownerA")
|
|
token_b = await _register_and_login(auth, "ownerB")
|
|
|
|
# User A creates an account
|
|
create_resp = await api.post("/api/v1/accounts", json={
|
|
"weibo_user_id": "111",
|
|
"cookie": "c=a",
|
|
}, headers=_auth_header(token_a))
|
|
account_id = create_resp.json()["data"]["id"]
|
|
|
|
# User B tries to access it
|
|
resp = await api.get(f"/api/v1/accounts/{account_id}", headers=_auth_header(token_b))
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthenticated_request_rejected(self, client):
|
|
_, api = client
|
|
resp = await api.get("/api/v1/accounts")
|
|
assert resp.status_code in (401, 403)
|