fix

.github/workflows/pre-release.yml

@@ -13,6 +13,8 @@ jobs:
     # ──────────────────── Docker Pre-Release Image ────────────────────
     pre-release:
         runs-on: ubuntu-latest
+        # Only run for prerelease releases
+        if: ${{ github.event.release.prerelease }}
         permissions:
             contents: read
             packages: write

.github/workflows/release.yml

@@ -13,6 +13,8 @@ jobs:
     # ──────────────────── Docker Release Image ────────────────────
     release:
         runs-on: ubuntu-latest
+        # Only run for non-prerelease releases
+        if: ${{ !github.event.release.prerelease }}
         permissions:
             contents: read
             packages: write

tests/README.md

@@ -0,0 +1,201 @@
+# Wizarr Testing Documentation
+
+This directory contains comprehensive tests for the Wizarr invitation system.
+
+## Test Structure
+
+### Core Test Files
+
+- **`test_invitation_comprehensive.py`** - Complete invitation workflow tests with API simulation
+- **`test_invitation_performance.py`** - Performance and load testing
+- **`e2e/test_invitation_e2e.py`** - End-to-end tests using Playwright
+- **`mocks/media_server_mocks.py`** - Mock implementations for media server APIs
+
+### Mock System
+
+The mock system simulates various media server APIs (Jellyfin, Plex, Audiobookshelf) without requiring actual server instances:
+
+```python
+from tests.mocks import create_mock_client, setup_mock_servers
+
+# Setup
+setup_mock_servers()
+
+# Create mock client
+mock_client = create_mock_client("jellyfin", server_id=1)
+
+# Simulate failures
+simulate_server_failure()
+simulate_user_creation_failure(["problematic_username"])
+```
+
+## Test Categories
+
+### 1. Unit Tests
+- **Invitation validation** - Test `is_invite_valid()` logic
+- **Expiry calculations** - Test duration and expiry date handling
+- **Library assignment** - Test invitation-specific library restrictions
+
+### 2. Integration Tests  
+- **Single-server invitations** - Test complete workflow for one server
+- **Multi-server invitations** - Test cross-server invitation processing
+- **Error handling** - Test rollback and error recovery
+- **Identity linking** - Test user identity linking across servers
+
+### 3. End-to-End Tests
+- **Complete user journey** - From invitation link to account creation
+- **Form validation** - Test UI validation and error handling
+- **Multi-server UI flow** - Test UI for complex invitation scenarios
+- **Accessibility** - Test keyboard navigation and screen reader support
+
+### 4. Performance Tests
+- **Single invitation timing** - Ensure <1s processing time
+- **Concurrent processing** - Test multiple simultaneous invitations
+- **Database performance** - Test with large datasets (500+ invitations)
+- **Memory usage** - Ensure no memory leaks during processing
+
+## Running Tests
+
+### All Tests
+```bash
+uv run pytest tests/ -v
+```
+
+### Specific Test Categories
+```bash
+# Unit and integration tests
+uv run pytest tests/test_invitation_comprehensive.py -v
+
+# Performance tests
+uv run pytest tests/test_invitation_performance.py -v
+
+# End-to-end tests
+uv run pytest tests/e2e/test_invitation_e2e.py -v
+```
+
+### With Coverage
+```bash
+uv run pytest tests/ --cov=app/services/invitation_manager --cov=app/services/invites --cov-report=html
+```
+
+## Test Scenarios Covered
+
+### Happy Path Scenarios
+- ✅ Single server invitation (Jellyfin, Plex, Audiobookshelf)
+- ✅ Multi-server invitation with all servers succeeding
+- ✅ Unlimited invitation reuse
+- ✅ Library-specific invitations
+- ✅ User expiry date calculation
+
+### Error Scenarios
+- ✅ Expired invitations
+- ✅ Already used limited invitations
+- ✅ Invalid invitation codes
+- ✅ Server connection failures
+- ✅ User creation failures
+- ✅ Multi-server partial failures
+- ✅ Complete multi-server failures
+
+### Edge Cases
+- ✅ Password mismatch validation
+- ✅ Email format validation
+- ✅ Concurrent invitation processing
+- ✅ Database transaction rollbacks
+- ✅ Identity linking for same invitation code
+
+### Performance Cases
+- ✅ Single invitation processing time (<1s)
+- ✅ 10 concurrent invitations (<5s)
+- ✅ Multi-server invitation (<3s)
+- ✅ Large dataset queries (500+ records)
+- ✅ Memory usage under load
+
+## Mock API Behavior
+
+The mock system simulates realistic API behavior:
+
+### Jellyfin Mock
+```python
+# Success response
+{
+    "Id": "user-uuid",
+    "Name": "username", 
+    "Primary": "email@example.com",
+    "Policy": {"EnableDownloads": True}
+}
+
+# Library assignment
+client._set_specific_folders(user_id, ["lib1", "lib2"])
+
+# Error simulation
+simulate_user_creation_failure(["problematic_user"])
+```
+
+### Plex Mock
+```python
+# Uses email as primary identifier
+# Automatically assigns all libraries
+# Supports OAuth flow simulation
+```
+
+### State Management
+```python
+from tests.mocks import get_mock_state
+
+# Check created users
+state = get_mock_state()
+print(f"Created {len(state.users)} users")
+
+# Reset for clean tests
+state.reset()
+```
+
+## Configuration
+
+### Test Database
+Tests use SQLite in-memory database by default. Configure via `conftest.py`.
+
+### Mock Server URLs
+- Jellyfin: `http://localhost:8096`  
+- Plex: `http://localhost:32400`
+- Audiobookshelf: `http://localhost:13378`
+
+### Test Data
+Mock servers come with predefined libraries:
+- `lib1` - Movies
+- `lib2` - TV Shows  
+- `lib3` - Music
+- `movies_4k` - Movies 4K
+- `anime` - Anime
+- `audiobooks` - Audiobooks
+
+## Best Practices
+
+### Writing New Tests
+1. Use `setup_mock_servers()` in test setup
+2. Create realistic test data using the models
+3. Use `@patch('app.services.media.service.get_client_for_media_server')` for API mocking
+4. Test both success and failure scenarios
+5. Verify database state after operations
+6. Check mock state for API calls
+
+### Debugging Failed Tests
+1. Check mock state: `get_mock_state().users`
+2. Examine database records: `User.query.all()`
+3. Review error messages in test output
+4. Use `-s` flag to see print statements: `pytest -s`
+
+### Performance Testing
+1. Use `time.time()` for timing measurements
+2. Set reasonable performance thresholds
+3. Test with realistic data volumes
+4. Monitor memory usage for long-running tests
+
+## Future Enhancements
+
+- [ ] Add visual regression tests for invitation pages
+- [ ] Test invitation email notifications
+- [ ] Add API rate limiting tests  
+- [ ] Test invitation analytics and metrics
+- [ ] Add security penetration tests
+- [ ] Test invitation QR code generation

tests/e2e/test_invitation_e2e.py

