From 0cc81356a76c813e91f3b5e5b19409df895d4c17 Mon Sep 17 00:00:00 2001 From: itqop Date: Sun, 9 Nov 2025 02:23:31 +0300 Subject: [PATCH] feat: add tests for whitelist --- poetry.lock | 22 +- pyproject.toml | 1 + src/hubgw/repositories/whitelist_repo.py | 4 +- tests/conftest.py | 22 +- tests/integration/test_whitelist.py | 695 +++++++++++++++++++++++ 5 files changed, 726 insertions(+), 18 deletions(-) create mode 100644 tests/integration/test_whitelist.py diff --git a/poetry.lock b/poetry.lock index 00ee6a0..96e64a6 100644 --- a/poetry.lock +++ b/poetry.lock @@ -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" diff --git a/pyproject.toml b/pyproject.toml index ca30863..61f1d60 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/hubgw/repositories/whitelist_repo.py b/src/hubgw/repositories/whitelist_repo.py index d164d1e..36dd180 100644 --- a/src/hubgw/repositories/whitelist_repo.py +++ b/src/hubgw/repositories/whitelist_repo.py @@ -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, diff --git a/tests/conftest.py b/tests/conftest.py index 3028dd1..900a10d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 diff --git a/tests/integration/test_whitelist.py b/tests/integration/test_whitelist.py new file mode 100644 index 0000000..59270c2 --- /dev/null +++ b/tests/integration/test_whitelist.py @@ -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