feat: add tests for whitelist
Build and Push Docker Image / build-and-push (push) Has been skipped Details

This commit is contained in:
itqop 2025-11-09 02:23:31 +03:00
parent c67c4f749c
commit 0cc81356a7
5 changed files with 726 additions and 18 deletions

22
poetry.lock generated
View File

@ -1,5 +1,24 @@
# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand.
[[package]]
name = "aiosqlite"
version = "0.21.0"
description = "asyncio bridge to the standard sqlite3 module"
optional = false
python-versions = ">=3.9"
groups = ["dev"]
files = [
{file = "aiosqlite-0.21.0-py3-none-any.whl", hash = "sha256:2549cf4057f95f53dcba16f2b64e8e2791d7e1adedb13197dd8ed77bb226d7d0"},
{file = "aiosqlite-0.21.0.tar.gz", hash = "sha256:131bb8056daa3bc875608c631c678cda73922a2d4ba8aec373b19f18c17e7aa3"},
]
[package.dependencies]
typing_extensions = ">=4.0"
[package.extras]
dev = ["attribution (==1.7.1)", "black (==24.3.0)", "build (>=1.2)", "coverage[toml] (==7.6.10)", "flake8 (==7.0.0)", "flake8-bugbear (==24.12.12)", "flit (==3.10.1)", "mypy (==1.14.1)", "ufmt (==2.5.1)", "usort (==1.0.8.post1)"]
docs = ["sphinx (==8.1.3)", "sphinx-mdinclude (==0.6.1)"]
[[package]]
name = "annotated-types"
version = "0.7.0"
@ -1137,7 +1156,6 @@ files = [
{file = "typing_extensions-4.15.0-py3-none-any.whl", hash = "sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548"},
{file = "typing_extensions-4.15.0.tar.gz", hash = "sha256:0cea48d173cc12fa28ecabc3b837ea3cf6f38c6d1136f85cbaaf598984861466"},
]
markers = {dev = "python_version < \"3.13\""}
[[package]]
name = "typing-inspection"
@ -1453,4 +1471,4 @@ dev = ["black (>=19.3b0) ; python_version >= \"3.6\"", "pytest (>=4.6.2)"]
[metadata]
lock-version = "2.1"
python-versions = "^3.11"
content-hash = "4f06b26593d87c90ac1c49788e813b728aba79cb609a3bcb4794d1537a4d7f58"
content-hash = "509a66a7f210f5887d96f5170a1ec4e33ebdca94b3b0acfeb76e00cac21794c0"

View File

@ -29,6 +29,7 @@ ruff = "^0.14.0"
black = "^24.10.0"
isort = "^6.0.1"
pytest-cov = "^5.0.0"
aiosqlite = "^0.21.0"
[tool.poetry.scripts]
hubgw = "hubgw.__main__:main"

View File