@@ -0,0 +1,405 @@
+"""
+End-to-end tests for invitation workflows using Playwright.
+
+These tests simulate the complete user journey from receiving an invitation
+link to successfully creating accounts on media servers.
+"""
+
+from unittest.mock import patch
+
+import pytest
+from playwright.sync_api import Page, expect
+
+from app.extensions import db
+from app.models import Invitation, MediaServer
+from tests.mocks.media_server_mocks import (
+    create_mock_client,
+    get_mock_state,
+    setup_mock_servers,
+)
+
+
+@pytest.fixture
+def invitation_setup(app):
+    """Setup test data for invitation E2E tests."""
+    with app.app_context():
+        setup_mock_servers()
+
+        # Create media server
+        server = MediaServer(
+            name="Test Jellyfin Server",
+            server_type="jellyfin",
+            url="http://localhost:8096",
+            api_key="test-api-key",
+        )
+        db.session.add(server)
+        db.session.flush()
+
+        # Create valid invitation
+        invitation = Invitation(
+            code="E2ETEST123",
+            server_id=server.id,
+            duration="30",
+            used=False,
+            unlimited=False,
+        )
+        db.session.add(invitation)
+        db.session.commit()
+
+        yield {
+            "server": server,
+            "invitation": invitation,
+            "invitation_url": f"/j/{invitation.code}",
+        }
+
+
+class TestInvitationUserJourney:
+    """Test complete user journey through invitation process."""
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_successful_invitation_flow(
+        self, mock_get_client, page: Page, live_server, invitation_setup
+    ):
+        """Test successful invitation acceptance and account creation."""
+        # Setup mock client
+        mock_client = create_mock_client(
+            "jellyfin", server_id=invitation_setup["server"].id
+        )
+        mock_get_client.return_value = mock_client
+
+        # Navigate to invitation page
+        page.goto(f"{live_server.url}{invitation_setup['invitation_url']}")
+
+        # Verify invitation page loads
+        expect(page.locator("h1")).to_contain_text("Join Test Jellyfin Server")
+
+        # Check that form is present
+        expect(page.locator("form")).to_be_visible()
+        expect(page.locator("input[name='username']")).to_be_visible()
+        expect(page.locator("input[name='password']")).to_be_visible()
+        expect(page.locator("input[name='confirm']")).to_be_visible()
+        expect(page.locator("input[name='email']")).to_be_visible()
+
+        # Fill out the form
+        page.fill("input[name='username']", "e2etestuser")
+        page.fill("input[name='password']", "testpassword123")
+        page.fill("input[name='confirm']", "testpassword123")
+        page.fill("input[name='email']", "e2etest@example.com")
+
+        # Submit the form
+        page.click("button[type='submit']")
+
+        # Wait for success (should redirect to wizard or success page)
+        # Adjust this based on your actual success flow
+        page.wait_for_url("**/wizard/**", timeout=10000)
+
+        # Verify success indicators
+        expect(page.locator("body")).to_contain_text("success", ignore_case=True)
+
+        # Verify user was created in mock server
+        mock_users = get_mock_state().users
+        assert len(mock_users) == 1
+        created_user = list(mock_users.values())[0]
+        assert created_user.username == "e2etestuser"
+        assert created_user.email == "e2etest@example.com"
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_invitation_form_validation(
+        self, mock_get_client, page: Page, live_server, invitation_setup
+    ):
+        """Test form validation on invitation page."""
+        mock_client = create_mock_client(
+            "jellyfin", server_id=invitation_setup["server"].id
+        )
+        mock_get_client.return_value = mock_client
+
+        # Navigate to invitation page
+        page.goto(f"{live_server.url}{invitation_setup['invitation_url']}")
+
+        # Test empty form submission
+        page.click("button[type='submit']")
+
+        # Should show validation errors
+        expect(page.locator(".error, .alert-danger, [data-error]")).to_be_visible()
+
+        # Test password mismatch
+        page.fill("input[name='username']", "testuser")
+        page.fill("input[name='password']", "password123")
+        page.fill("input[name='confirm']", "differentpassword")
+        page.fill("input[name='email']", "test@example.com")
+        page.click("button[type='submit']")
+
+        # Should show password mismatch error
+        expect(page.locator("body")).to_contain_text("password", ignore_case=True)
+        expect(page.locator("body")).to_contain_text("match", ignore_case=True)
+
+        # Test invalid email
+        page.fill("input[name='confirm']", "password123")  # Fix password
+        page.fill("input[name='email']", "invalid-email")
+        page.click("button[type='submit']")
+
+        # Should show email validation error
+        expect(page.locator("body")).to_contain_text("email", ignore_case=True)
+
+    def test_expired_invitation(self, page: Page, live_server, app):
+        """Test that expired invitations show appropriate error."""
+        with app.app_context():
+            from datetime import UTC, datetime, timedelta
+
+            # Create expired invitation
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            expired_invitation = Invitation(
+                code="EXPIRED123",
+                server_id=server.id,
+                expires=datetime.now(UTC) - timedelta(hours=1),
+                used=False,
+            )
+            db.session.add(expired_invitation)
+            db.session.commit()
+
+        # Navigate to expired invitation
+        page.goto(f"{live_server.url}/j/EXPIRED123")
+
+        # Should show expiry error
+        expect(page.locator("body")).to_contain_text("expired", ignore_case=True)
+
+        # Form should not be present
+        expect(page.locator("form")).not_to_be_visible()
+
+    def test_invalid_invitation_code(self, page: Page, live_server):
+        """Test that invalid invitation codes show 404 or error page."""
+        # Navigate to non-existent invitation
+        page.goto(f"{live_server.url}/j/INVALIDCODE123")
+
+        # Should show error (either 404 or invalid invitation message)
+        page.wait_for_load_state("networkidle")
+
+        # Check for either 404 or invalid invitation message
+        body_text = page.locator("body").inner_text().lower()
+        assert any(
+            phrase in body_text
+            for phrase in ["not found", "invalid", "expired", "does not exist", "404"]
+        )
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_server_error_handling(
+        self, mock_get_client, page: Page, live_server, invitation_setup
+    ):
+        """Test handling when media server is unavailable."""
+        # Setup failing mock client
+        mock_client = create_mock_client(
+            "jellyfin", server_id=invitation_setup["server"].id
+        )
+        mock_client._do_join = lambda *args, **kwargs: (
+            False,
+            "Server temporarily unavailable",
+        )
+        mock_get_client.return_value = mock_client
+
+        # Navigate to invitation page
+        page.goto(f"{live_server.url}{invitation_setup['invitation_url']}")
+
+        # Fill and submit form
+        page.fill("input[name='username']", "erroruser")
+        page.fill("input[name='password']", "testpass123")
+        page.fill("input[name='confirm']", "testpass123")
+        page.fill("input[name='email']", "error@example.com")
+        page.click("button[type='submit']")
+
+        # Should show server error
+        expect(page.locator("body")).to_contain_text("server", ignore_case=True)
+        expect(page.locator("body")).to_contain_text("unavailable", ignore_case=True)
+
+        # Form should still be visible for retry
+        expect(page.locator("form")).to_be_visible()
+
+
+class TestMultiServerInvitationFlow:
+    """Test E2E flows for multi-server invitations."""
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_invitation_success(
+        self, mock_get_client, page: Page, live_server, app
+    ):
+        """Test successful multi-server invitation flow."""
+        with app.app_context():
+            setup_mock_servers()
+
+            # Create multiple servers
+            jellyfin_server = MediaServer(
+                name="Jellyfin Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="jellyfin-key",
+            )
+            plex_server = MediaServer(
+                name="Plex Server",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="plex-key",
+            )
+            db.session.add_all([jellyfin_server, plex_server])
+            db.session.flush()
+
+            # Create multi-server invitation
+            invitation = Invitation(code="MULTI123", duration="30", used=False)
+            invitation.servers = [jellyfin_server, plex_server]
+            db.session.add(invitation)
+            db.session.commit()
+
+        # Setup mock clients
+        def get_client_side_effect(server):
+            if server.server_type == "jellyfin":
+                return create_mock_client("jellyfin", server_id=server.id)
+            if server.server_type == "plex":
+                return create_mock_client("plex", server_id=server.id)
+            return None
+
+        mock_get_client.side_effect = get_client_side_effect
+
+        # Navigate to invitation page
+        page.goto(f"{live_server.url}/j/MULTI123")
+
+        # Should show multi-server invitation info
+        expect(page.locator("body")).to_contain_text("Jellyfin Server")
+        expect(page.locator("body")).to_contain_text("Plex Server")
+
+        # Fill and submit form
+        page.fill("input[name='username']", "multiuser")
+        page.fill("input[name='password']", "testpass123")
+        page.fill("input[name='confirm']", "testpass123")
+        page.fill("input[name='email']", "multi@example.com")
+        page.click("button[type='submit']")
+
+        # Wait for success
+        page.wait_for_url("**/wizard/**", timeout=15000)
+
+        # Verify users created on both servers
+        mock_users = get_mock_state().users
+        assert len(mock_users) == 2  # One user per server
+
+        usernames = [user.username for user in mock_users.values()]
+        assert all(username == "multiuser" for username in usernames)
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_partial_failure(
+        self, mock_get_client, page: Page, live_server, app
+    ):
+        """Test multi-server invitation with partial failures."""
+        with app.app_context():
+            setup_mock_servers()
+
+            # Create servers
+            jellyfin_server = MediaServer(
+                name="Working Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="working-key",
+            )
+            broken_server = MediaServer(
+                name="Broken Server",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="broken-key",
+            )
+            db.session.add_all([jellyfin_server, broken_server])
+            db.session.flush()
+
+            invitation = Invitation(code="PARTIAL123", used=False)
+            invitation.servers = [jellyfin_server, broken_server]
+            db.session.add(invitation)
+            db.session.commit()
+
+        # Setup clients - one working, one failing
+        def get_client_side_effect(server):
+            if server.name == "Working Server":
+                return create_mock_client("jellyfin", server_id=server.id)
+            client = create_mock_client("plex", server_id=server.id)
+            client._do_join = lambda *args, **kwargs: (
+                False,
+                "Server is down for maintenance",
+            )
+            return client
+
+        mock_get_client.side_effect = get_client_side_effect
+
+        # Navigate and submit
+        page.goto(f"{live_server.url}/j/PARTIAL123")
+        page.fill("input[name='username']", "partialuser")
+        page.fill("input[name='password']", "testpass123")
+        page.fill("input[name='confirm']", "testpass123")
+        page.fill("input[name='email']", "partial@example.com")
+        page.click("button[type='submit']")
+
+        # Should show partial success message
+        page.wait_for_timeout(2000)  # Allow processing
+
+        # Should mention both success and failure
+        body_text = page.locator("body").inner_text().lower()
+
+        # Should indicate partial success (at least one server worked)
+        # and show error for the failed server
+        assert "working server" in body_text or "success" in body_text
+        assert "broken server" in body_text or "maintenance" in body_text
+
+
+class TestInvitationUIComponents:
+    """Test UI components and interactions on invitation pages."""
+
+    def test_invitation_page_accessibility(
+        self, page: Page, live_server, invitation_setup
+    ):
+        """Test basic accessibility of invitation form."""
+        page.goto(f"{live_server.url}{invitation_setup['invitation_url']}")
+
+        # Check for form labels
+        expect(
+            page.locator("label[for*='username'], input[name='username'][aria-label]")
+        ).to_be_visible()
+        expect(
+            page.locator("label[for*='password'], input[name='password'][aria-label]")
+        ).to_be_visible()
+        expect(
+            page.locator("label[for*='email'], input[name='email'][aria-label]")
+        ).to_be_visible()
+
+        # Check form has proper structure
+        expect(page.locator("form")).to_have_attribute("method", "post")
+
+        # Test keyboard navigation
+        page.keyboard.press("Tab")  # Should focus first input
+        focused_element = page.evaluate("document.activeElement.name")
+        assert focused_element in ["username", "password", "email"]
+
+    def test_responsive_design(self, page: Page, live_server, invitation_setup):
+        """Test invitation page on different screen sizes."""
+        # Test desktop
+        page.set_viewport_size({"width": 1920, "height": 1080})
+        page.goto(f"{live_server.url}{invitation_setup['invitation_url']}")
+        expect(page.locator("form")).to_be_visible()
+
+        # Test tablet
+        page.set_viewport_size({"width": 768, "height": 1024})
+        page.reload()
+        expect(page.locator("form")).to_be_visible()
+
+        # Test mobile
+        page.set_viewport_size({"width": 375, "height": 667})
+        page.reload()
+        expect(page.locator("form")).to_be_visible()
+
+        # Form should remain usable at all sizes
+        expect(page.locator("input[name='username']")).to_be_visible()
+        expect(page.locator("button[type='submit']")).to_be_visible()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "--headed"])  # Run with visible browser for debugging

tests/mocks/__init__.py

@@ -0,0 +1,32 @@
+"""
+Mock implementations for testing.
+
+This package provides mock implementations of external services
+for testing Wizarr functionality without requiring real services.
+"""
+
+from .media_server_mocks import (
+    MockAudiobookshelfClient,
+    MockJellyfinClient,
+    MockPlexClient,
+    create_mock_client,
+    get_mock_state,
+    mock_state,
+    setup_mock_servers,
+    simulate_auth_failure,
+    simulate_server_failure,
+    simulate_user_creation_failure,
+)
+
+__all__ = [
+    "create_mock_client",
+    "setup_mock_servers",
+    "simulate_server_failure",
+    "simulate_auth_failure",
+    "simulate_user_creation_failure",
+    "get_mock_state",
+    "mock_state",
+    "MockJellyfinClient",
+    "MockPlexClient",
+    "MockAudiobookshelfClient",
+]

tests/mocks/media_server_mocks.py

@@ -0,0 +1,333 @@
+"""
+Mock implementations for media server APIs used in invitation testing.
+
+These mocks simulate the behavior of external media server APIs (Plex, Jellyfin, etc.)
+to enable testing of invitation flows without requiring actual server instances.
+"""
+
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+from unittest.mock import Mock
+
+
+@dataclass
+class MockUser:
+    """Represents a user in a mock media server."""
+
+    id: str
+    username: str
+    email: str
+    enabled: bool = True
+    libraries: list[str] = field(default_factory=list)
+    policy: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class MockLibrary:
+    """Represents a library in a mock media server."""
+
+    id: str
+    name: str
+    type: str
+    enabled: bool = True
+
+
+class MockMediaServerState:
+    """Shared state for mock media servers to simulate persistent data."""
+
+    def __init__(self):
+        self.users: dict[str, MockUser] = {}
+        self.libraries: dict[str, MockLibrary] = {
+            "lib1": MockLibrary("lib1", "Movies", "movie"),
+            "lib2": MockLibrary("lib2", "TV Shows", "show"),
+            "lib3": MockLibrary("lib3", "Music", "artist"),
+        }
+        self.connection_healthy = True
+        self.api_key_valid = True
+        self.create_user_failures = []  # List of usernames that should fail
+
+    def reset(self):
+        """Reset state for fresh tests."""
+        self.users.clear()
+        self.connection_healthy = True
+        self.api_key_valid = True
+        self.create_user_failures.clear()
+
+
+# Global state instance for sharing across mock clients
+mock_state = MockMediaServerState()
+
+
+class MockJellyfinClient:
+    """Mock Jellyfin client that simulates API responses."""
+
+    def __init__(self, url: str = None, token: str = None, **kwargs):
+        self.url = url or "http://localhost:8096"
+        self.token = token or "mock-api-key"
+        self.server_id = kwargs.get("server_id", 1)
+
+    def validate_connection(self) -> tuple[bool, str]:
+        """Simulate connection validation."""
+        if not mock_state.connection_healthy:
+            return False, "Connection failed"
+        if not mock_state.api_key_valid:
+            return False, "Invalid API key"
+        return True, "Connection successful"
+
+    def libraries(self) -> dict[str, str]:
+        """Return mock library mapping."""
+        return {
+            lib.name: lib.id for lib in mock_state.libraries.values() if lib.enabled
+        }
+
+    def create_user(self, username: str, password: str) -> str:
+        """Simulate user creation and return user ID."""
+        if username in mock_state.create_user_failures:
+            raise Exception(f"Failed to create user {username}")
+
+        user_id = str(uuid.uuid4())
+        mock_state.users[user_id] = MockUser(
+            id=user_id,
+            username=username,
+            email="",  # Set later
+        )
+        return user_id
+
+    def get_user(self, user_id: str) -> dict[str, Any]:
+        """Get user details."""
+        if user_id not in mock_state.users:
+            raise Exception(f"User {user_id} not found")
+
+        user = mock_state.users[user_id]
+        return {
+            "Id": user.id,
+            "Name": user.username,
+            "Primary": user.email,
+            "Policy": user.policy,
+            "Configuration": {
+                "DisplayMissingEpisodes": False,
+                "GroupedFolders": [],
+                "SubtitleMode": "Default",
+                "EnableLocalPassword": True,
+            },
+        }
+
+    def set_policy(self, user_id: str, policy: dict[str, Any]) -> None:
+        """Update user policy."""
+        if user_id in mock_state.users:
+            mock_state.users[user_id].policy.update(policy)
+
+    def _set_specific_folders(self, user_id: str, library_ids: list[str]) -> None:
+        """Set library access for user."""
+        if user_id in mock_state.users:
+            mock_state.users[user_id].libraries = library_ids
+
+    def get(self, endpoint: str):
+        """Mock HTTP GET requests."""
+        response_mock = Mock()
+        if "/Users/" in endpoint:
+            user_id = endpoint.split("/Users/")[1]
+            try:
+                response_mock.json.return_value = self.get_user(user_id)
+            except Exception:
+                response_mock.json.return_value = {"error": "User not found"}
+        else:
+            response_mock.json.return_value = {"success": True}
+        return response_mock
+
+    def post(self, endpoint: str, **kwargs):
+        """Mock HTTP POST requests."""
+        response_mock = Mock()
+        response_mock.json.return_value = {"success": True}
+        return response_mock
+
+    def patch(self, endpoint: str, **kwargs):
+        """Mock HTTP PATCH requests."""
+        response_mock = Mock()
+        response_mock.json.return_value = {"success": True}
+        return response_mock
+
+    def delete(self, endpoint: str, **kwargs):
+        """Mock HTTP DELETE requests."""
+        response_mock = Mock()
+        response_mock.json.return_value = {"success": True}
+        return response_mock
+
+    def _do_join(
+        self, username: str, password: str, confirm: str, email: str, code: str
+    ) -> tuple[bool, str]:
+        """Mock implementation of invitation join process."""
+        # Basic validation
+        if password != confirm:
+            return False, "Passwords do not match"
+        if len(password) < 8:
+            return False, "Password too short"
+        if username in mock_state.create_user_failures:
+            return False, f"Failed to create user {username}"
+
+        try:
+            # Simulate user creation
+            user_id = self.create_user(username, password)
+
+            # Set email
+            mock_state.users[user_id].email = email
+
+            # Set default libraries (simulate library assignment)
+            enabled_libs = [
+                lib_id for lib_id, lib in mock_state.libraries.items() if lib.enabled
+            ]
+            self._set_specific_folders(user_id, enabled_libs)
+
+            # Set default policy
+            default_policy = {
+                "EnableDownloads": True,
+                "EnableLiveTvAccess": False,
+                "IsAdministrator": False,
+            }
+            self.set_policy(user_id, default_policy)
+
+            return True, ""
+
+        except Exception as e:
+            return False, str(e)
+
+
+class MockPlexClient:
+    """Mock Plex client that simulates PlexAPI responses."""
+
+    def __init__(self, url: str = None, token: str = None, **kwargs):
+        self.url = url or "http://localhost:32400"
+        self.token = token or "mock-plex-token"
+        self.server_id = kwargs.get("server_id", 2)
+
+    def validate_connection(self) -> tuple[bool, str]:
+        """Simulate connection validation."""
+        if not mock_state.connection_healthy:
+            return False, "Plex server unreachable"
+        if not mock_state.api_key_valid:
+            return False, "Invalid Plex token"
+        return True, "Connected to Plex"
+
+    def libraries(self) -> dict[str, str]:
+        """Return mock Plex library mapping."""
+        return {
+            lib.name: lib.id for lib in mock_state.libraries.values() if lib.enabled
+        }
+
+    def _do_join(
+        self, username: str, password: str, confirm: str, email: str, code: str
+    ) -> tuple[bool, str]:
+        """Mock Plex invitation join process."""
+        if password != confirm:
+            return False, "Passwords do not match"
+        if "@" not in email:
+            return False, "Invalid email format"
+        if username in mock_state.create_user_failures:
+            return False, f"Plex user creation failed for {username}"
+
+        try:
+            # Simulate Plex user creation (Plex uses email as identifier)
+            user_id = str(uuid.uuid4())
+            mock_state.users[user_id] = MockUser(
+                id=user_id,
+                username=username,
+                email=email,
+                libraries=list(
+                    mock_state.libraries.keys()
+                ),  # Plex gets all libraries by default
+            )
+            return True, ""
+
+        except Exception as e:
+            return False, f"Plex error: {str(e)}"
+
+
+class MockAudiobookshelfClient:
+    """Mock Audiobookshelf client."""
+
+    def __init__(self, url: str = None, token: str = None, **kwargs):
+        self.url = url or "http://localhost:13378"
+        self.token = token or "mock-abs-token"
+        self.server_id = kwargs.get("server_id", 3)
+
+    def validate_connection(self) -> tuple[bool, str]:
+        if not mock_state.connection_healthy:
+            return False, "Audiobookshelf server unreachable"
+        return True, "Connected to Audiobookshelf"
+
+    def libraries(self) -> dict[str, str]:
+        # Audiobookshelf typically has audiobook and podcast libraries
+        return {"audiobooks": "Audiobooks", "podcasts": "Podcasts"}
+
+    def _do_join(
+        self, username: str, password: str, confirm: str, email: str, code: str
+    ) -> tuple[bool, str]:
+        if password != confirm:
+            return False, "Passwords do not match"
+        if username in mock_state.create_user_failures:
+            return False, f"ABS user creation failed for {username}"
+
+        try:
+            user_id = str(uuid.uuid4())
+            mock_state.users[user_id] = MockUser(
+                id=user_id,
+                username=username,
+                email=email,
+                policy={"isActive": True, "canDownload": True},
+            )
+            return True, ""
+
+        except Exception as e:
+            return False, f"Audiobookshelf error: {str(e)}"
+
+
+# Factory function to create mock clients
+def create_mock_client(server_type: str, **kwargs):
+    """Create appropriate mock client based on server type."""
+    mock_clients = {
+        "jellyfin": MockJellyfinClient,
+        "plex": MockPlexClient,
+        "audiobookshelf": MockAudiobookshelfClient,
+    }
+
+    client_class = mock_clients.get(server_type.lower())
+    if not client_class:
+        raise ValueError(f"No mock client available for server type: {server_type}")
+
+    return client_class(**kwargs)
+
+
+# Helper functions for test setup
+def setup_mock_servers():
+    """Setup mock servers with realistic data."""
+    mock_state.reset()
+
+    # Add some realistic libraries
+    mock_state.libraries.update(
+        {
+            "movies_4k": MockLibrary("movies_4k", "Movies 4K", "movie"),
+            "anime": MockLibrary("anime", "Anime", "show"),
+            "audiobooks": MockLibrary("audiobooks", "Audiobooks", "audiobook"),
+        }
+    )
+
+
+def simulate_server_failure():
+    """Simulate server connection failure."""
+    mock_state.connection_healthy = False
+
+
+def simulate_auth_failure():
+    """Simulate authentication failure."""
+    mock_state.api_key_valid = False
+
+
+def simulate_user_creation_failure(usernames: list[str]):
+    """Simulate user creation failures for specific usernames."""
+    mock_state.create_user_failures.extend(usernames)
+
+
+def get_mock_state():
+    """Get current mock state for assertions."""
+    return mock_state