@ -1,7 +1,7 @@
"""Whitelist repository."""
from typing import List, Optional
from uuid import UUID
from uuid import UUID, uuid4
from sqlalchemy import and_, delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession
@ -18,7 +18,9 @@ class WhitelistRepository:
async def create(self, request: WhitelistAddRequest) -> WhitelistEntry:
"""Add player to whitelist."""
entry = WhitelistEntry(
id=uuid4(),
player_name=request.player_name,
added_by=request.added_by,
added_at=request.added_at,

View File

@ -1,7 +1,7 @@
"""Pytest configuration and fixtures."""
import pytest
import asyncio
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi.testclient import TestClient
@ -10,33 +10,25 @@ from hubgw.context import AppContext
from hubgw.core.config import AppSettings
@pytest.fixture(scope="session")
def event_loop():
"""Create event loop for async tests."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
@pytest_asyncio.fixture
async def test_db():
"""Create test database engine."""
settings = AppSettings()
settings.DB_DSN = "postgresql+asyncpg://test:test@localhost:5432/hubgw_test"
engine = create_async_engine(settings.DB_DSN)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
yield engine, session_factory
await engine.dispose()
@pytest.fixture
@pytest_asyncio.fixture
async def test_session(test_db):
"""Create test database session."""
engine, session_factory = test_db
async with session_factory() as session:
yield session

View File

@ -0,0 +1,695 @@
"""Integration tests for whitelist endpoints."""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch
from uuid import uuid4
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy import Column, String, Boolean, DateTime, event
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from hubgw.context import APP_CTX
from hubgw.main import create_app
from hubgw.models.base import Base
@pytest_asyncio.fixture(scope="function")
async def test_engine():
"""Create test database engine and setup tables."""
from sqlalchemy import Table, MetaData, Index
from uuid import uuid4 as _uuid4
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
)
metadata = MetaData()
whitelist_table = Table(
"hub_whitelist",
metadata,
Column("id", String(36), primary_key=True),
Column("player_name", String(255), nullable=False, unique=True, index=True),
Column("added_by", String(255), nullable=False),
Column("added_at", DateTime(timezone=True), nullable=False),
Column("expires_at", DateTime(timezone=True)),
Column("is_active", Boolean, default=True),
Column("reason", String(500)),
Column("created_at", DateTime(timezone=True)),
Column("updated_at", DateTime(timezone=True)),
)
async with engine.begin() as conn:
await conn.run_sync(metadata.create_all)
yield engine
await engine.dispose()
@pytest_asyncio.fixture(scope="function")
async def test_session(test_engine):
"""Create test database session."""
session_factory = async_sessionmaker(
bind=test_engine,
class_=AsyncSession,
expire_on_commit=False,
)
async with session_factory() as session:
yield session
@pytest_asyncio.fixture(scope="function")
async def client(test_engine):
"""Create async test client."""
app = create_app()
original_engine = APP_CTX._engine
original_factory = APP_CTX._session_factory
APP_CTX._engine = test_engine
APP_CTX._session_factory = async_sessionmaker(
bind=test_engine,
class_=AsyncSession,
expire_on_commit=False,
)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
yield ac
APP_CTX._engine = original_engine
APP_CTX._session_factory = original_factory
@pytest.fixture
def api_key():
"""Get API key from config."""
return APP_CTX.settings.security.api_key
class TestWhitelistAdd:
"""Tests for POST /api/v1/whitelist/add endpoint."""
@pytest.mark.asyncio
async def test_add_player_success(self, client, api_key):
"""Test successfully adding a new player to whitelist."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_lp.return_value.create_player = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"expires_at": (now + timedelta(days=30)).isoformat(),
"is_active": True,
"reason": "Testing",
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 201
data = response.json()
assert data["player_name"] == "TestPlayer"
assert data["added_by"] == "admin"
assert data["is_active"] is True
assert data["reason"] == "Testing"
assert "id" in data
@pytest.mark.asyncio
async def test_add_player_invalid_username(self, client, api_key):
"""Test adding player with invalid username format."""
now = datetime.now(timezone.utc)
payload = {
"player_name": "Invalid Name!", # Contains space and special char
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 422
assert "Invalid Minecraft username" in response.text
@pytest.mark.asyncio
async def test_add_player_username_too_short(self, client, api_key):
"""Test adding player with username too short."""
now = datetime.now(timezone.utc)
payload = {
"player_name": "AB", # Only 2 chars
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_add_player_expires_before_added(self, client, api_key):
"""Test adding player with expires_at before added_at."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"expires_at": (now - timedelta(minutes=1)).isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 422
assert "expires_at" in response.text and ("cannot be in the past" in response.text or "must be after added_at" in response.text)
@pytest.mark.asyncio
async def test_add_player_duplicate_active(self, client, api_key):
"""Test adding duplicate player when already active."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "DuplicatePlayer",
"added_by": "admin",
"added_at": now.isoformat(),
}
response1 = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response1.status_code == 201
response2 = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response2.status_code == 409
assert "already whitelisted" in response2.text
@pytest.mark.asyncio
async def test_add_player_reactivate_inactive(self, client, api_key, test_session):
"""Test reactivating an inactive player."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload1 = {
"player_name": "InactivePlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"is_active": False,
}
response1 = await client.post(
"/api/v1/whitelist/add",
json=payload1,
headers={"X-API-Key": api_key},
)
assert response1.status_code == 201
payload2 = {
"player_name": "InactivePlayer",
"added_by": "moderator",
"added_at": now.isoformat(),
"is_active": True,
"reason": "Reactivated",
}
response2 = await client.post(
"/api/v1/whitelist/add",
json=payload2,
headers={"X-API-Key": api_key},
)
assert response2.status_code == 201
data = response2.json()
assert data["added_by"] == "moderator"
assert data["is_active"] is True
assert data["reason"] == "Reactivated"
@pytest.mark.asyncio
async def test_add_player_unauthorized(self, client):
"""Test adding player without API key."""
now = datetime.now(timezone.utc)
payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post("/api/v1/whitelist/add", json=payload)
assert response.status_code == 422 # Missing header
@pytest.mark.asyncio
async def test_add_player_invalid_api_key(self, client):
"""Test adding player with invalid API key."""
now = datetime.now(timezone.utc)
payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": "wrong-key"},
)
assert response.status_code == 401
class TestWhitelistRemove:
"""Tests for POST /api/v1/whitelist/remove endpoint."""
@pytest.mark.asyncio
async def test_remove_player_success(self, client, api_key):
"""Test successfully removing a player."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
add_payload = {
"player_name": "PlayerToRemove",
"added_by": "admin",
"added_at": now.isoformat(),
}
await client.post(
"/api/v1/whitelist/add",
json=add_payload,
headers={"X-API-Key": api_key},
)
remove_payload = {"player_name": "PlayerToRemove"}
response = await client.post(
"/api/v1/whitelist/remove",
json=remove_payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 204
@pytest.mark.asyncio
async def test_remove_player_not_found(self, client, api_key):
"""Test removing non-existent player."""
payload = {"player_name": "NonExistent123"}
response = await client.post(
"/api/v1/whitelist/remove",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 404
assert "not found" in response.text
@pytest.mark.asyncio
async def test_remove_player_invalid_username(self, client, api_key):
"""Test removing player with invalid username."""
payload = {"player_name": "Invalid Name!"}
response = await client.post(
"/api/v1/whitelist/remove",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 422
class TestWhitelistCheck:
"""Tests for POST /api/v1/whitelist/check endpoint."""
@pytest.mark.asyncio
async def test_check_player_active(self, client, api_key):
"""Test checking active whitelisted player."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
add_payload = {
"player_name": "ActivePlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"is_active": True,
}
await client.post(
"/api/v1/whitelist/add",
json=add_payload,
headers={"X-API-Key": api_key},
)
check_payload = {"player_name": "ActivePlayer"}
response = await client.post(
"/api/v1/whitelist/check",
json=check_payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["is_whitelisted"] is True
@pytest.mark.asyncio
async def test_check_player_inactive(self, client, api_key):
"""Test checking inactive player."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
add_payload = {
"player_name": "InactivePlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"is_active": False,
}
await client.post(
"/api/v1/whitelist/add",
json=add_payload,
headers={"X-API-Key": api_key},
)
check_payload = {"player_name": "InactivePlayer"}
response = await client.post(
"/api/v1/whitelist/check",
json=check_payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["is_whitelisted"] is False
@pytest.mark.asyncio
async def test_check_player_not_found(self, client, api_key):
"""Test checking non-existent player."""
payload = {"player_name": "NonExistent123"}
response = await client.post(
"/api/v1/whitelist/check",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["is_whitelisted"] is False
class TestWhitelistList:
"""Tests for GET /api/v1/whitelist/ endpoint."""
@pytest.mark.asyncio
async def test_list_empty(self, client, api_key):
"""Test listing when whitelist is empty."""
response = await client.get(
"/api/v1/whitelist/",
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["entries"] == []
assert data["total"] == 0
@pytest.mark.asyncio
async def test_list_multiple_players(self, client, api_key):
"""Test listing multiple players."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
for i, name in enumerate(["Player1", "Player2", "Player3"]):
payload = {
"player_name": name,
"added_by": "admin",
"added_at": (now + timedelta(seconds=i)).isoformat(),
"is_active": i % 2 == 0,
}
await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
response = await client.get(
"/api/v1/whitelist/",
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert len(data["entries"]) == 3
assert data["total"] == 3
assert data["entries"][0]["player_name"] == "Player3"
assert data["entries"][1]["player_name"] == "Player2"
assert data["entries"][2]["player_name"] == "Player1"
class TestWhitelistCount:
"""Tests for GET /api/v1/whitelist/count endpoint."""
@pytest.mark.asyncio
async def test_count_empty(self, client, api_key):
"""Test count when whitelist is empty."""
response = await client.get(
"/api/v1/whitelist/count",
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
@pytest.mark.asyncio
async def test_count_multiple(self, client, api_key):
"""Test count with multiple entries."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
for i in range(5):
payload = {
"player_name": f"Player{i}",
"added_by": "admin",
"added_at": now.isoformat(),
}
await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
response = await client.get(
"/api/v1/whitelist/count",
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 5
@pytest.mark.asyncio
async def test_count_includes_inactive(self, client, api_key):
"""Test that count includes both active and inactive entries."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
for i in range(3):
payload = {
"player_name": f"Player{i}",
"added_by": "admin",
"added_at": now.isoformat(),
"is_active": i % 2 == 0,
}
await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
response = await client.get(
"/api/v1/whitelist/count",
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 3
class TestWhitelistEdgeCases:
"""Edge case tests for whitelist endpoints."""
@pytest.mark.asyncio
async def test_add_player_with_whitespace_reason(self, client, api_key):
"""Test adding player with whitespace-only reason."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
"reason": " ",
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 201
data = response.json()
assert data["reason"] is None
@pytest.mark.asyncio
async def test_add_player_max_length_username(self, client, api_key):
"""Test adding player with maximum length username."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "A" * 16,
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_add_player_min_length_username(self, client, api_key):
"""Test adding player with minimum length username."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
payload = {
"player_name": "ABC",
"added_by": "admin",
"added_at": now.isoformat(),
}
response = await client.post(
"/api/v1/whitelist/add",
json=payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_check_player_case_sensitive(self, client, api_key):
"""Test that player name check is case-sensitive."""
with patch("hubgw.api.deps.get_luckperms_service") as mock_lp, \
patch("hubgw.api.deps.get_user_service") as mock_us:
mock_lp.return_value = AsyncMock()
mock_us.return_value = AsyncMock()
now = datetime.now(timezone.utc)
add_payload = {
"player_name": "TestPlayer",
"added_by": "admin",
"added_at": now.isoformat(),
}
await client.post(
"/api/v1/whitelist/add",
json=add_payload,
headers={"X-API-Key": api_key},
)
check_payload = {"player_name": "testplayer"}
response = await client.post(
"/api/v1/whitelist/check",
json=check_payload,
headers={"X-API-Key": api_key},
)
assert response.status_code == 200
data = response.json()
assert data["is_whitelisted"] is False