tests/test_debug_invitation.py

@@ -0,0 +1,38 @@
+"""Debug test to understand invitation validation logic."""
+
+import pytest
+
+from app.extensions import db
+from app.models import Invitation
+from app.services.invites import is_invite_valid
+
+
+def test_debug_unlimited_invitation(app):
+    """Debug test for unlimited invitation validation."""
+    with app.app_context():
+        # Create used unlimited invitation
+        invite = Invitation(code="UNLIMITED123", used=True, unlimited=True)
+        db.session.add(invite)
+        db.session.commit()
+
+        # Debug the invitation state
+        print(f"Invitation code: {invite.code}")
+        print(f"Invitation used: {invite.used}")
+        print(f"Invitation unlimited: {invite.unlimited}")
+        print(f"Used is True: {invite.used is True}")
+        print(f"Unlimited is not True: {invite.unlimited is not True}")
+        print(
+            f"Both conditions: {invite.used is True and invite.unlimited is not True}"
+        )
+
+        # Test validation
+        is_valid, message = is_invite_valid(invite.code)
+        print(f"Is valid: {is_valid}")
+        print(f"Message: {message}")
+
+        # This should be valid for unlimited invitations
+        assert is_valid, f"Expected valid but got: {message}"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "-s"])

tests/test_invitation_comprehensive.py

@@ -0,0 +1,691 @@
+"""
+Comprehensive invitation system tests with API simulation.
+
+This test suite covers the complete invitation workflow including:
+- Single and multi-server invitations
+- Various media server types (Jellyfin, Plex, Audiobookshelf)
+- Error scenarios and rollback behavior
+- Library assignment and user permissions
+- Expiry and limitation handling
+"""
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import Mock, patch
+
+import pytest
+
+from app.extensions import db
+from app.models import Invitation, Library, MediaServer, User
+from app.services.invitation_manager import InvitationManager
+from app.services.invites import create_invite, is_invite_valid, mark_server_used
+from tests.mocks.media_server_mocks import (
+    create_mock_client,
+    get_mock_state,
+    setup_mock_servers,
+    simulate_server_failure,
+    simulate_user_creation_failure,
+)
+
+
+class TestInvitationValidation:
+    """Test invitation validation logic."""
+
+    def test_valid_invitation(self, app):
+        """Test that valid invitations pass validation."""
+        with app.app_context():
+            # Create a valid invitation
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation using form-like data
+            form_data = {
+                "duration": "30",
+                "expires": "month",
+                "unlimited": False,
+                "server_ids": [str(server.id)],
+            }
+            invite = create_invite(form_data)
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert is_valid
+            assert message == ""
+
+    def test_expired_invitation(self, app):
+        """Test that expired invitations are rejected."""
+        with app.app_context():
+            # Create invitation that expires immediately
+            invite = Invitation(
+                code="EXPIRED123",
+                expires=datetime.now(UTC) - timedelta(hours=1),
+                used=False,
+                unlimited=False,
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert not is_valid
+            assert "expired" in message.lower()
+
+    def test_used_limited_invitation(self, app):
+        """Test that used limited invitations are rejected."""
+        with app.app_context():
+            # Create used limited invitation
+            invite = Invitation(code="USED123", used=True, unlimited=False)
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert not is_valid
+            assert "already been used" in message
+
+    def test_used_unlimited_invitation(self, app):
+        """Test that used unlimited invitations are still valid."""
+        with app.app_context():
+            # Create used unlimited invitation
+            invite = Invitation(code="UNLIMITED123", used=True, unlimited=True)
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert is_valid
+            assert message in ["okay", ""]
+
+
+class TestSingleServerInvitations:
+    """Test invitation processing for single media servers."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_successful_jellyfin_invitation(self, mock_get_client, app):
+        """Test successful invitation process for Jellyfin server."""
+        with app.app_context():
+            # Setup mock client
+            mock_client = create_mock_client("jellyfin", server_id=1)
+            mock_get_client.return_value = mock_client
+
+            # Create server and invitation
+            server = MediaServer(
+                name="Test Jellyfin",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invite = Invitation(
+                code="JELLYFIN123",
+                server_id=server.id,
+                duration="30",
+                used=False,
+                unlimited=False,
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="JELLYFIN123",
+                username="testuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="test@example.com",
+            )
+
+            # Verify results
+            assert success
+            assert redirect_code == "JELLYFIN123"
+            assert len(errors) == 0
+
+            # Verify user was created in mock
+            mock_users = get_mock_state().users
+            assert len(mock_users) == 1
+            created_user = list(mock_users.values())[0]
+            assert created_user.username == "testuser"
+            assert created_user.email == "test@example.com"
+
+            # Verify database user was created
+            db_user = User.query.filter_by(code="JELLYFIN123").first()
+            assert db_user is not None
+            assert db_user.username == "testuser"
+            assert db_user.server_id == server.id
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_plex_invitation_with_oauth(self, mock_get_client, app):
+        """Test Plex invitation that might use OAuth flow."""
+        with app.app_context():
+            mock_client = create_mock_client("plex", server_id=2)
+            mock_get_client.return_value = mock_client
+
+            # Create Plex server
+            server = MediaServer(
+                name="Test Plex",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="plex-token",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invite = Invitation(
+                code="PLEX123",
+                server_id=server.id,
+                plex_home=True,
+                plex_allow_sync=True,
+                unlimited=True,
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="PLEX123",
+                username="plexuser",
+                password="plexpass123",
+                confirm_password="plexpass123",
+                email="plex@example.com",
+            )
+
+            assert success
+            assert redirect_code == "PLEX123"
+
+            # Verify Plex-specific behavior
+            mock_users = get_mock_state().users
+            assert len(mock_users) == 1
+            created_user = list(mock_users.values())[0]
+            assert (
+                created_user.email == "plex@example.com"
+            )  # Plex uses email as primary identifier
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_server_connection_failure(self, mock_get_client, app):
+        """Test handling of server connection failures."""
+        with app.app_context():
+            # Simulate connection failure
+            simulate_server_failure()
+
+            mock_client = create_mock_client("jellyfin", server_id=1)
+            mock_get_client.return_value = mock_client
+
+            # Create server and invitation
+            server = MediaServer(
+                name="Failing Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invite = Invitation(code="FAIL123", server_id=server.id)
+            db.session.add(invite)
+            db.session.commit()
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="FAIL123",
+                username="testuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="test@example.com",
+            )
+
+            # Should fail due to connection issue
+            assert not success
+            assert redirect_code is None
+            assert len(errors) > 0
+            assert any(
+                "Connection failed" in error or "server unreachable" in error.lower()
+                for error in errors
+            )
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_user_creation_failure(self, mock_get_client, app):
+        """Test handling when user creation fails on media server."""
+        with app.app_context():
+            # Simulate user creation failure
+            simulate_user_creation_failure(["baduser"])
+
+            mock_client = create_mock_client("jellyfin", server_id=1)
+            mock_get_client.return_value = mock_client
+
+            # Create server and invitation
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invite = Invitation(code="BADUSER123", server_id=server.id)
+            db.session.add(invite)
+            db.session.commit()
+
+            # Process invitation with username that will fail
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="BADUSER123",
+                username="baduser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="test@example.com",
+            )
+
+            # Should fail due to user creation issue
+            assert not success
+            assert redirect_code is None
+            assert len(errors) > 0
+            assert any("Failed to create user baduser" in error for error in errors)
+
+
+class TestMultiServerInvitations:
+    """Test invitation processing across multiple media servers."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_success(self, mock_get_client, app):
+        """Test successful multi-server invitation."""
+        with app.app_context():
+            # Setup multiple servers
+            jellyfin_server = MediaServer(
+                name="Jellyfin Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="jellyfin-key",
+            )
+            plex_server = MediaServer(
+                name="Plex Server",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="plex-key",
+            )
+            db.session.add_all([jellyfin_server, plex_server])
+            db.session.flush()
+
+            # Create multi-server invitation
+            invite = Invitation(code="MULTI123", duration="30", unlimited=False)
+            invite.servers = [jellyfin_server, plex_server]
+            db.session.add(invite)
+            db.session.commit()
+
+            # Setup mock clients
+            def get_client_side_effect(server):
+                if server.server_type == "jellyfin":
+                    return create_mock_client("jellyfin", server_id=server.id)
+                if server.server_type == "plex":
+                    return create_mock_client("plex", server_id=server.id)
+                return None
+
+            mock_get_client.side_effect = get_client_side_effect
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="MULTI123",
+                username="multiuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="multi@example.com",
+            )
+
+            # Should succeed on both servers
+            assert success
+            assert redirect_code == "MULTI123"
+            assert len(errors) == 0  # No errors expected
+
+            # Verify users created on both servers
+            mock_users = get_mock_state().users
+            assert len(mock_users) == 2  # One user per server
+
+            # Verify database users
+            db_users = User.query.filter_by(code="MULTI123").all()
+            assert len(db_users) == 2
+            server_ids = {user.server_id for user in db_users}
+            assert server_ids == {jellyfin_server.id, plex_server.id}
+
+            # Verify identity linking (users should share same identity)
+            identities = {user.identity_id for user in db_users if user.identity_id}
+            assert (
+                len(identities) <= 1
+            )  # Should be linked to same identity or both None
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_partial_failure(self, mock_get_client, app):
+        """Test multi-server invitation where one server fails."""
+        with app.app_context():
+            # Setup servers
+            jellyfin_server = MediaServer(
+                name="Jellyfin Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="jellyfin-key",
+            )
+            plex_server = MediaServer(
+                name="Plex Server",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="plex-key",
+            )
+            db.session.add_all([jellyfin_server, plex_server])
+            db.session.flush()
+
+            # Create invitation
+            invite = Invitation(code="PARTIAL123", unlimited=False)
+            invite.servers = [jellyfin_server, plex_server]
+            db.session.add(invite)
+            db.session.commit()
+
+            # Setup clients - make Plex fail
+            def get_client_side_effect(server):
+                if server.server_type == "jellyfin":
+                    return create_mock_client("jellyfin", server_id=server.id)
+                if server.server_type == "plex":
+                    # Simulate Plex failure
+                    client = create_mock_client("plex", server_id=server.id)
+                    client._do_join = Mock(return_value=(False, "Plex server error"))
+                    return client
+                return None
+
+            mock_get_client.side_effect = get_client_side_effect
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="PARTIAL123",
+                username="partialuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="partial@example.com",
+            )
+
+            # Should succeed overall (at least one server worked)
+            assert success
+            assert redirect_code == "PARTIAL123"
+            assert len(errors) == 1  # One error from Plex
+            assert "Plex server error" in errors[0]
+
+            # Verify only Jellyfin user was created
+            mock_users = get_mock_state().users
+            jellyfin_users = [
+                u for u in mock_users.values() if "Plex server error" not in str(u)
+            ]
+            assert len(jellyfin_users) == 1
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_complete_failure(self, mock_get_client, app):
+        """Test multi-server invitation where all servers fail."""
+        with app.app_context():
+            # Setup servers
+            server1 = MediaServer(
+                name="Server 1",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="key1",
+            )
+            server2 = MediaServer(
+                name="Server 2",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="key2",
+            )
+            db.session.add_all([server1, server2])
+            db.session.flush()
+
+            # Create invitation
+            invite = Invitation(code="ALLFAIL123", unlimited=False)
+            invite.servers = [server1, server2]
+            db.session.add(invite)
+            db.session.commit()
+
+            # Make all servers fail
+            def failing_client(server):
+                client = create_mock_client("jellyfin", server_id=server.id)
+                client._do_join = Mock(return_value=(False, f"{server.name} failed"))
+                return client
+
+            mock_get_client.side_effect = failing_client
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="ALLFAIL123",
+                username="failuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="fail@example.com",
+            )
+
+            # Should fail completely
+            assert not success
+            assert redirect_code is None
+            assert len(errors) == 2  # Errors from both servers
+            assert all("failed" in error for error in errors)
+
+
+class TestInvitationLibraryAssignment:
+    """Test library assignment during invitation processing."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_specific_library_assignment(self, mock_get_client, app):
+        """Test invitation with specific library restrictions."""
+        with app.app_context():
+            # Setup server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create libraries
+            lib1 = Library(
+                name="Movies", external_id="lib1", server_id=server.id, enabled=True
+            )
+            lib2 = Library(
+                name="TV Shows", external_id="lib2", server_id=server.id, enabled=True
+            )
+            lib3 = Library(
+                name="Music", external_id="lib3", server_id=server.id, enabled=True
+            )
+            db.session.add_all([lib1, lib2, lib3])
+            db.session.flush()
+
+            # Create invitation with specific libraries
+            invite = Invitation(code="SPECIFIC123", server_id=server.id)
+            invite.libraries = [lib1, lib2]  # Only Movies and TV
+            db.session.add(invite)
+            db.session.commit()
+
+            # Setup mock client
+            mock_client = create_mock_client("jellyfin", server_id=server.id)
+            mock_get_client.return_value = mock_client
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="SPECIFIC123",
+                username="libuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="lib@example.com",
+            )
+
+            assert success
+
+            # Verify user was assigned only specific libraries
+            mock_users = get_mock_state().users
+            assert len(mock_users) == 1
+            created_user = list(mock_users.values())[0]
+
+            # Should have access to lib1 and lib2, but not lib3
+            expected_libs = ["lib1", "lib2"]
+            assert set(created_user.libraries) == set(expected_libs)
+
+
+class TestInvitationExpiry:
+    """Test invitation expiry and duration handling."""
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_user_expiry_calculation(self, mock_get_client, app):
+        """Test that users get proper expiry dates based on invitation duration."""
+        with app.app_context():
+            # Setup server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation with 7-day duration
+            invite = Invitation(
+                code="EXPIRY123",
+                server_id=server.id,
+                duration="7",  # 7 days
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            mock_client = create_mock_client("jellyfin", server_id=server.id)
+            mock_get_client.return_value = mock_client
+
+            # Process invitation
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="EXPIRY123",
+                username="expiryuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="expiry@example.com",
+            )
+
+            assert success
+
+            # Verify user has expiry date set
+            db_user = User.query.filter_by(code="EXPIRY123").first()
+            assert db_user is not None
+            assert db_user.expires is not None
+
+            # Should expire in approximately 7 days
+            expected_expiry = datetime.now(UTC) + timedelta(days=7)
+            time_diff = abs((db_user.expires - expected_expiry).total_seconds())
+            assert time_diff < 60  # Within 1 minute of expected
+
+
+class TestInvitationMarkingUsed:
+    """Test invitation usage tracking and server marking."""
+
+    def test_mark_server_used_single_server(self, app):
+        """Test marking invitation as used for single server."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation properly with servers relationship
+            form_data = {
+                "expires": "month",
+                "unlimited": False,
+                "server_ids": [str(server.id)],
+            }
+            invite = create_invite(form_data)
+            invite.code = "MARK123"  # Set specific code for test
+
+            user = User(
+                username="testuser",
+                email="test@example.com",
+                token="user-token",
+                code="MARK123",
+                server_id=server.id,
+            )
+            db.session.add(user)
+            db.session.commit()
+
+            # Mark server as used
+            mark_server_used(invite, server.id, user)
+
+            # Verify invitation is marked as used (since only one server and it's limited)
+            db.session.refresh(invite)
+            assert invite.used is True
+            assert invite.used_by == user
+
+            # Verify user is in invitation's users collection
+            assert user in invite.users
+
+    def test_mark_server_used_multi_server_partial(self, app):
+        """Test marking one server as used in multi-server invitation."""
+        with app.app_context():
+            # Create multiple servers
+            server1 = MediaServer(
+                name="Server 1",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="key1",
+            )
+            server2 = MediaServer(
+                name="Server 2",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="key2",
+            )
+            db.session.add_all([server1, server2])
+            db.session.flush()
+
+            # Create invitation for both servers
+            invite = Invitation(code="MULTIMARK123", used=False, unlimited=False)
+            invite.servers = [server1, server2]
+            db.session.add(invite)
+            db.session.flush()
+
+            # Create user for first server
+            user1 = User(
+                username="user1",
+                email="user1@example.com",
+                token="token1",
+                code="MULTIMARK123",
+                server_id=server1.id,
+            )
+            db.session.add(user1)
+            db.session.commit()
+
+            # Mark only first server as used
+            mark_server_used(invite, server1.id, user1)
+            db.session.commit()
+
+            # Invitation should not be fully used yet (limited invitation with multiple servers)
+            db.session.refresh(invite)
+            assert invite.used is False  # Not all servers used yet
+
+            # But user should be tracked
+            assert user1 in invite.users
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

tests/test_invitation_performance.py

@@ -0,0 +1,449 @@
+"""
+Performance and load tests for invitation system.
+
+These tests ensure the invitation system can handle concurrent users
+and process invitations efficiently.
+"""
+
+import time
+from concurrent.futures import ThreadPoolExecutor
+from statistics import mean, median
+from unittest.mock import patch
+
+import pytest
+
+from app.extensions import db
+from app.models import Invitation, MediaServer, User
+from app.services.invitation_manager import InvitationManager
+from tests.mocks.media_server_mocks import (
+    create_mock_client,
+    get_mock_state,
+    setup_mock_servers,
+)
+
+
+class TestInvitationPerformance:
+    """Test invitation processing performance."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_single_invitation_processing_time(self, mock_get_client, app):
+        """Test time to process a single invitation."""
+        with app.app_context():
+            # Setup
+            server = MediaServer(
+                name="Performance Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="perf-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invitation = Invitation(code="PERF123", server_id=server.id, unlimited=True)
+            db.session.add(invitation)
+            db.session.commit()
+
+            mock_client = create_mock_client("jellyfin", server_id=server.id)
+            mock_get_client.return_value = mock_client
+
+            # Measure processing time
+            start_time = time.time()
+
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="PERF123",
+                username="perfuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="perf@example.com",
+            )
+
+            end_time = time.time()
+            processing_time = end_time - start_time
+
+            # Verify success
+            assert success
+            assert len(errors) == 0
+
+            # Performance assertion - should complete in under 1 second
+            assert processing_time < 1.0, (
+                f"Processing took {processing_time:.3f}s, expected < 1.0s"
+            )
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_concurrent_invitation_processing(self, mock_get_client, app):
+        """Test processing multiple invitations concurrently."""
+        with app.app_context():
+            # Setup server and unlimited invitation
+            server = MediaServer(
+                name="Concurrent Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="concurrent-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invitation = Invitation(
+                code="CONCURRENT123",
+                server_id=server.id,
+                unlimited=True,  # Allow multiple uses
+            )
+            db.session.add(invitation)
+            db.session.commit()
+
+            mock_client = create_mock_client("jellyfin", server_id=server.id)
+            mock_get_client.return_value = mock_client
+
+            # Function to process invitation
+            def process_invitation(user_id):
+                with app.app_context():
+                    success, redirect_code, errors = (
+                        InvitationManager.process_invitation(
+                            code="CONCURRENT123",
+                            username=f"user{user_id}",
+                            password="testpass123",
+                            confirm_password="testpass123",
+                            email=f"user{user_id}@example.com",
+                        )
+                    )
+                    return success, redirect_code, errors, user_id
+
+            # Process 10 invitations concurrently
+            num_users = 10
+            start_time = time.time()
+
+            with ThreadPoolExecutor(max_workers=5) as executor:
+                futures = [
+                    executor.submit(process_invitation, i) for i in range(num_users)
+                ]
+                results = [future.result() for future in futures]
+
+            end_time = time.time()
+            total_time = end_time - start_time
+
+            # Verify all succeeded
+            successful_count = sum(1 for success, _, _, _ in results if success)
+            assert successful_count == num_users, (
+                f"Only {successful_count}/{num_users} invitations succeeded"
+            )
+
+            # Performance assertion - should handle 10 concurrent users in reasonable time
+            assert total_time < 5.0, (
+                f"Concurrent processing took {total_time:.3f}s, expected < 5.0s"
+            )
+
+            # Verify users were created
+            created_users = get_mock_state().users
+            assert len(created_users) == num_users
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_multi_server_invitation_performance(self, mock_get_client, app):
+        """Test performance of multi-server invitation processing."""
+        with app.app_context():
+            setup_mock_servers()
+
+            # Create multiple servers
+            servers = []
+            for i in range(3):  # 3 servers
+                server = MediaServer(
+                    name=f"Server {i + 1}",
+                    server_type="jellyfin",
+                    url=f"http://localhost:809{6 + i}",
+                    api_key=f"key{i + 1}",
+                )
+                servers.append(server)
+                db.session.add(server)
+            db.session.flush()
+
+            # Create multi-server invitation
+            invitation = Invitation(code="MULTIPERFORMANCE123", unlimited=False)
+            invitation.servers = servers
+            db.session.add(invitation)
+            db.session.commit()
+
+            # Setup mock clients
+            def get_client_side_effect(server):
+                return create_mock_client("jellyfin", server_id=server.id)
+
+            mock_get_client.side_effect = get_client_side_effect
+
+            # Measure processing time
+            start_time = time.time()
+
+            success, redirect_code, errors = InvitationManager.process_invitation(
+                code="MULTIPERFORMANCE123",
+                username="multiuser",
+                password="testpass123",
+                confirm_password="testpass123",
+                email="multi@example.com",
+            )
+
+            end_time = time.time()
+            processing_time = end_time - start_time
+
+            # Verify success
+            assert success
+            assert len(errors) == 0
+
+            # Should process 3 servers in reasonable time
+            assert processing_time < 3.0, (
+                f"Multi-server processing took {processing_time:.3f}s, expected < 3.0s"
+            )
+
+            # Verify users created on all servers
+            created_users = get_mock_state().users
+            assert len(created_users) == 3
+
+
+class TestInvitationLoadTesting:
+    """Load testing for invitation system."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_invitation_validation_performance(self, mock_get_client, app):
+        """Test performance of invitation validation under load."""
+        with app.app_context():
+            # Create multiple invitations
+            server = MediaServer(
+                name="Load Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="load-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create 100 invitations
+            invitations = []
+            for i in range(100):
+                invitation = Invitation(
+                    code=f"LOAD{i:03d}", server_id=server.id, unlimited=True
+                )
+                invitations.append(invitation)
+                db.session.add(invitation)
+            db.session.commit()
+
+            # Test validation performance
+            from app.services.invites import is_invite_valid
+
+            validation_times = []
+
+            for invitation in invitations[:20]:  # Test first 20
+                start_time = time.time()
+                is_valid, message = is_invite_valid(invitation.code)
+                end_time = time.time()
+
+                validation_times.append(end_time - start_time)
+                assert is_valid  # Should all be valid
+
+            # Performance analysis
+            avg_time = mean(validation_times)
+            median_time = median(validation_times)
+            max_time = max(validation_times)
+
+            # Assertions
+            assert avg_time < 0.01, f"Average validation time {avg_time:.4f}s too slow"
+            assert median_time < 0.01, (
+                f"Median validation time {median_time:.4f}s too slow"
+            )
+            assert max_time < 0.05, f"Max validation time {max_time:.4f}s too slow"
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_database_performance_under_load(self, mock_get_client, app):
+        """Test database performance with many invitation records."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="DB Load Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="db-load-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create many invitations and users to simulate realistic load
+            for i in range(500):  # 500 invitations
+                invitation = Invitation(
+                    code=f"DBLOAD{i:04d}",
+                    server_id=server.id,
+                    unlimited=False,
+                    used=(i % 3 == 0),  # Every 3rd invitation is used
+                )
+                db.session.add(invitation)
+
+                if invitation.used:
+                    # Create corresponding user
+                    user = User(
+                        username=f"user{i}",
+                        email=f"user{i}@example.com",
+                        token=f"token{i}",
+                        code=invitation.code,
+                        server_id=server.id,
+                    )
+                    db.session.add(user)
+
+            db.session.commit()
+
+            # Test query performance
+            start_time = time.time()
+
+            # Simulate typical queries during invitation processing
+            for i in range(10):
+                code = f"DBLOAD{i:04d}"
+                invitation = Invitation.query.filter_by(code=code).first()
+                assert invitation is not None
+
+                # Check if user exists (typical validation)
+                user = User.query.filter_by(code=code, server_id=server.id).first()
+
+            end_time = time.time()
+            query_time = end_time - start_time
+
+            # Should handle queries efficiently even with large dataset
+            assert query_time < 1.0, (
+                f"Database queries took {query_time:.3f}s, expected < 1.0s"
+            )
+
+
+class TestInvitationMemoryUsage:
+    """Test memory usage during invitation processing."""
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_memory_efficient_processing(self, mock_get_client, app):
+        """Test that invitation processing doesn't leak memory."""
+        import gc
+        import tracemalloc
+
+        with app.app_context():
+            # Setup
+            server = MediaServer(
+                name="Memory Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="memory-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invitation = Invitation(
+                code="MEMORY123", server_id=server.id, unlimited=True
+            )
+            db.session.add(invitation)
+            db.session.commit()
+
+            mock_client = create_mock_client("jellyfin", server_id=server.id)
+            mock_get_client.return_value = mock_client
+
+            # Start memory tracking
+            tracemalloc.start()
+
+            # Process multiple invitations
+            for i in range(50):
+                success, redirect_code, errors = InvitationManager.process_invitation(
+                    code="MEMORY123",
+                    username=f"memuser{i}",
+                    password="testpass123",
+                    confirm_password="testpass123",
+                    email=f"memuser{i}@example.com",
+                )
+                assert success
+
+                # Force garbage collection periodically
+                if i % 10 == 0:
+                    gc.collect()
+
+            # Check memory usage
+            current, peak = tracemalloc.get_traced_memory()
+            tracemalloc.stop()
+
+            # Memory should be reasonable (less than 50MB peak)
+            peak_mb = peak / 1024 / 1024
+            assert peak_mb < 50, f"Peak memory usage {peak_mb:.2f}MB too high"
+
+
+class TestInvitationErrorRecovery:
+    """Test system recovery from errors during high load."""
+
+    def setup_method(self):
+        """Setup for each test."""
+        setup_mock_servers()
+
+    @patch("app.services.media.service.get_client_for_media_server")
+    def test_recovery_from_server_failures(self, mock_get_client, app):
+        """Test system recovery when servers fail under load."""
+        with app.app_context():
+            # Setup server
+            server = MediaServer(
+                name="Failure Recovery Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="recovery-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            invitation = Invitation(
+                code="RECOVERY123", server_id=server.id, unlimited=True
+            )
+            db.session.add(invitation)
+            db.session.commit()
+
+            # Setup client that fails intermittently
+            success_count = 0
+            failure_count = 0
+
+            def intermittent_client(server):
+                nonlocal success_count, failure_count
+                client = create_mock_client("jellyfin", server_id=server.id)
+
+                original_join = client._do_join
+
+                def failing_join(*args, **kwargs):
+                    nonlocal success_count, failure_count
+                    # Fail every 3rd request
+                    if (success_count + failure_count) % 3 == 2:
+                        failure_count += 1
+                        return False, "Temporary server error"
+                    success_count += 1
+                    return original_join(*args, **kwargs)
+
+                client._do_join = failing_join
+                return client
+
+            mock_get_client.side_effect = intermittent_client
+
+            # Process invitations - some should fail, others succeed
+            results = []
+            for i in range(20):
+                success, redirect_code, errors = InvitationManager.process_invitation(
+                    code="RECOVERY123",
+                    username=f"recoveryuser{i}",
+                    password="testpass123",
+                    confirm_password="testpass123",
+                    email=f"recovery{i}@example.com",
+                )
+                results.append(success)
+
+            # Should have both successes and failures
+            successful_count = sum(results)
+            failed_count = len(results) - successful_count
+
+            assert successful_count > 0, "No invitations succeeded"
+            assert failed_count > 0, "No invitations failed (test setup issue)"
+
+            # System should remain stable
+            assert successful_count >= failed_count, (
+                "More failures than successes indicates system instability"
+            )
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "-s"])  # -s to see print output

tests/test_invitation_unit.py

@@ -0,0 +1,469 @@
+"""
+Unit tests for invitation system core functionality.
+
+These tests focus on isolated testing of individual components
+without complex mocking or external dependencies.
+"""
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from app.extensions import db
+from app.models import Invitation, Library, MediaServer, User
+from app.services.invites import create_invite, is_invite_valid, mark_server_used
+
+
+class DictFormWrapper:
+    """Wrapper to make dict behave like a WTForm for testing."""
+
+    def __init__(self, data):
+        self.data = data
+
+    def get(self, key, default=None):
+        return self.data.get(key, default)
+
+    def getlist(self, key):
+        value = self.data.get(key, [])
+        if isinstance(value, list):
+            return value
+        return [value] if value is not None else []
+
+
+class TestInvitationValidation:
+    """Test invitation validation logic."""
+
+    def test_valid_invitation_basic(self, app):
+        """Test basic invitation validation."""
+        with app.app_context():
+            # Create server first
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation using form-like data
+            form_data = DictFormWrapper(
+                {"expires": "month", "unlimited": False, "server_ids": [str(server.id)]}
+            )
+            invite = create_invite(form_data)
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert is_valid
+            assert message == "okay"
+
+    def test_expired_invitation(self, app):
+        """Test that expired invitations are rejected."""
+        with app.app_context():
+            # Create invitation that expires immediately
+            invite = Invitation(
+                code="EXPIRED12",  # 9 chars, within limit
+                expires=datetime.now(UTC) - timedelta(hours=1),
+                used=False,
+                unlimited=False,
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert not is_valid
+            assert "expired" in message.lower()
+
+    def test_used_limited_invitation(self, app):
+        """Test that used limited invitations are rejected."""
+        with app.app_context():
+            # Create used limited invitation
+            invite = Invitation(code="USED123", used=True, unlimited=False)
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert not is_valid
+            assert "already been used" in message
+
+    def test_used_unlimited_invitation(self, app):
+        """Test that used unlimited invitations are still valid."""
+        with app.app_context():
+            # Create used unlimited invitation (code must be 6-10 chars)
+            invite = Invitation(
+                code="UNLIMIT123",  # 10 chars, within limit
+                used=True,
+                unlimited=True,
+            )
+            db.session.add(invite)
+            db.session.commit()
+
+            # Test validation
+            is_valid, message = is_invite_valid(invite.code)
+            assert is_valid
+            assert message == "okay"
+
+    def test_invalid_code_length(self, app):
+        """Test that codes with invalid length are rejected."""
+        with app.app_context():
+            # Test too short
+            is_valid, message = is_invite_valid("AB")
+            assert not is_valid
+            assert "Invalid code length" in message
+
+            # Test too long
+            is_valid, message = is_invite_valid("A" * 50)
+            assert not is_valid
+            assert "Invalid code length" in message
+
+    def test_nonexistent_code(self, app):
+        """Test that non-existent codes are rejected."""
+        with app.app_context():
+            is_valid, message = is_invite_valid(
+                "NOTEXIST12"
+            )  # 10 chars, valid length but nonexistent
+            assert not is_valid
+            assert "Invalid code" in message
+
+
+class TestInvitationCreation:
+    """Test invitation creation functionality."""
+
+    def test_create_basic_invitation(self, app):
+        """Test creating a basic invitation."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation
+            form_data = DictFormWrapper(
+                {
+                    "expires": "week",
+                    "unlimited": True,
+                    "server_ids": [str(server.id)],
+                    "duration": "14",
+                }
+            )
+            invite = create_invite(form_data)
+
+            # Verify invitation properties
+            assert invite.code is not None
+            assert len(invite.code) >= 3  # Minimum code size
+            assert invite.unlimited is True
+            assert invite.duration == "14"
+            assert invite.expires is not None
+            assert len(invite.servers) == 1
+            assert invite.servers[0] == server
+
+    def test_create_invitation_with_libraries(self, app):
+        """Test creating invitation with specific libraries."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create libraries
+            lib1 = Library(
+                name="Movies", external_id="lib1", server_id=server.id, enabled=True
+            )
+            lib2 = Library(
+                name="TV Shows", external_id="lib2", server_id=server.id, enabled=True
+            )
+            db.session.add_all([lib1, lib2])
+            db.session.flush()
+
+            # Create invitation with specific libraries
+            form_data = DictFormWrapper(
+                {
+                    "expires": "month",
+                    "unlimited": False,
+                    "server_ids": [str(server.id)],
+                    "libraries": [str(lib1.id), str(lib2.id)],
+                }
+            )
+            invite = create_invite(form_data)
+
+            # Verify library associations
+            assert len(invite.libraries) == 2
+            library_ids = {lib.id for lib in invite.libraries}
+            assert library_ids == {lib1.id, lib2.id}
+
+    def test_create_multi_server_invitation(self, app):
+        """Test creating invitation for multiple servers."""
+        with app.app_context():
+            # Create multiple servers
+            server1 = MediaServer(
+                name="Jellyfin Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="jellyfin-key",
+            )
+            server2 = MediaServer(
+                name="Plex Server",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="plex-key",
+            )
+            db.session.add_all([server1, server2])
+            db.session.flush()
+
+            # Create multi-server invitation
+            form_data = DictFormWrapper(
+                {
+                    "expires": "never",
+                    "unlimited": False,
+                    "server_ids": [str(server1.id), str(server2.id)],
+                }
+            )
+            invite = create_invite(form_data)
+
+            # Verify server associations
+            assert len(invite.servers) == 2
+            server_ids = {server.id for server in invite.servers}
+            assert server_ids == {server1.id, server2.id}
+
+    def test_create_invitation_validation_errors(self, app):
+        """Test invitation creation validation."""
+        with app.app_context():
+            # Test missing servers
+            form_data = DictFormWrapper(
+                {
+                    "expires": "month",
+                    "unlimited": False,
+                    "server_ids": [],  # No servers
+                }
+            )
+
+            with pytest.raises(
+                ValueError, match="At least one server must be selected"
+            ):
+                create_invite(form_data)
+
+    def test_create_invitation_with_custom_code(self, app):
+        """Test creating invitation with custom code."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create invitation with custom code
+            form_data = DictFormWrapper(
+                {
+                    "code": "CUSTOM123",
+                    "expires": "month",
+                    "unlimited": False,
+                    "server_ids": [str(server.id)],
+                }
+            )
+            invite = create_invite(form_data)
+
+            # Verify custom code
+            assert invite.code == "CUSTOM123"
+
+
+class TestInvitationMarkingUsed:
+    """Test invitation usage tracking."""
+
+    def test_mark_unlimited_invitation_used(self, app):
+        """Test marking unlimited invitation as used."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create unlimited invitation
+            form_data = DictFormWrapper(
+                {"expires": "month", "unlimited": True, "server_ids": [str(server.id)]}
+            )
+            invite = create_invite(form_data)
+
+            # Create user
+            user = User(
+                username="testuser",
+                email="test@example.com",
+                token="user-token",
+                code=invite.code,
+                server_id=server.id,
+            )
+            db.session.add(user)
+            db.session.commit()
+
+            # Verify invitation is not marked as used initially
+            assert invite.used is False
+
+            # Mark server as used
+            mark_server_used(invite, server.id, user)
+
+            # Verify unlimited invitation is marked as used for admin interface
+            db.session.refresh(invite)
+            assert (
+                invite.used is True
+            )  # Unlimited invitations get marked used for admin interface
+            assert invite.used_by == user
+
+            # But should still be valid for future use
+            is_valid, message = is_invite_valid(invite.code)
+            assert is_valid
+
+    def test_mark_limited_single_server_used(self, app):
+        """Test marking limited single-server invitation as used."""
+        with app.app_context():
+            # Create server
+            server = MediaServer(
+                name="Test Server",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="test-key",
+            )
+            db.session.add(server)
+            db.session.flush()
+
+            # Create limited invitation
+            form_data = DictFormWrapper(
+                {"expires": "month", "unlimited": False, "server_ids": [str(server.id)]}
+            )
+            invite = create_invite(form_data)
+
+            # Create user
+            user = User(
+                username="testuser",
+                email="test@example.com",
+                token="user-token",
+                code=invite.code,
+                server_id=server.id,
+            )
+            db.session.add(user)
+            db.session.commit()
+
+            # Mark server as used
+            mark_server_used(invite, server.id, user)
+
+            # Verify limited invitation is fully used (all servers used)
+            db.session.refresh(invite)
+            assert invite.used is True
+            assert invite.used_by == user
+
+            # Should no longer be valid
+            is_valid, message = is_invite_valid(invite.code)
+            assert not is_valid
+            assert "already been used" in message
+
+
+class TestInvitationRelationships:
+    """Test invitation model relationships."""
+
+    def test_invitation_user_relationship(self, app):
+        """Test invitation-user many-to-many relationship."""
+        with app.app_context():
+            # Create invitation
+            invite = Invitation(code="RELATION123", used=False, unlimited=True)
+            db.session.add(invite)
+            db.session.flush()
+
+            # Create users
+            user1 = User(
+                username="user1",
+                email="user1@example.com",
+                token="token1",
+                code="RELATION123",
+                server_id=1,
+            )
+            user2 = User(
+                username="user2",
+                email="user2@example.com",
+                token="token2",
+                code="RELATION123",
+                server_id=2,
+            )
+            db.session.add_all([user1, user2])
+            db.session.commit()
+
+            # Test helper methods
+            assert invite.get_user_count() == 0  # Not added to relationship yet
+
+            # Add users to invitation
+            invite.users.append(user1)
+            invite.users.append(user2)
+            db.session.commit()
+
+            # Test relationship methods
+            assert invite.get_user_count() == 2
+            assert invite.has_user(user1)
+            assert invite.has_user(user2)
+
+            all_users = invite.get_all_users()
+            assert len(all_users) == 2
+            assert user1 in all_users
+            assert user2 in all_users
+
+            first_user = invite.get_first_user()
+            assert first_user in [user1, user2]
+
+    def test_invitation_server_relationship(self, app):
+        """Test invitation-server many-to-many relationship."""
+        with app.app_context():
+            # Create servers
+            server1 = MediaServer(
+                name="Server 1",
+                server_type="jellyfin",
+                url="http://localhost:8096",
+                api_key="key1",
+            )
+            server2 = MediaServer(
+                name="Server 2",
+                server_type="plex",
+                url="http://localhost:32400",
+                api_key="key2",
+            )
+            db.session.add_all([server1, server2])
+            db.session.flush()
+
+            # Create invitation
+            form_data = DictFormWrapper(
+                {
+                    "expires": "month",
+                    "unlimited": False,
+                    "server_ids": [str(server1.id), str(server2.id)],
+                }
+            )
+            invite = create_invite(form_data)
+
+            # Verify server relationships
+            assert len(invite.servers) == 2
+            assert server1 in invite.servers
+            assert server2 in invite.servers
+
+            # Verify reverse relationship
+            assert invite in server1.invites
+            assert invite in server2.invites
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])