This commit is contained in:
DCCONSTRUCTIONS
2026-04-18 18:39:25 +03:00
commit 3ba092b60c
4944 changed files with 497564 additions and 0 deletions
+143
View File
@@ -0,0 +1,143 @@
# Plane Tests
This directory contains tests for the Plane application. The tests are organized using pytest.
## Test Structure
Tests are organized into the following categories:
- **Unit tests**: Test individual functions or classes in isolation.
- **Contract tests**: Test interactions between components and verify API contracts are fulfilled.
- **API tests**: Test the external API endpoints (under `/api/v1/`).
- **App tests**: Test the web application API endpoints (under `/api/`).
- **Smoke tests**: Basic tests to verify that the application runs correctly.
## API vs App Endpoints
Plane has two types of API endpoints:
1. **External API** (`plane.api`):
- Available at `/api/v1/` endpoint
- Uses API key authentication (X-Api-Key header)
- Designed for external API contracts and third-party access
- Tests use the `api_key_client` fixture for authentication
- Test files are in `contract/api/`
2. **Web App API** (`plane.app`):
- Available at `/api/` endpoint
- Uses session-based authentication (CSRF disabled)
- Designed for the web application frontend
- Tests use the `session_client` fixture for authentication
- Test files are in `contract/app/`
## Running Tests
To run all tests:
```bash
python -m pytest
```
To run specific test categories:
```bash
# Run unit tests
python -m pytest plane/tests/unit/
# Run API contract tests
python -m pytest plane/tests/contract/api/
# Run App contract tests
python -m pytest plane/tests/contract/app/
# Run smoke tests
python -m pytest plane/tests/smoke/
```
For convenience, we also provide a helper script:
```bash
# Run all tests
./run_tests.py
# Run only unit tests
./run_tests.py -u
# Run contract tests with coverage report
./run_tests.py -c -o
# Run tests in parallel
./run_tests.py -p
```
## Fixtures
The following fixtures are available for testing:
- `api_client`: Unauthenticated API client
- `create_user`: Creates a test user
- `api_token`: API token for the test user
- `api_key_client`: API client with API key authentication (for external API tests)
- `session_client`: API client with session authentication (for app API tests)
- `plane_server`: Live Django test server for HTTP-based smoke tests
## Writing Tests
When writing tests, follow these guidelines:
1. Place tests in the appropriate directory based on their type.
2. Use the correct client fixture based on the API being tested:
- For external API (`/api/v1/`), use `api_key_client`
- For web app API (`/api/`), use `session_client`
- For smoke tests with real HTTP, use `plane_server`
3. Use the correct URL namespace when reverse-resolving URLs:
- For external API, use `reverse("api:endpoint_name")`
- For web app API, use `reverse("endpoint_name")`
4. Add the `@pytest.mark.django_db` decorator to tests that interact with the database.
5. Add the appropriate markers (`@pytest.mark.contract`, etc.) to categorize tests.
## Test Fixtures
Common fixtures are defined in:
- `conftest.py`: General fixtures for authentication, database access, etc.
- `conftest_external.py`: Fixtures for external services (Redis, Elasticsearch, Celery, MongoDB)
- `factories.py`: Test factories for easy model instance creation
## Best Practices
When writing tests, follow these guidelines:
1. **Use pytest's assert syntax** instead of Django's `self.assert*` methods.
2. **Add markers to categorize tests**:
```python
@pytest.mark.unit
@pytest.mark.contract
@pytest.mark.smoke
```
3. **Use fixtures instead of setUp/tearDown methods** for cleaner, more reusable test code.
4. **Mock external dependencies** with the provided fixtures to avoid external service dependencies.
5. **Write focused tests** that verify one specific behavior or edge case.
6. **Keep test files small and organized** by logical components or endpoints.
7. **Target 90% code coverage** for models, serializers, and business logic.
## External Dependencies
Tests for components that interact with external services should:
1. Use the `mock_redis`, `mock_elasticsearch`, `mock_mongodb`, and `mock_celery` fixtures for unit and most contract tests.
2. For more comprehensive contract tests, use Docker-based test containers (optional).
## Coverage Reports
Generate a coverage report with:
```bash
python -m pytest --cov=plane --cov-report=term --cov-report=html
```
This creates an HTML report in the `htmlcov/` directory.
## Migration from Old Tests
Some tests are still in the old format in the `api/` directory. These need to be migrated to the new contract test structure in the appropriate directories.
@@ -0,0 +1,151 @@
# Testing Guide for Plane
This guide explains how to write tests for Plane using our pytest-based testing strategy.
## Test Categories
We divide tests into three categories:
1. **Unit Tests**: Testing individual components in isolation.
2. **Contract Tests**: Testing API endpoints and verifying contracts between components.
3. **Smoke Tests**: Basic end-to-end tests for critical flows.
## Writing Unit Tests
Unit tests should be placed in the appropriate directory under `tests/unit/` depending on what you're testing:
- `tests/unit/models/` - For model tests
- `tests/unit/serializers/` - For serializer tests
- `tests/unit/utils/` - For utility function tests
### Example Unit Test:
```python
import pytest
from plane.api.serializers import MySerializer
@pytest.mark.unit
class TestMySerializer:
def test_serializer_valid_data(self):
# Create input data
data = {"field1": "value1", "field2": 42}
# Initialize the serializer
serializer = MySerializer(data=data)
# Validate
assert serializer.is_valid()
# Check validated data
assert serializer.validated_data["field1"] == "value1"
assert serializer.validated_data["field2"] == 42
```
## Writing Contract Tests
Contract tests should be placed in `tests/contract/api/` or `tests/contract/app/` directories and should test the API endpoints.
### Example Contract Test:
```python
import pytest
from django.urls import reverse
from rest_framework import status
@pytest.mark.contract
class TestMyEndpoint:
@pytest.mark.django_db
def test_my_endpoint_get(self, auth_client):
# Get the URL
url = reverse("my-endpoint")
# Make request
response = auth_client.get(url)
# Check response
assert response.status_code == status.HTTP_200_OK
assert "data" in response.data
```
## Writing Smoke Tests
Smoke tests should be placed in `tests/smoke/` directory and use the `plane_server` fixture to test against a real HTTP server.
### Example Smoke Test:
```python
import pytest
import requests
@pytest.mark.smoke
class TestCriticalFlow:
@pytest.mark.django_db
def test_login_flow(self, plane_server, create_user, user_data):
# Get login URL
url = f"{plane_server.url}/api/auth/signin/"
# Test login
response = requests.post(
url,
json={
"email": user_data["email"],
"password": user_data["password"]
}
)
# Verify
assert response.status_code == 200
data = response.json()
assert "access_token" in data
```
## Useful Fixtures
Our test setup provides several useful fixtures:
1. `api_client`: An unauthenticated DRF APIClient
2. `api_key_client`: API client with API key authentication (for external API tests)
3. `session_client`: API client with session authentication (for web app API tests)
4. `create_user`: Creates and returns a test user
5. `mock_redis`: Mocks Redis interactions
6. `mock_elasticsearch`: Mocks Elasticsearch interactions
7. `mock_celery`: Mocks Celery task execution
## Using Factory Boy
For more complex test data setup, use the provided factories:
```python
from plane.tests.factories import UserFactory, WorkspaceFactory
# Create a user
user = UserFactory()
# Create a workspace with a specific owner
workspace = WorkspaceFactory(owner=user)
# Create multiple objects
users = UserFactory.create_batch(5)
```
## Running Tests
Use pytest to run tests:
```bash
# Run all tests
python -m pytest
# Run only unit tests with coverage
python -m pytest -m unit --cov=plane
```
## Best Practices
1. **Keep tests small and focused** - Each test should verify one specific behavior.
2. **Use markers** - Always add appropriate markers (`@pytest.mark.unit`, etc.).
3. **Mock external dependencies** - Use the provided mock fixtures.
4. **Use factories** - For complex data setup, use factories.
5. **Don't test the framework** - Focus on testing your business logic, not Django/DRF itself.
6. **Write readable assertions** - Use plain `assert` statements with clear messaging.
7. **Focus on coverage** - Aim for ≥90% code coverage for critical components.
@@ -0,0 +1,5 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
# Test package initialization
+9
View File
@@ -0,0 +1,9 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
from django.apps import AppConfig
class ApiConfig(AppConfig):
name = "plane.tests"
+140
View File
@@ -0,0 +1,140 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from rest_framework.test import APIClient
from pytest_django.fixtures import django_db_setup
from plane.db.models import User, Workspace, WorkspaceMember
from plane.db.models.api import APIToken
@pytest.fixture(scope="session")
def django_db_setup(django_db_setup): # noqa: F811
"""Set up the Django database for the test session"""
pass
@pytest.fixture
def api_client():
"""Return an unauthenticated API client"""
return APIClient()
@pytest.fixture
def user_data():
"""Return standard user data for tests"""
return {
"email": "test@plane.so",
"password": "test-password",
"first_name": "Test",
"last_name": "User",
}
@pytest.fixture
def create_user(db, user_data):
"""Create and return a user instance"""
user = User.objects.create(
email=user_data["email"],
first_name=user_data["first_name"],
last_name=user_data["last_name"],
)
user.set_password(user_data["password"])
user.save()
return user
@pytest.fixture
def api_token(db, create_user):
"""Create and return an API token for testing the external API"""
token = APIToken.objects.create(
user=create_user,
label="Test API Token",
token="test-api-token-12345",
)
return token
@pytest.fixture
def api_key_client(api_client, api_token):
"""Return an API key authenticated client for external API testing"""
api_client.credentials(HTTP_X_API_KEY=api_token.token)
return api_client
@pytest.fixture
def session_client(api_client, create_user):
"""Return a session authenticated API client for app API testing, which is what plane.app uses"""
api_client.force_authenticate(user=create_user)
return api_client
@pytest.fixture
def create_bot_user(db):
"""Create and return a bot user instance"""
from uuid import uuid4
unique_id = uuid4().hex[:8]
user = User.objects.create(
email=f"bot-{unique_id}@plane.so",
username=f"bot_user_{unique_id}",
first_name="Bot",
last_name="User",
is_bot=True,
)
user.set_password("bot@123")
user.save()
return user
@pytest.fixture
def api_token_data():
"""Return sample API token data for testing"""
from django.utils import timezone
from datetime import timedelta
return {
"label": "Test API Token",
"description": "Test description for API token",
"expired_at": (timezone.now() + timedelta(days=30)).isoformat(),
}
@pytest.fixture
def create_api_token_for_user(db, create_user):
"""Create and return an API token for a specific user"""
return APIToken.objects.create(
label="Test Token",
description="Test token description",
user=create_user,
user_type=0,
)
@pytest.fixture
def plane_server(live_server):
"""
Renamed version of live_server fixture to avoid name clashes.
Returns a live Django server for testing HTTP requests.
"""
return live_server
@pytest.fixture
def workspace(create_user):
"""
Create a new workspace and return the
corresponding Workspace model instance.
"""
# Create the workspace using the model
created_workspace = Workspace.objects.create(
name="Test Workspace",
owner=create_user,
slug="test-workspace",
)
WorkspaceMember.objects.create(workspace=created_workspace, member=create_user, role=20)
return created_workspace
@@ -0,0 +1,99 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from unittest.mock import MagicMock, patch
@pytest.fixture
def mock_redis():
"""
Mock Redis for testing without actual Redis connection.
This fixture patches the redis_instance function to return a MagicMock
that behaves like a Redis client.
"""
mock_redis_client = MagicMock()
# Configure the mock to handle common Redis operations
mock_redis_client.get.return_value = None
mock_redis_client.set.return_value = True
mock_redis_client.delete.return_value = True
mock_redis_client.exists.return_value = 0
mock_redis_client.ttl.return_value = -1
# Start the patch
with patch("plane.settings.redis.redis_instance", return_value=mock_redis_client):
yield mock_redis_client
@pytest.fixture
def mock_elasticsearch():
"""
Mock Elasticsearch for testing without actual ES connection.
This fixture patches Elasticsearch to return a MagicMock
that behaves like an Elasticsearch client.
"""
mock_es_client = MagicMock()
# Configure the mock to handle common ES operations
mock_es_client.indices.exists.return_value = True
mock_es_client.indices.create.return_value = {"acknowledged": True}
mock_es_client.search.return_value = {"hits": {"total": {"value": 0}, "hits": []}}
mock_es_client.index.return_value = {"_id": "test_id", "result": "created"}
mock_es_client.update.return_value = {"_id": "test_id", "result": "updated"}
mock_es_client.delete.return_value = {"_id": "test_id", "result": "deleted"}
# Start the patch
with patch("elasticsearch.Elasticsearch", return_value=mock_es_client):
yield mock_es_client
@pytest.fixture
def mock_mongodb():
"""
Mock MongoDB for testing without actual MongoDB connection.
This fixture patches PyMongo to return a MagicMock that behaves like a MongoDB client.
"""
# Create mock MongoDB clients and collections
mock_mongo_client = MagicMock()
mock_mongo_db = MagicMock()
mock_mongo_collection = MagicMock()
# Set up the chain: client -> database -> collection
mock_mongo_client.__getitem__.return_value = mock_mongo_db
mock_mongo_client.get_database.return_value = mock_mongo_db
mock_mongo_db.__getitem__.return_value = mock_mongo_collection
# Configure common MongoDB collection operations
mock_mongo_collection.find_one.return_value = None
mock_mongo_collection.find.return_value = MagicMock(__iter__=lambda x: iter([]), count=lambda: 0)
mock_mongo_collection.insert_one.return_value = MagicMock(inserted_id="mock_id_123", acknowledged=True)
mock_mongo_collection.insert_many.return_value = MagicMock(
inserted_ids=["mock_id_123", "mock_id_456"], acknowledged=True
)
mock_mongo_collection.update_one.return_value = MagicMock(modified_count=1, matched_count=1, acknowledged=True)
mock_mongo_collection.update_many.return_value = MagicMock(modified_count=2, matched_count=2, acknowledged=True)
mock_mongo_collection.delete_one.return_value = MagicMock(deleted_count=1, acknowledged=True)
mock_mongo_collection.delete_many.return_value = MagicMock(deleted_count=2, acknowledged=True)
mock_mongo_collection.count_documents.return_value = 0
# Start the patch
with patch("pymongo.MongoClient", return_value=mock_mongo_client):
yield mock_mongo_client
@pytest.fixture
def mock_celery():
"""
Mock Celery for testing without actual task execution.
This fixture patches Celery's task.delay() to prevent actual task execution.
"""
# Start the patch
with patch("celery.app.task.Task.delay") as mock_delay:
mock_delay.return_value = MagicMock(id="mock-task-id")
yield mock_delay
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,384 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from rest_framework import status
from django.utils import timezone
from datetime import timedelta
from uuid import uuid4
from plane.db.models import Cycle, Project, ProjectMember
@pytest.fixture
def project(db, workspace, create_user):
"""Create a test project with the user as a member"""
project = Project.objects.create(
name="Test Project",
identifier="TP",
workspace=workspace,
created_by=create_user,
)
ProjectMember.objects.create(
project=project,
member=create_user,
role=20, # Admin role
is_active=True,
)
return project
@pytest.fixture
def cycle_data():
"""Sample cycle data for tests"""
return {
"name": "Test Cycle",
"description": "A test cycle for unit tests",
}
@pytest.fixture
def draft_cycle_data():
"""Sample draft cycle data (no dates)"""
return {
"name": "Draft Cycle",
"description": "A draft cycle without dates",
}
@pytest.fixture
def create_cycle(db, project, create_user):
"""Create a test cycle"""
return Cycle.objects.create(
name="Existing Cycle",
description="An existing cycle",
start_date=timezone.now() + timedelta(days=1),
end_date=timezone.now() + timedelta(days=7),
project=project,
workspace=project.workspace,
owned_by=create_user,
)
@pytest.mark.contract
class TestCycleListCreateAPIEndpoint:
"""Test Cycle List and Create API Endpoint"""
def get_cycle_url(self, workspace_slug, project_id):
"""Helper to get cycle endpoint URL"""
return f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/cycles/"
@pytest.mark.django_db
def test_create_cycle_success(self, api_key_client, workspace, project, cycle_data):
"""Test successful cycle creation"""
url = self.get_cycle_url(workspace.slug, project.id)
response = api_key_client.post(url, cycle_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
assert Cycle.objects.count() == 1
created_cycle = Cycle.objects.first()
assert created_cycle.name == cycle_data["name"]
assert created_cycle.description == cycle_data["description"]
assert created_cycle.project == project
assert created_cycle.owned_by_id is not None
@pytest.mark.django_db
def test_create_cycle_invalid_data(self, api_key_client, workspace, project):
"""Test cycle creation with invalid data"""
url = self.get_cycle_url(workspace.slug, project.id)
# Test with empty data
response = api_key_client.post(url, {}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
# Test with missing name
response = api_key_client.post(url, {"description": "Test cycle"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_cycle_invalid_date_combination(self, api_key_client, workspace, project):
"""Test cycle creation with invalid date combination (only start_date)"""
url = self.get_cycle_url(workspace.slug, project.id)
invalid_data = {
"name": "Invalid Cycle",
"start_date": (timezone.now() + timedelta(days=1)).isoformat(),
# Missing end_date
}
response = api_key_client.post(url, invalid_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "Both start date and end date are either required or are to be null" in response.data["error"]
@pytest.mark.django_db
def test_create_cycle_with_external_id(self, api_key_client, workspace, project):
"""Test creating cycle with external ID"""
url = self.get_cycle_url(workspace.slug, project.id)
cycle_data = {
"name": "External Cycle",
"description": "A cycle with external ID",
"external_id": "ext-123",
"external_source": "github",
}
response = api_key_client.post(url, cycle_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
created_cycle = Cycle.objects.first()
assert created_cycle.external_id == "ext-123"
assert created_cycle.external_source == "github"
@pytest.mark.django_db
def test_create_cycle_duplicate_external_id(self, api_key_client, workspace, project, create_user):
"""Test creating cycle with duplicate external ID"""
url = self.get_cycle_url(workspace.slug, project.id)
# Create first cycle
Cycle.objects.create(
name="First Cycle",
project=project,
workspace=workspace,
external_id="ext-123",
external_source="github",
owned_by=create_user,
)
# Try to create second cycle with same external ID
cycle_data = {
"name": "Second Cycle",
"external_id": "ext-123",
"external_source": "github",
"owned_by": create_user.id,
}
response = api_key_client.post(url, cycle_data, format="json")
assert response.status_code == status.HTTP_409_CONFLICT
assert "same external id" in response.data["error"]
@pytest.mark.django_db
def test_list_cycles_success(self, api_key_client, workspace, project, create_cycle, create_user):
"""Test successful cycle listing"""
url = self.get_cycle_url(workspace.slug, project.id)
# Create additional cycles
Cycle.objects.create(
name="Cycle 2",
project=project,
workspace=workspace,
start_date=timezone.now() + timedelta(days=10),
end_date=timezone.now() + timedelta(days=17),
owned_by=create_user,
)
Cycle.objects.create(
name="Cycle 3",
project=project,
workspace=workspace,
start_date=timezone.now() + timedelta(days=20),
end_date=timezone.now() + timedelta(days=27),
owned_by=create_user,
)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert "results" in response.data
assert len(response.data["results"]) == 3 # Including create_cycle fixture
@pytest.mark.django_db
def test_list_cycles_with_view_filter(self, api_key_client, workspace, project, create_user):
"""Test cycle listing with different view filters"""
url = self.get_cycle_url(workspace.slug, project.id)
# Create cycles in different states
now = timezone.now()
# Current cycle (started but not ended)
Cycle.objects.create(
name="Current Cycle",
project=project,
workspace=workspace,
start_date=now - timedelta(days=1),
end_date=now + timedelta(days=6),
owned_by=create_user,
)
# Upcoming cycle
Cycle.objects.create(
name="Upcoming Cycle",
project=project,
workspace=workspace,
start_date=now + timedelta(days=1),
end_date=now + timedelta(days=8),
owned_by=create_user,
)
# Completed cycle
Cycle.objects.create(
name="Completed Cycle",
project=project,
workspace=workspace,
start_date=now - timedelta(days=10),
end_date=now - timedelta(days=3),
owned_by=create_user,
)
# Draft cycle
Cycle.objects.create(
name="Draft Cycle",
project=project,
workspace=workspace,
owned_by=create_user,
)
# Test current cycles
response = api_key_client.get(url, {"cycle_view": "current"})
assert response.status_code == status.HTTP_200_OK
assert len(response.data) == 1
assert response.data[0]["name"] == "Current Cycle"
# Test upcoming cycles
response = api_key_client.get(url, {"cycle_view": "upcoming"})
assert response.status_code == status.HTTP_200_OK
assert len(response.data["results"]) == 1
assert response.data["results"][0]["name"] == "Upcoming Cycle"
# Test completed cycles
response = api_key_client.get(url, {"cycle_view": "completed"})
assert response.status_code == status.HTTP_200_OK
assert len(response.data["results"]) == 1
assert response.data["results"][0]["name"] == "Completed Cycle"
# Test draft cycles
response = api_key_client.get(url, {"cycle_view": "draft"})
assert response.status_code == status.HTTP_200_OK
assert len(response.data["results"]) == 1
assert response.data["results"][0]["name"] == "Draft Cycle"
@pytest.mark.contract
class TestCycleDetailAPIEndpoint:
"""Test Cycle Detail API Endpoint"""
def get_cycle_detail_url(self, workspace_slug, project_id, cycle_id):
"""Helper to get cycle detail endpoint URL"""
return f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/cycles/{cycle_id}/"
@pytest.mark.django_db
def test_get_cycle_success(self, api_key_client, workspace, project, create_cycle):
"""Test successful cycle retrieval"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert str(response.data["id"]) == str(create_cycle.id)
assert response.data["name"] == create_cycle.name
assert response.data["description"] == create_cycle.description
@pytest.mark.django_db
def test_get_cycle_not_found(self, api_key_client, workspace, project):
"""Test getting non-existent cycle"""
fake_id = uuid4()
url = self.get_cycle_detail_url(workspace.slug, project.id, fake_id)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_update_cycle_success(self, api_key_client, workspace, project, create_cycle):
"""Test successful cycle update"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
update_data = {
"name": f"Updated Cycle {uuid4()}",
"description": "Updated description",
}
response = api_key_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_200_OK
create_cycle.refresh_from_db()
assert create_cycle.name == update_data["name"]
assert create_cycle.description == update_data["description"]
@pytest.mark.django_db
def test_update_cycle_invalid_data(self, api_key_client, workspace, project, create_cycle):
"""Test cycle update with invalid data"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
update_data = {"name": ""}
response = api_key_client.patch(url, update_data, format="json")
# This might be 400 if name is required, or 200 if empty names are allowed
assert response.status_code in [status.HTTP_400_BAD_REQUEST, status.HTTP_200_OK]
@pytest.mark.django_db
def test_update_cycle_with_external_id_conflict(
self, api_key_client, workspace, project, create_cycle, create_user
):
"""Test cycle update with conflicting external ID"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
# Create another cycle with external ID
Cycle.objects.create(
name="Another Cycle",
project=project,
workspace=workspace,
external_id="ext-456",
external_source="github",
owned_by=create_user,
)
# Try to update cycle with same external ID
update_data = {
"external_id": "ext-456",
"external_source": "github",
}
response = api_key_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_409_CONFLICT
assert "same external id" in response.data["error"]
@pytest.mark.django_db
def test_delete_cycle_success(self, api_key_client, workspace, project, create_cycle):
"""Test successful cycle deletion"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
response = api_key_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Cycle.objects.filter(id=create_cycle.id).exists()
@pytest.mark.django_db
def test_cycle_metrics_annotation(self, api_key_client, workspace, project, create_cycle):
"""Test that cycle includes issue metrics annotations"""
url = self.get_cycle_detail_url(workspace.slug, project.id, create_cycle.id)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_200_OK
# Check that metrics are included in response
cycle_data = response.data
assert "total_issues" in cycle_data
assert "completed_issues" in cycle_data
assert "cancelled_issues" in cycle_data
assert "started_issues" in cycle_data
assert "unstarted_issues" in cycle_data
assert "backlog_issues" in cycle_data
# All should be 0 for a new cycle
assert cycle_data["total_issues"] == 0
assert cycle_data["completed_issues"] == 0
assert cycle_data["cancelled_issues"] == 0
assert cycle_data["started_issues"] == 0
assert cycle_data["unstarted_issues"] == 0
assert cycle_data["backlog_issues"] == 0
@@ -0,0 +1,217 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from rest_framework import status
from uuid import uuid4
from plane.db.models import Label, Project, ProjectMember
@pytest.fixture
def project(db, workspace, create_user):
"""Create a test project with the user as a member"""
project = Project.objects.create(
name="Test Project",
identifier="TP",
workspace=workspace,
created_by=create_user,
)
ProjectMember.objects.create(
project=project,
member=create_user,
role=20, # Admin role
is_active=True,
)
return project
@pytest.fixture
def label_data():
"""Sample label data for tests"""
return {
"name": "Test Label",
"color": "#FF5733",
"description": "A test label for unit tests",
}
@pytest.fixture
def create_label(db, project, create_user):
"""Create a test label"""
return Label.objects.create(
name="Existing Label",
color="#00FF00",
description="An existing label",
project=project,
workspace=project.workspace,
created_by=create_user,
)
@pytest.mark.contract
class TestLabelListCreateAPIEndpoint:
"""Test Label List and Create API Endpoint"""
def get_label_url(self, workspace_slug, project_id):
"""Helper to get label endpoint URL"""
return f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/labels/"
@pytest.mark.django_db
def test_create_label_success(self, api_key_client, workspace, project, label_data):
"""Test successful label creation"""
url = self.get_label_url(workspace.slug, project.id)
response = api_key_client.post(url, label_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
assert Label.objects.count() == 1
created_label = Label.objects.first()
assert created_label.name == label_data["name"]
assert created_label.color == label_data["color"]
assert created_label.description == label_data["description"]
assert created_label.project == project
@pytest.mark.django_db
def test_create_label_invalid_data(self, api_key_client, workspace, project):
"""Test label creation with invalid data"""
url = self.get_label_url(workspace.slug, project.id)
# Test with empty data
response = api_key_client.post(url, {}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
# Test with missing name
response = api_key_client.post(url, {"color": "#FF5733"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_label_with_external_id(self, api_key_client, workspace, project):
"""Test creating label with external ID"""
url = self.get_label_url(workspace.slug, project.id)
label_data = {
"name": "External Label",
"color": "#FF5733",
"external_id": "ext-123",
"external_source": "github",
}
response = api_key_client.post(url, label_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
created_label = Label.objects.first()
assert created_label.external_id == "ext-123"
assert created_label.external_source == "github"
@pytest.mark.django_db
def test_create_label_duplicate_external_id(self, api_key_client, workspace, project):
"""Test creating label with duplicate external ID"""
url = self.get_label_url(workspace.slug, project.id)
# Create first label
Label.objects.create(
name="First Label",
project=project,
workspace=workspace,
external_id="ext-123",
external_source="github",
)
# Try to create second label with same external ID
label_data = {
"name": "Second Label",
"external_id": "ext-123",
"external_source": "github",
}
response = api_key_client.post(url, label_data, format="json")
assert response.status_code == status.HTTP_409_CONFLICT
assert "same external id" in response.data["error"]
@pytest.mark.django_db
def test_list_labels_success(self, api_key_client, workspace, project, create_label):
"""Test successful label listing"""
url = self.get_label_url(workspace.slug, project.id)
# Create additional labels
Label.objects.create(name="Label 2", project=project, workspace=workspace, color="#00FF00")
Label.objects.create(name="Label 3", project=project, workspace=workspace, color="#0000FF")
response = api_key_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert "results" in response.data
assert len(response.data["results"]) == 3 # Including create_label fixture
@pytest.mark.contract
class TestLabelDetailAPIEndpoint:
"""Test Label Detail API Endpoint"""
def get_label_detail_url(self, workspace_slug, project_id, label_id):
"""Helper to get label detail endpoint URL"""
return f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/labels/{label_id}/"
@pytest.mark.django_db
def test_get_label_success(self, api_key_client, workspace, project, create_label):
"""Test successful label retrieval"""
url = self.get_label_detail_url(workspace.slug, project.id, create_label.id)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data["id"] == create_label.id
assert response.data["name"] == create_label.name
assert response.data["color"] == create_label.color
@pytest.mark.django_db
def test_get_label_not_found(self, api_key_client, workspace, project):
"""Test getting non-existent label"""
from uuid import uuid4
fake_id = uuid4()
url = self.get_label_detail_url(workspace.slug, project.id, fake_id)
response = api_key_client.get(url)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_update_label_success(self, api_key_client, workspace, project, create_label):
"""Test successful label update"""
url = self.get_label_detail_url(workspace.slug, project.id, create_label.id)
update_data = {
"name": f"Updated Label {uuid4()}",
}
response = api_key_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_200_OK
create_label.refresh_from_db()
assert create_label.name == update_data["name"]
@pytest.mark.django_db
def test_update_label_invalid_data(self, api_key_client, workspace, project, create_label):
"""Test label update with invalid data"""
url = self.get_label_detail_url(workspace.slug, project.id, create_label.id)
update_data = {"name": ""}
response = api_key_client.patch(url, update_data, format="json")
# This might be 400 if name is required, or 200 if empty names are allowed
assert response.status_code in [status.HTTP_400_BAD_REQUEST, status.HTTP_200_OK]
@pytest.mark.django_db
def test_delete_label_success(self, api_key_client, workspace, project, create_label):
"""Test successful label deletion"""
url = self.get_label_detail_url(workspace.slug, project.id, create_label.id)
response = api_key_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Label.objects.filter(id=create_label.id).exists()
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,402 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from datetime import timedelta
from uuid import uuid4
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from plane.db.models import APIToken, User
@pytest.mark.contract
class TestApiTokenEndpoint:
"""Test cases for ApiTokenEndpoint"""
# POST /user/api-tokens/ tests
@pytest.mark.django_db
def test_create_api_token_success(self, session_client, create_user, api_token_data):
"""Test successful API token creation"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert "token" in response.data
assert response.data["label"] == api_token_data["label"]
assert response.data["description"] == api_token_data["description"]
assert response.data["user_type"] == 0 # Human user
# Verify token was created in database
token = APIToken.objects.get(pk=response.data["id"])
assert token.user == create_user
assert token.label == api_token_data["label"]
@pytest.mark.django_db
def test_create_api_token_for_bot_user(self, session_client, create_bot_user, api_token_data):
"""Test API token creation for bot user"""
# Arrange
session_client.force_authenticate(user=create_bot_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert response.data["user_type"] == 1 # Bot user
@pytest.mark.django_db
def test_create_api_token_minimal_data(self, session_client, create_user):
"""Test API token creation with minimal data"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, {}, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert "token" in response.data
assert len(response.data["label"]) == 32 # UUID hex length
assert response.data["description"] == ""
@pytest.mark.django_db
def test_create_api_token_with_expiry(self, session_client, create_user):
"""Test API token creation with expiry date"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
future_date = timezone.now() + timedelta(days=30)
data = {"label": "Expiring Token", "expired_at": future_date.isoformat()}
# Act
response = session_client.post(url, data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
# Verify expiry date was set
token = APIToken.objects.get(pk=response.data["id"])
assert token.expired_at is not None
@pytest.mark.django_db
def test_create_api_token_unauthenticated(self, api_client, api_token_data):
"""Test API token creation without authentication"""
# Arrange
url = reverse("api-tokens")
# Act
response = api_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_401_UNAUTHORIZED
# GET /user/api-tokens/ tests
@pytest.mark.django_db
def test_get_all_api_tokens(self, session_client, create_user):
"""Test retrieving all API tokens for user"""
# Arrange
session_client.force_authenticate(user=create_user)
# Create multiple tokens
APIToken.objects.create(label="Token 1", user=create_user, user_type=0)
APIToken.objects.create(label="Token 2", user=create_user, user_type=0)
# Create a service token (should be excluded)
APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
url = reverse("api-tokens")
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert len(response.data) == 2 # Only non-service tokens
assert all(token["is_service"] is False for token in response.data)
@pytest.mark.django_db
def test_get_empty_api_tokens_list(self, session_client, create_user):
"""Test retrieving API tokens when none exist"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data == []
# GET /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_get_specific_api_token(self, session_client, create_user, create_api_token_for_user):
"""Test retrieving a specific API token"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert str(response.data["id"]) == str(create_api_token_for_user.pk)
assert response.data["label"] == create_api_token_for_user.label
assert "token" not in response.data # Token should not be visible in read serializer
@pytest.mark.django_db
def test_get_nonexistent_api_token(self, session_client, create_user):
"""Test retrieving a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_get_other_users_api_token(self, session_client, create_user, db):
"""Test retrieving another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"other-{unique_id}@plane.so"
unique_username = f"other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# DELETE /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_delete_api_token_success(self, session_client, create_user, create_api_token_for_user):
"""Test successful API token deletion"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not APIToken.objects.filter(pk=create_api_token_for_user.pk).exists()
@pytest.mark.django_db
def test_delete_nonexistent_api_token(self, session_client, create_user):
"""Test deleting a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_delete_other_users_api_token(self, session_client, create_user, db):
"""Test deleting another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"delete-other-{unique_id}@plane.so"
unique_username = f"delete_other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token still exists
assert APIToken.objects.filter(pk=other_token.pk).exists()
@pytest.mark.django_db
def test_delete_service_api_token_forbidden(self, session_client, create_user):
"""Test deleting a service API token (should fail)"""
# Arrange
service_token = APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": service_token.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token still exists
assert APIToken.objects.filter(pk=service_token.pk).exists()
# PATCH /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_patch_api_token_success(self, session_client, create_user, create_api_token_for_user):
"""Test successful API token update"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
update_data = {
"label": "Updated Token Label",
"description": "Updated description",
}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data["label"] == update_data["label"]
assert response.data["description"] == update_data["description"]
# Verify database was updated
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.label == update_data["label"]
assert create_api_token_for_user.description == update_data["description"]
@pytest.mark.django_db
def test_patch_api_token_partial_update(self, session_client, create_user, create_api_token_for_user):
"""Test partial API token update"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
original_description = create_api_token_for_user.description
update_data = {"label": "Only Label Updated"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data["label"] == update_data["label"]
assert response.data["description"] == original_description
@pytest.mark.django_db
def test_patch_nonexistent_api_token(self, session_client, create_user):
"""Test updating a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
update_data = {"label": "New Label"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_patch_other_users_api_token(self, session_client, create_user, db):
"""Test updating another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"patch-other-{unique_id}@plane.so"
unique_username = f"patch_other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
update_data = {"label": "Hacked Label"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token was not updated
other_token.refresh_from_db()
assert other_token.label == "Other Token"
@pytest.mark.django_db
def test_patch_cannot_modify_token(self, session_client, create_user, create_api_token_for_user):
"""Test that token value cannot be modified via PATCH"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
original_token = create_api_token_for_user.token
update_data = {"token": "plane_api_malicious_token_value"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.token == original_token
@pytest.mark.django_db
def test_patch_cannot_modify_user_type(self, session_client, create_user, create_api_token_for_user):
"""Test that user_type cannot be modified via PATCH"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
update_data = {"user_type": 1}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.user_type == 0
@pytest.mark.django_db
def test_patch_cannot_modify_service_token(self, session_client, create_user):
"""Test that service tokens cannot be modified through user token endpoint"""
# Arrange
service_token = APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": service_token.pk})
update_data = {"label": "Hacked Service Token"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
service_token.refresh_from_db()
assert service_token.label == "Service Token"
# Authentication tests
@pytest.mark.django_db
def test_all_endpoints_require_authentication(self, api_client):
"""Test that all endpoints require authentication"""
# Arrange
endpoints = [
(reverse("api-tokens"), "get"),
(reverse("api-tokens"), "post"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "get"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "patch"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "delete"),
]
# Act & Assert
for url, method in endpoints:
response = getattr(api_client, method)(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@@ -0,0 +1,429 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import json
import uuid
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from django.test import Client
from django.core.exceptions import ValidationError
from unittest.mock import patch
from plane.db.models import User
from plane.settings.redis import redis_instance
from plane.license.models import Instance
@pytest.fixture
def setup_instance(db):
"""Create and configure an instance for authentication tests"""
instance_id = uuid.uuid4() if not Instance.objects.exists() else Instance.objects.first().id
# Create or update instance with all required fields
instance, _ = Instance.objects.update_or_create(
id=instance_id,
defaults={
"instance_name": "Test Instance",
"instance_id": str(uuid.uuid4()),
"current_version": "1.0.0",
"domain": "http://localhost:8000",
"last_checked_at": timezone.now(),
"is_setup_done": True,
},
)
return instance
@pytest.fixture
def django_client():
"""Return a Django test client with User-Agent header for handling redirects"""
client = Client(HTTP_USER_AGENT="Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) Gecko/20100101 Firefox/15.0.1")
return client
@pytest.mark.contract
class TestMagicLinkGenerate:
"""Test magic link generation functionality"""
@pytest.fixture
def setup_user(self, db):
"""Create a test user for magic link tests"""
user = User.objects.create(email="user@plane.so")
user.set_password("user@123")
user.save()
return user
@pytest.mark.django_db
def test_without_data(self, api_client, setup_user, setup_instance):
"""Test magic link generation with empty data"""
url = reverse("magic-generate")
try:
response = api_client.post(url, {}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
except ValidationError:
# If a ValidationError is raised directly, that's also acceptable
# as it indicates the empty email was rejected
assert True
@pytest.mark.django_db
def test_email_validity(self, api_client, setup_user, setup_instance):
"""Test magic link generation with invalid email format"""
url = reverse("magic-generate")
try:
response = api_client.post(url, {"email": "useremail.com"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "error_code" in response.data # Check for error code in response
except ValidationError:
# If a ValidationError is raised directly, that's also acceptable
# as it indicates the invalid email was rejected
assert True
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_magic_generate(self, mock_magic_link, api_client, setup_user, setup_instance):
"""Test successful magic link generation"""
url = reverse("magic-generate")
ri = redis_instance()
ri.delete("magic_user@plane.so")
response = api_client.post(url, {"email": "user@plane.so"}, format="json")
assert response.status_code == status.HTTP_200_OK
assert "key" in response.data # Check for key in response
# Verify the mock was called with the expected arguments
mock_magic_link.assert_called_once()
args = mock_magic_link.call_args[0]
assert args[0] == "user@plane.so" # First arg should be the email
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_max_generate_attempt(self, mock_magic_link, api_client, setup_user, setup_instance):
"""Test exceeding maximum magic link generation attempts"""
url = reverse("magic-generate")
ri = redis_instance()
ri.delete("magic_user@plane.so")
for _ in range(4):
api_client.post(url, {"email": "user@plane.so"}, format="json")
response = api_client.post(url, {"email": "user@plane.so"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "error_code" in response.data # Check for error code in response
@pytest.mark.contract
class TestSignInEndpoint:
"""Test sign-in functionality"""
@pytest.fixture
def setup_user(self, db):
"""Create a test user for authentication tests"""
user = User.objects.create(email="user@plane.so")
user.set_password("user@123")
user.save()
return user
@pytest.mark.django_db
def test_without_data(self, django_client, setup_user, setup_instance):
"""Test sign-in with empty data"""
url = reverse("sign-in")
response = django_client.post(url, {}, follow=True)
# Check redirect contains error code
assert "REQUIRED_EMAIL_PASSWORD_SIGN_IN" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_email_validity(self, django_client, setup_user, setup_instance):
"""Test sign-in with invalid email format"""
url = reverse("sign-in")
response = django_client.post(url, {"email": "useremail.com", "password": "user@123"}, follow=True)
# Check redirect contains error code
assert "INVALID_EMAIL_SIGN_IN" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_user_exists(self, django_client, setup_user, setup_instance):
"""Test sign-in with non-existent user"""
url = reverse("sign-in")
response = django_client.post(url, {"email": "user@email.so", "password": "user123"}, follow=True)
# Check redirect contains error code
assert "USER_DOES_NOT_EXIST" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_password_validity(self, django_client, setup_user, setup_instance):
"""Test sign-in with incorrect password"""
url = reverse("sign-in")
response = django_client.post(url, {"email": "user@plane.so", "password": "user123"}, follow=True)
# Check for the specific authentication error in the URL
redirect_urls = [url for url, _ in response.redirect_chain]
redirect_contents = " ".join(redirect_urls)
# The actual error code for invalid password is AUTHENTICATION_FAILED_SIGN_IN
assert "AUTHENTICATION_FAILED_SIGN_IN" in redirect_contents
@pytest.mark.django_db
def test_user_login(self, django_client, setup_user, setup_instance):
"""Test successful sign-in"""
url = reverse("sign-in")
# First make the request without following redirects
response = django_client.post(url, {"email": "user@plane.so", "password": "user@123"}, follow=False)
# Check that the initial response is a redirect (302) without error code
assert response.status_code == 302
assert "error_code" not in response.url
# Now follow just the first redirect to avoid 404s
response = django_client.get(response.url, follow=False)
# The user should be authenticated regardless of the final page
assert "_auth_user_id" in django_client.session
@pytest.mark.django_db
def test_next_path_redirection(self, django_client, setup_user, setup_instance):
"""Test sign-in with next_path parameter"""
url = reverse("sign-in")
next_path = "workspaces"
# First make the request without following redirects
response = django_client.post(
url,
{"email": "user@plane.so", "password": "user@123", "next_path": next_path},
follow=False,
)
# Check that the initial response is a redirect (302) without error code
assert response.status_code == 302
assert "error_code" not in response.url
# In a real browser, the next_path would be used to build the absolute URL
# Since we're just testing the authentication logic, we won't check for the exact URL structure
# Instead, just verify that we're authenticated
assert "_auth_user_id" in django_client.session
@pytest.mark.contract
class TestMagicSignIn:
"""Test magic link sign-in functionality"""
@pytest.fixture
def setup_user(self, db):
"""Create a test user for magic sign-in tests"""
user = User.objects.create(email="user@plane.so")
user.set_password("user@123")
user.save()
return user
@pytest.mark.django_db
def test_without_data(self, django_client, setup_user, setup_instance):
"""Test magic link sign-in with empty data"""
url = reverse("magic-sign-in")
response = django_client.post(url, {}, follow=True)
# Check redirect contains error code
assert "MAGIC_SIGN_IN_EMAIL_CODE_REQUIRED" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_expired_invalid_magic_link(self, django_client, setup_user, setup_instance):
"""Test magic link sign-in with expired/invalid link"""
ri = redis_instance()
ri.delete("magic_user@plane.so")
url = reverse("magic-sign-in")
response = django_client.post(url, {"email": "user@plane.so", "code": "xxxx-xxxxx-xxxx"}, follow=False)
# Check that we get a redirect
assert response.status_code == 302
# The actual error code is EXPIRED_MAGIC_CODE_SIGN_IN (when key doesn't exist)
# or INVALID_MAGIC_CODE_SIGN_IN (when key exists but code doesn't match)
assert "EXPIRED_MAGIC_CODE_SIGN_IN" in response.url or "INVALID_MAGIC_CODE_SIGN_IN" in response.url
@pytest.mark.django_db
def test_user_does_not_exist(self, django_client, setup_instance):
"""Test magic sign-in with non-existent user"""
url = reverse("magic-sign-in")
response = django_client.post(
url,
{"email": "nonexistent@plane.so", "code": "xxxx-xxxxx-xxxx"},
follow=True,
)
# Check redirect contains error code
assert "USER_DOES_NOT_EXIST" in response.redirect_chain[-1][0]
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_magic_code_sign_in(self, mock_magic_link, django_client, api_client, setup_user, setup_instance):
"""Test successful magic link sign-in process"""
# First generate a magic link token
gen_url = reverse("magic-generate")
response = api_client.post(gen_url, {"email": "user@plane.so"}, format="json")
# Check that the token generation was successful
assert response.status_code == status.HTTP_200_OK
# Since we're mocking the magic_link task, we need to manually get the token from Redis
ri = redis_instance()
user_data = json.loads(ri.get("magic_user@plane.so"))
token = user_data["token"]
# Use Django client to test the redirect flow without following redirects
url = reverse("magic-sign-in")
response = django_client.post(url, {"email": "user@plane.so", "code": token}, follow=False)
# Check that the initial response is a redirect without error code
assert response.status_code == 302
assert "error_code" not in response.url
# The user should now be authenticated
assert "_auth_user_id" in django_client.session
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_magic_sign_in_with_next_path(self, mock_magic_link, django_client, api_client, setup_user, setup_instance):
"""Test magic sign-in with next_path parameter"""
# First generate a magic link token
gen_url = reverse("magic-generate")
response = api_client.post(gen_url, {"email": "user@plane.so"}, format="json")
# Check that the token generation was successful
assert response.status_code == status.HTTP_200_OK
# Since we're mocking the magic_link task, we need to manually get the token from Redis
ri = redis_instance()
user_data = json.loads(ri.get("magic_user@plane.so"))
token = user_data["token"]
# Use Django client to test the redirect flow without following redirects
url = reverse("magic-sign-in")
next_path = "workspaces"
response = django_client.post(
url,
{"email": "user@plane.so", "code": token, "next_path": next_path},
follow=False,
)
# Check that the initial response is a redirect without error code
assert response.status_code == 302
assert "error_code" not in response.url
# Check that the redirect URL contains the next_path
assert next_path in response.url
# The user should now be authenticated
assert "_auth_user_id" in django_client.session
@pytest.mark.contract
class TestMagicSignUp:
"""Test magic link sign-up functionality"""
@pytest.mark.django_db
def test_without_data(self, django_client, setup_instance):
"""Test magic link sign-up with empty data"""
url = reverse("magic-sign-up")
response = django_client.post(url, {}, follow=True)
# Check redirect contains error code
assert "MAGIC_SIGN_UP_EMAIL_CODE_REQUIRED" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_user_already_exists(self, django_client, db, setup_instance):
"""Test magic sign-up with existing user"""
# Create a user that already exists
User.objects.create(email="existing@plane.so")
url = reverse("magic-sign-up")
response = django_client.post(url, {"email": "existing@plane.so", "code": "xxxx-xxxxx-xxxx"}, follow=True)
# Check redirect contains error code
assert "USER_ALREADY_EXIST" in response.redirect_chain[-1][0]
@pytest.mark.django_db
def test_expired_invalid_magic_link(self, django_client, setup_instance):
"""Test magic link sign-up with expired/invalid link"""
url = reverse("magic-sign-up")
response = django_client.post(url, {"email": "new@plane.so", "code": "xxxx-xxxxx-xxxx"}, follow=False)
# Check that we get a redirect
assert response.status_code == 302
# The actual error code is EXPIRED_MAGIC_CODE_SIGN_UP (when key doesn't exist)
# or INVALID_MAGIC_CODE_SIGN_UP (when key exists but code doesn't match)
assert "EXPIRED_MAGIC_CODE_SIGN_UP" in response.url or "INVALID_MAGIC_CODE_SIGN_UP" in response.url
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_magic_code_sign_up(self, mock_magic_link, django_client, api_client, setup_instance):
"""Test successful magic link sign-up process"""
email = "newuser@plane.so"
# First generate a magic link token
gen_url = reverse("magic-generate")
response = api_client.post(gen_url, {"email": email}, format="json")
# Check that the token generation was successful
assert response.status_code == status.HTTP_200_OK
# Since we're mocking the magic_link task, we need to manually get the token from Redis
ri = redis_instance()
user_data = json.loads(ri.get(f"magic_{email}"))
token = user_data["token"]
# Use Django client to test the redirect flow without following redirects
url = reverse("magic-sign-up")
response = django_client.post(url, {"email": email, "code": token}, follow=False)
# Check that the initial response is a redirect without error code
assert response.status_code == 302
assert "error_code" not in response.url
# Check if user was created
assert User.objects.filter(email=email).exists()
# Check if user is authenticated
assert "_auth_user_id" in django_client.session
@pytest.mark.django_db
@patch("plane.bgtasks.magic_link_code_task.magic_link.delay")
def test_magic_sign_up_with_next_path(self, mock_magic_link, django_client, api_client, setup_instance):
"""Test magic sign-up with next_path parameter"""
email = "newuser2@plane.so"
# First generate a magic link token
gen_url = reverse("magic-generate")
response = api_client.post(gen_url, {"email": email}, format="json")
# Check that the token generation was successful
assert response.status_code == status.HTTP_200_OK
# Since we're mocking the magic_link task, we need to manually get the token from Redis
ri = redis_instance()
user_data = json.loads(ri.get(f"magic_{email}"))
token = user_data["token"]
# Use Django client to test the redirect flow without following redirects
url = reverse("magic-sign-up")
next_path = "onboarding"
response = django_client.post(url, {"email": email, "code": token, "next_path": next_path}, follow=False)
# Check that the initial response is a redirect without error code
assert response.status_code == 302
assert "error_code" not in response.url
# In a real browser, the next_path would be used to build the absolute URL
# Since we're just testing the authentication logic, we won't check for the exact URL structure
# Check if user was created
assert User.objects.filter(email=email).exists()
# Check if user is authenticated
assert "_auth_user_id" in django_client.session
@@ -0,0 +1,524 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from rest_framework import status
import uuid
from django.utils import timezone
from plane.db.models import (
Project,
ProjectMember,
ProjectUserProperty,
State,
WorkspaceMember,
User,
)
class TestProjectBase:
def get_project_url(self, workspace_slug: str, pk: uuid.UUID = None, details: bool = False) -> str:
"""
Constructs the project endpoint URL for the given workspace as reverse() is
unreliable due to duplicate 'name' values in URL patterns ('api' and 'app').
Args:
workspace_slug (str): The slug of the workspace.
pk (uuid.UUID, optional): The primary key of a specific project.
details (bool, optional): If True, constructs the URL for the
project details endpoint. Defaults to False.
"""
# Establish the common base URL for all project-related endpoints.
base_url = f"/api/workspaces/{workspace_slug}/projects/"
# Specific project instance URL.
if pk:
return f"{base_url}{pk}/"
# Append 'details/' to the base URL.
if details:
return f"{base_url}details/"
# Return the base project list URL.
return base_url
@pytest.mark.contract
class TestProjectAPIPost(TestProjectBase):
"""Test project POST operations"""
@pytest.mark.django_db
def test_create_project_empty_data(self, session_client, workspace):
"""Test creating a project with empty data"""
url = self.get_project_url(workspace.slug)
# Test with empty data
response = session_client.post(url, {}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_project_valid_data(self, session_client, workspace, create_user):
url = self.get_project_url(workspace.slug)
project_data = {
"name": "New Project Test",
"identifier": "NPT",
}
user = create_user
# Make the request
response = session_client.post(url, project_data, format="json")
# Check response status
assert response.status_code == status.HTTP_201_CREATED
# Verify project was created
assert Project.objects.count() == 1
project = Project.objects.get(name=project_data["name"])
assert project.workspace == workspace
# Check if the member is created with the correct role
assert ProjectMember.objects.count() == 1
project_member = ProjectMember.objects.filter(project=project, member=user).first()
assert project_member.role == 20 # Administrator
assert project_member.is_active is True
# Verify ProjectUserProperty was created
assert ProjectUserProperty.objects.filter(project=project, user=user).exists()
# Verify default states were created
states = State.objects.filter(project=project)
assert states.count() == 5
expected_states = ["Backlog", "Todo", "In Progress", "Done", "Cancelled"]
state_names = list(states.values_list("name", flat=True))
assert set(state_names) == set(expected_states)
@pytest.mark.django_db
def test_create_project_with_project_lead(self, session_client, workspace, create_user):
"""Test creating project with a different project lead"""
# Create another user to be project lead
project_lead = User.objects.create_user(email="lead@example.com", username="projectlead")
# Add project lead to workspace
WorkspaceMember.objects.create(workspace=workspace, member=project_lead, role=15)
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Project with Lead",
"identifier": "PWL",
"project_lead": project_lead.id,
}
response = session_client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
# Verify both creator and project lead are administrators
project = Project.objects.get(name=project_data["name"])
assert ProjectMember.objects.filter(project=project, role=20).count() == 2
# Verify both have ProjectUserProperty
assert ProjectUserProperty.objects.filter(project=project).count() == 2
@pytest.mark.django_db
def test_create_project_guest_forbidden(self, session_client, workspace):
"""Test that guests cannot create projects"""
guest_user = User.objects.create_user(email="guest@example.com", username="guest")
WorkspaceMember.objects.create(workspace=workspace, member=guest_user, role=5)
session_client.force_authenticate(user=guest_user)
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Guest Project",
"identifier": "GP",
}
response = session_client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_403_FORBIDDEN
assert Project.objects.count() == 0
@pytest.mark.django_db
def test_create_project_unauthenticated(self, client, workspace):
"""Test unauthenticated access"""
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Unauth Project",
"identifier": "UP",
}
response = client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.django_db
def test_create_project_duplicate_name(self, session_client, workspace, create_user):
"""Test creating project with duplicate name"""
# Create first project
Project.objects.create(name="Duplicate Name", identifier="DN1", workspace=workspace)
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Duplicate Name",
"identifier": "DN2",
}
response = session_client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_project_duplicate_identifier(self, session_client, workspace, create_user):
"""Test creating project with duplicate identifier"""
Project.objects.create(name="First Project", identifier="DUP", workspace=workspace)
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Second Project",
"identifier": "DUP",
}
response = session_client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_project_missing_required_fields(self, session_client, workspace, create_user):
"""Test validation with missing required fields"""
url = self.get_project_url(workspace.slug)
# Test missing name
response = session_client.post(url, {"identifier": "MN"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
# Test missing identifier
response = session_client.post(url, {"name": "Missing Identifier"}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_create_project_with_all_optional_fields(self, session_client, workspace, create_user):
"""Test creating project with all optional fields"""
url = self.get_project_url(workspace.slug)
project_data = {
"name": "Full Project",
"identifier": "FP",
"description": "A comprehensive test project",
"network": 2,
"cycle_view": True,
"issue_views_view": False,
"module_view": True,
"page_view": False,
"inbox_view": True,
"guest_view_all_features": True,
"logo_props": {
"in_use": "emoji",
"emoji": {"value": "🚀", "unicode": "1f680"},
},
}
response = session_client.post(url, project_data, format="json")
assert response.status_code == status.HTTP_201_CREATED
response_data = response.json()
assert response_data["description"] == project_data["description"]
assert response_data["network"] == project_data["network"]
@pytest.mark.contract
class TestProjectAPIGet(TestProjectBase):
"""Test project GET operations"""
@pytest.mark.django_db
def test_list_projects_authenticated_admin(self, session_client, workspace, create_user):
"""Test listing projects as workspace admin"""
# Create a project
project = Project.objects.create(name="Test Project", identifier="TP", workspace=workspace)
# Add user as project member
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug)
response = session_client.get(url)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert len(data) == 1
assert data[0]["name"] == "Test Project"
assert data[0]["identifier"] == "TP"
@pytest.mark.django_db
def test_list_projects_authenticated_guest(self, session_client, workspace):
"""Test listing projects as workspace guest"""
# Create a guest user
guest_user = User.objects.create_user(email="guest@example.com", username="guest")
WorkspaceMember.objects.create(workspace=workspace, member=guest_user, role=5, is_active=True)
# Create projects
project1 = Project.objects.create(name="Project 1", identifier="P1", workspace=workspace)
Project.objects.create(name="Project 2", identifier="P2", workspace=workspace)
# Add guest to only one project
ProjectMember.objects.create(project=project1, member=guest_user, role=10, is_active=True)
session_client.force_authenticate(user=guest_user)
url = self.get_project_url(workspace.slug)
response = session_client.get(url)
assert response.status_code == status.HTTP_200_OK
data = response.json()
# Guest should only see projects they're members of
assert len(data) == 1
assert data[0]["name"] == "Project 1"
@pytest.mark.django_db
def test_list_projects_unauthenticated(self, client, workspace):
"""Test listing projects without authentication"""
url = self.get_project_url(workspace.slug)
response = client.get(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.django_db
def test_list_detail_projects(self, session_client, workspace, create_user):
"""Test listing projects with detailed information"""
# Create a project
project = Project.objects.create(
name="Detailed Project",
identifier="DP",
workspace=workspace,
description="A detailed test project",
)
# Add user as project member
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, details=True)
response = session_client.get(url)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert len(data) == 1
assert data[0]["name"] == "Detailed Project"
assert data[0]["description"] == "A detailed test project"
@pytest.mark.django_db
def test_retrieve_project_success(self, session_client, workspace, create_user):
"""Test retrieving a specific project"""
# Create a project
project = Project.objects.create(
name="Retrieve Test Project",
identifier="RTP",
workspace=workspace,
description="Test project for retrieval",
)
# Add user as project member
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project.id)
response = session_client.get(url)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["name"] == "Retrieve Test Project"
assert data["identifier"] == "RTP"
assert data["description"] == "Test project for retrieval"
@pytest.mark.django_db
def test_retrieve_project_not_found(self, session_client, workspace, create_user):
"""Test retrieving a non-existent project"""
fake_uuid = uuid.uuid4()
url = self.get_project_url(workspace.slug, pk=fake_uuid)
response = session_client.get(url)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_retrieve_archived_project(self, session_client, workspace, create_user):
"""Test retrieving an archived project"""
# Create an archived project
project = Project.objects.create(
name="Archived Project",
identifier="AP",
workspace=workspace,
archived_at=timezone.now(),
)
# Add user as project member
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project.id)
response = session_client.get(url)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.contract
class TestProjectAPIPatchDelete(TestProjectBase):
"""Test project PATCH, and DELETE operations"""
@pytest.mark.django_db
def test_partial_update_project_success(self, session_client, workspace, create_user):
"""Test successful partial update of project"""
# Create a project
project = Project.objects.create(
name="Original Project",
identifier="OP",
workspace=workspace,
description="Original description",
)
# Add user as project administrator
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project.id)
update_data = {
"name": "Updated Project",
"description": "Updated description",
"cycle_view": True,
"module_view": False,
}
response = session_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_200_OK
# Verify project was updated
project.refresh_from_db()
assert project.name == "Updated Project"
assert project.description == "Updated description"
assert project.cycle_view is True
assert project.module_view is False
@pytest.mark.django_db
def test_partial_update_project_forbidden_non_admin(self, session_client, workspace):
"""Test that non-admin project members cannot update project"""
# Create a project
project = Project.objects.create(name="Protected Project", identifier="PP", workspace=workspace)
# Create a member user (not admin)
member_user = User.objects.create_user(email="member@example.com", username="member")
WorkspaceMember.objects.create(workspace=workspace, member=member_user, role=15, is_active=True)
ProjectMember.objects.create(project=project, member=member_user, role=15, is_active=True)
session_client.force_authenticate(user=member_user)
url = self.get_project_url(workspace.slug, pk=project.id)
update_data = {"name": "Hacked Project"}
response = session_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db
def test_partial_update_duplicate_name_conflict(self, session_client, workspace, create_user):
"""Test updating project with duplicate name returns conflict"""
# Create two projects
Project.objects.create(name="Project One", identifier="P1", workspace=workspace)
project2 = Project.objects.create(name="Project Two", identifier="P2", workspace=workspace)
ProjectMember.objects.create(project=project2, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project2.id)
update_data = {"name": "Project One"} # Duplicate name
response = session_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_partial_update_duplicate_identifier_conflict(self, session_client, workspace, create_user):
"""Test updating project with duplicate identifier returns conflict"""
# Create two projects
Project.objects.create(name="Project One", identifier="P1", workspace=workspace)
project2 = Project.objects.create(name="Project Two", identifier="P2", workspace=workspace)
ProjectMember.objects.create(project=project2, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project2.id)
update_data = {"identifier": "P1"} # Duplicate identifier
response = session_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_partial_update_invalid_data(self, session_client, workspace, create_user):
"""Test partial update with invalid data"""
project = Project.objects.create(name="Valid Project", identifier="VP", workspace=workspace)
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project.id)
update_data = {"name": ""}
response = session_client.patch(url, update_data, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_delete_project_success_project_admin(self, session_client, workspace, create_user):
"""Test successful project deletion by project admin"""
project = Project.objects.create(name="Delete Me", identifier="DM", workspace=workspace)
ProjectMember.objects.create(project=project, member=create_user, role=20, is_active=True)
url = self.get_project_url(workspace.slug, pk=project.id)
response = session_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Project.objects.filter(id=project.id).exists()
@pytest.mark.django_db
def test_delete_project_success_workspace_admin(self, session_client, workspace):
"""Test successful project deletion by workspace admin"""
# Create workspace admin user
workspace_admin = User.objects.create_user(email="admin@example.com", username="admin")
WorkspaceMember.objects.create(workspace=workspace, member=workspace_admin, role=20, is_active=True)
project = Project.objects.create(name="Delete Me", identifier="DM", workspace=workspace)
session_client.force_authenticate(user=workspace_admin)
url = self.get_project_url(workspace.slug, pk=project.id)
response = session_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Project.objects.filter(id=project.id).exists()
@pytest.mark.django_db
def test_delete_project_forbidden_non_admin(self, session_client, workspace):
"""Test that non-admin users cannot delete projects"""
# Create a member user (not admin)
member_user = User.objects.create_user(email="member@example.com", username="member")
WorkspaceMember.objects.create(workspace=workspace, member=member_user, role=15, is_active=True)
project = Project.objects.create(name="Protected Project", identifier="PP", workspace=workspace)
ProjectMember.objects.create(project=project, member=member_user, role=15, is_active=True)
session_client.force_authenticate(user=member_user)
url = self.get_project_url(workspace.slug, pk=project.id)
response = session_client.delete(url)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert Project.objects.filter(id=project.id).exists()
@pytest.mark.django_db
def test_delete_project_unauthenticated(self, client, workspace):
"""Test unauthenticated project deletion"""
project = Project.objects.create(name="Protected Project", identifier="PP", workspace=workspace)
url = self.get_project_url(workspace.slug, pk=project.id)
response = client.delete(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert Project.objects.filter(id=project.id).exists()
@@ -0,0 +1,77 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from django.urls import reverse
from rest_framework import status
from unittest.mock import patch
from plane.db.models import Workspace, WorkspaceMember
@pytest.mark.contract
class TestWorkspaceAPI:
"""Test workspace CRUD operations"""
@pytest.mark.django_db
def test_create_workspace_empty_data(self, session_client):
"""Test creating a workspace with empty data"""
url = reverse("workspace")
# Test with empty data
response = session_client.post(url, {}, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
@patch("plane.bgtasks.workspace_seed_task.workspace_seed.delay")
def test_create_workspace_valid_data(self, mock_workspace_seed, session_client, create_user):
"""Test creating a workspace with valid data"""
url = reverse("workspace")
user = create_user # Use the create_user fixture directly as it returns a user object
# Test with valid data - include all required fields
workspace_data = {
"name": "Plane",
"slug": "pla-ne-test",
"company_name": "Plane Inc.",
}
# Make the request
response = session_client.post(url, workspace_data, format="json")
# Check response status
assert response.status_code == status.HTTP_201_CREATED
# Verify workspace was created
assert Workspace.objects.count() == 1
# Check if the member is created
assert WorkspaceMember.objects.count() == 1
# Check other values
workspace = Workspace.objects.get(slug=workspace_data["slug"])
workspace_member = WorkspaceMember.objects.filter(workspace=workspace, member=user).first()
assert workspace.owner == user
assert workspace_member.role == 20
# Verify the workspace_seed task was called
mock_workspace_seed.assert_called_once_with(response.data["id"])
@pytest.mark.django_db
@patch("plane.bgtasks.workspace_seed_task.workspace_seed.delay")
def test_create_duplicate_workspace(self, mock_workspace_seed, session_client):
"""Test creating a duplicate workspace"""
url = reverse("workspace")
# Create first workspace
session_client.post(url, {"name": "Plane", "slug": "pla-ne"}, format="json")
# Try to create a workspace with the same slug
response = session_client.post(url, {"name": "Plane", "slug": "pla-ne"}, format="json")
# The API returns 400 BAD REQUEST for duplicate slugs, not 409 CONFLICT
assert response.status_code == status.HTTP_400_BAD_REQUEST
# Optionally check the error message to confirm it's related to the duplicate slug
assert "slug" in response.data
@@ -0,0 +1,85 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import factory
from uuid import uuid4
from django.utils import timezone
from plane.db.models import User, Workspace, WorkspaceMember, Project, ProjectMember
class UserFactory(factory.django.DjangoModelFactory):
"""Factory for creating User instances"""
class Meta:
model = User
django_get_or_create = ("email",)
id = factory.LazyFunction(uuid4)
email = factory.Sequence(lambda n: f"user{n}@plane.so")
password = factory.PostGenerationMethodCall("set_password", "password")
first_name = factory.Sequence(lambda n: f"First{n}")
last_name = factory.Sequence(lambda n: f"Last{n}")
is_active = True
is_superuser = False
is_staff = False
class WorkspaceFactory(factory.django.DjangoModelFactory):
"""Factory for creating Workspace instances"""
class Meta:
model = Workspace
django_get_or_create = ("slug",)
id = factory.LazyFunction(uuid4)
name = factory.Sequence(lambda n: f"Workspace {n}")
slug = factory.Sequence(lambda n: f"workspace-{n}")
owner = factory.SubFactory(UserFactory)
created_at = factory.LazyFunction(timezone.now)
updated_at = factory.LazyFunction(timezone.now)
class WorkspaceMemberFactory(factory.django.DjangoModelFactory):
"""Factory for creating WorkspaceMember instances"""
class Meta:
model = WorkspaceMember
id = factory.LazyFunction(uuid4)
workspace = factory.SubFactory(WorkspaceFactory)
member = factory.SubFactory(UserFactory)
role = 20 # Admin role by default
created_at = factory.LazyFunction(timezone.now)
updated_at = factory.LazyFunction(timezone.now)
class ProjectFactory(factory.django.DjangoModelFactory):
"""Factory for creating Project instances"""
class Meta:
model = Project
django_get_or_create = ("name", "workspace")
id = factory.LazyFunction(uuid4)
name = factory.Sequence(lambda n: f"Project {n}")
workspace = factory.SubFactory(WorkspaceFactory)
created_by = factory.SelfAttribute("workspace.owner")
updated_by = factory.SelfAttribute("workspace.owner")
created_at = factory.LazyFunction(timezone.now)
updated_at = factory.LazyFunction(timezone.now)
class ProjectMemberFactory(factory.django.DjangoModelFactory):
"""Factory for creating ProjectMember instances"""
class Meta:
model = ProjectMember
id = factory.LazyFunction(uuid4)
project = factory.SubFactory(ProjectFactory)
member = factory.SubFactory(UserFactory)
role = 20 # Admin role by default
created_at = factory.LazyFunction(timezone.now)
updated_at = factory.LazyFunction(timezone.now)
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,101 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
import requests
from django.urls import reverse
@pytest.mark.smoke
class TestAuthSmoke:
"""Smoke tests for authentication endpoints"""
@pytest.mark.django_db
def test_login_endpoint_available(self, plane_server, create_user, user_data):
"""Test that the login endpoint is available and responds correctly"""
# Get the sign-in URL
relative_url = reverse("sign-in")
url = f"{plane_server.url}{relative_url}"
# 1. Test bad login - test with wrong password
response = requests.post(url, data={"email": user_data["email"], "password": "wrong-password"})
# For bad credentials, any of these status codes would be valid
# The test shouldn't be brittle to minor implementation changes
assert response.status_code != 500, "Authentication should not cause server errors"
assert response.status_code != 404, "Authentication endpoint should exist"
if response.status_code == 200:
# If API returns 200 for failures, check the response body for error indication
if hasattr(response, "json"):
try:
data = response.json()
# JSON response might indicate error in its structure
assert (
"error" in data or "error_code" in data or "detail" in data or response.url.endswith("sign-in")
), "Error response should contain error details"
except ValueError:
# It's ok if response isn't JSON format
pass
elif response.status_code in [302, 303]:
# If it's a redirect, it should redirect to a login page or error page
redirect_url = response.headers.get("Location", "")
assert "error" in redirect_url or "sign-in" in redirect_url, (
"Failed login should redirect to login page or error page"
)
# 2. Test good login with correct credentials
response = requests.post(
url,
data={"email": user_data["email"], "password": user_data["password"]},
allow_redirects=False, # Don't follow redirects
)
# Successful auth should not be a client error or server error
assert response.status_code not in range(400, 600), (
f"Authentication with valid credentials failed with status {response.status_code}"
)
# Specific validation based on response type
if response.status_code in [302, 303]:
# Redirect-based auth: check that redirect URL doesn't contain error
redirect_url = response.headers.get("Location", "")
assert "error" not in redirect_url and "error_code" not in redirect_url, (
"Successful login redirect should not contain error parameters"
)
elif response.status_code == 200:
# API token-based auth: check for tokens or user session
if hasattr(response, "json"):
try:
data = response.json()
# If it's a token response
if "access_token" in data:
assert "refresh_token" in data, "JWT auth should return both access and refresh tokens"
# If it's a user session response
elif "user" in data:
assert "is_authenticated" in data and data["is_authenticated"], (
"User session response should indicate authentication"
)
# Otherwise it should at least indicate success
else:
assert not any(error_key in data for error_key in ["error", "error_code", "detail"]), (
"Success response should not contain error keys"
)
except ValueError:
# Non-JSON is acceptable if it's a redirect or HTML response
pass
@pytest.mark.smoke
class TestHealthCheckSmoke:
"""Smoke test for health check endpoint"""
def test_healthcheck_endpoint(self, plane_server):
"""Test that the health check endpoint is available and responds correctly"""
# Make a request to the health check endpoint
response = requests.get(f"{plane_server.url}/")
# Should be OK
assert response.status_code == 200, "Health check endpoint should return 200 OK"
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,174 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from plane.db.models import Project, ProjectMember, Issue, FileAsset
from unittest.mock import patch, MagicMock
from plane.bgtasks.copy_s3_object import (
copy_s3_objects_of_description_and_assets,
copy_assets,
)
import base64
@pytest.mark.unit
class TestCopyS3Objects:
"""Test the copy_s3_objects_of_description_and_assets function"""
@pytest.fixture
def project(self, create_user, workspace):
project = Project.objects.create(name="Test Project", identifier="test-project", workspace=workspace)
ProjectMember.objects.create(project=project, member=create_user)
return project
@pytest.fixture
def issue(self, workspace, project):
return Issue.objects.create(
name="Test Issue",
workspace=workspace,
project_id=project.id,
description_html='<div><image-component src="35e8b958-6ee5-43ce-ae56-fb0e776f421e"></image-component><image-component src="97988198-274f-4dfe-aa7a-4c0ffc684214"></image-component></div>', # noqa: E501
)
@pytest.fixture
def file_asset(self, workspace, project, issue):
return FileAsset.objects.create(
issue=issue,
workspace=workspace,
project=project,
asset="workspace1/test-asset-1.jpg",
attributes={
"name": "test-asset-1.jpg",
"size": 100,
"type": "image/jpeg",
},
id="35e8b958-6ee5-43ce-ae56-fb0e776f421e",
entity_type="ISSUE_DESCRIPTION",
)
@pytest.mark.django_db
@patch("plane.bgtasks.copy_s3_object.S3Storage")
def test_copy_s3_objects_of_description_and_assets(
self, mock_s3_storage, create_user, workspace, project, issue, file_asset
):
FileAsset.objects.create(
issue=issue,
workspace=workspace,
project=project,
asset="workspace1/test-asset-2.pdf",
attributes={
"name": "test-asset-2.pdf",
"size": 100,
"type": "application/pdf",
},
id="97988198-274f-4dfe-aa7a-4c0ffc684214",
entity_type="ISSUE_DESCRIPTION",
)
issue.save()
# Set up mock S3 storage
mock_storage_instance = MagicMock()
mock_s3_storage.return_value = mock_storage_instance
# Mock the external service call to avoid actual HTTP requests
with patch("plane.bgtasks.copy_s3_object.sync_with_external_service") as mock_sync:
mock_sync.return_value = {
"description": "test description",
"description_binary": base64.b64encode(b"test binary").decode(),
}
# Call the actual function (not .delay())
copy_s3_objects_of_description_and_assets("ISSUE", issue.id, project.id, "test-workspace", create_user.id)
# Assert that copy_object was called for each asset
assert mock_storage_instance.copy_object.call_count == 2
# Get the updated issue and its new assets
updated_issue = Issue.objects.get(id=issue.id)
new_assets = FileAsset.objects.filter(
issue=updated_issue,
entity_type="ISSUE_DESCRIPTION",
)
# Verify new assets were created
assert new_assets.count() == 4 # 2 original + 2 copied
@pytest.mark.django_db
@patch("plane.bgtasks.copy_s3_object.S3Storage")
def test_copy_assets_successful(self, mock_s3_storage, workspace, project, issue, file_asset):
"""Test successful copying of assets"""
# Arrange
mock_storage_instance = MagicMock()
mock_s3_storage.return_value = mock_storage_instance
# Act
result = copy_assets(
entity=issue,
entity_identifier=issue.id,
project_id=project.id,
asset_ids=[file_asset.id],
user_id=issue.created_by_id,
)
# Assert
# Verify S3 copy was called
mock_storage_instance.copy_object.assert_called_once()
# Verify new asset was created
assert len(result) == 1
new_asset_id = result[0]["new_asset_id"]
new_asset = FileAsset.objects.get(id=new_asset_id)
# Verify asset properties were copied correctly
assert new_asset.workspace == workspace
assert new_asset.project_id == project.id
assert new_asset.entity_type == file_asset.entity_type
assert new_asset.attributes == file_asset.attributes
assert new_asset.size == file_asset.size
assert new_asset.is_uploaded is True
@pytest.mark.django_db
@patch("plane.bgtasks.copy_s3_object.S3Storage")
def test_copy_assets_empty_asset_ids(self, mock_s3_storage, workspace, project, issue):
"""Test copying with empty asset_ids list"""
# Arrange
mock_storage_instance = MagicMock()
mock_s3_storage.return_value = mock_storage_instance
# Act
result = copy_assets(
entity=issue,
entity_identifier=issue.id,
project_id=project.id,
asset_ids=[],
user_id=issue.created_by_id,
)
# Assert
assert result == []
mock_storage_instance.copy_object.assert_not_called()
@pytest.mark.django_db
@patch("plane.bgtasks.copy_s3_object.S3Storage")
def test_copy_assets_nonexistent_asset(self, mock_s3_storage, workspace, project, issue):
"""Test copying with non-existent asset ID"""
# Arrange
mock_storage_instance = MagicMock()
mock_s3_storage.return_value = mock_storage_instance
non_existent_id = "00000000-0000-0000-0000-000000000000"
# Act
result = copy_assets(
entity=issue,
entity_identifier=issue.id,
project_id=project.id,
asset_ids=[non_existent_id],
user_id=issue.created_by_id,
)
# Assert
assert result == []
mock_storage_instance.copy_object.assert_not_called()
@@ -0,0 +1,126 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from unittest.mock import patch, MagicMock
from plane.bgtasks.work_item_link_task import safe_get, validate_url_ip
def _make_response(status_code=200, headers=None, is_redirect=False, content=b""):
"""Create a mock requests.Response."""
resp = MagicMock()
resp.status_code = status_code
resp.is_redirect = is_redirect
resp.headers = headers or {}
resp.content = content
return resp
@pytest.mark.unit
class TestValidateUrlIp:
"""Test validate_url_ip blocks private/internal IPs."""
def test_rejects_private_ip(self):
with patch("plane.bgtasks.work_item_link_task.socket.getaddrinfo") as mock_dns:
mock_dns.return_value = [(None, None, None, None, ("192.168.1.1", 0))]
with pytest.raises(ValueError, match="private/internal"):
validate_url_ip("http://example.com")
def test_rejects_loopback(self):
with patch("plane.bgtasks.work_item_link_task.socket.getaddrinfo") as mock_dns:
mock_dns.return_value = [(None, None, None, None, ("127.0.0.1", 0))]
with pytest.raises(ValueError, match="private/internal"):
validate_url_ip("http://example.com")
def test_rejects_non_http_scheme(self):
with pytest.raises(ValueError, match="Only HTTP and HTTPS"):
validate_url_ip("file:///etc/passwd")
def test_allows_public_ip(self):
with patch("plane.bgtasks.work_item_link_task.socket.getaddrinfo") as mock_dns:
mock_dns.return_value = [(None, None, None, None, ("93.184.216.34", 0))]
validate_url_ip("https://example.com") # Should not raise
@pytest.mark.unit
class TestSafeGet:
"""Test safe_get follows redirects safely and blocks SSRF."""
@patch("plane.bgtasks.work_item_link_task.requests.get")
@patch("plane.bgtasks.work_item_link_task.validate_url_ip")
def test_returns_response_for_non_redirect(self, mock_validate, mock_get):
final_resp = _make_response(status_code=200, content=b"OK")
mock_get.return_value = final_resp
response, final_url = safe_get("https://example.com")
assert response is final_resp
assert final_url == "https://example.com"
mock_validate.assert_called_once_with("https://example.com")
@patch("plane.bgtasks.work_item_link_task.requests.get")
@patch("plane.bgtasks.work_item_link_task.validate_url_ip")
def test_follows_redirect_and_validates_each_hop(self, mock_validate, mock_get):
redirect_resp = _make_response(
status_code=301,
is_redirect=True,
headers={"Location": "https://other.com/page"},
)
final_resp = _make_response(status_code=200, content=b"OK")
mock_get.side_effect = [redirect_resp, final_resp]
response, final_url = safe_get("https://example.com")
assert response is final_resp
assert final_url == "https://other.com/page"
# Should validate both the initial URL and the redirect target
assert mock_validate.call_count == 2
mock_validate.assert_any_call("https://example.com")
mock_validate.assert_any_call("https://other.com/page")
@patch("plane.bgtasks.work_item_link_task.requests.get")
@patch("plane.bgtasks.work_item_link_task.validate_url_ip")
def test_blocks_redirect_to_private_ip(self, mock_validate, mock_get):
redirect_resp = _make_response(
status_code=302,
is_redirect=True,
headers={"Location": "http://192.168.1.1:8080"},
)
mock_get.return_value = redirect_resp
# First call (initial URL) succeeds, second call (redirect target) fails
mock_validate.side_effect = [None, ValueError("Access to private/internal networks is not allowed")]
with pytest.raises(ValueError, match="private/internal"):
safe_get("https://evil.com/redirect")
@patch("plane.bgtasks.work_item_link_task.requests.get")
@patch("plane.bgtasks.work_item_link_task.validate_url_ip")
def test_raises_on_too_many_redirects(self, mock_validate, mock_get):
redirect_resp = _make_response(
status_code=302,
is_redirect=True,
headers={"Location": "https://example.com/loop"},
)
mock_get.return_value = redirect_resp
with pytest.raises(RuntimeError, match="Too many redirects"):
safe_get("https://example.com/start")
@patch("plane.bgtasks.work_item_link_task.requests.get")
@patch("plane.bgtasks.work_item_link_task.validate_url_ip")
def test_succeeds_at_exact_max_redirects(self, mock_validate, mock_get):
"""After exactly MAX_REDIRECTS hops, if the final response is 200, it should succeed."""
redirect_resp = _make_response(
status_code=302,
is_redirect=True,
headers={"Location": "https://example.com/next"},
)
final_resp = _make_response(status_code=200, content=b"OK")
# 5 redirects then a 200
mock_get.side_effect = [redirect_resp] * 5 + [final_resp]
response, final_url = safe_get("https://example.com/start")
assert response is final_resp
assert not response.is_redirect
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,423 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
"""
Unit tests for ReadReplicaRoutingMiddleware.
This module contains comprehensive tests for the ReadReplicaRoutingMiddleware
that handles intelligent database routing to read replicas based on HTTP methods
and view configuration.
Test Organization:
- TestReadReplicaRoutingMiddleware: Core middleware functionality
- TestProcessView: process_view method behavior
- TestReplicaDecisionLogic: Decision logic for replica usage
- TestAttributeDetection: View attribute detection methods
- TestExceptionHandling: Exception handling and cleanup
- TestRealViewIntegration: Real Django/DRF view integration
- TestEdgeCases: Edge cases and error conditions
"""
import pytest
from unittest.mock import Mock, patch
from django.http import HttpResponse
from django.test import RequestFactory
from django.views import View
from rest_framework.views import APIView
from rest_framework.viewsets import ViewSet
from plane.middleware.db_routing import ReadReplicaRoutingMiddleware
# Pytest fixtures
@pytest.fixture
def mock_get_response():
"""Fixture for mocked get_response callable."""
return Mock(return_value=HttpResponse())
@pytest.fixture
def middleware(mock_get_response):
"""Fixture for ReadReplicaRoutingMiddleware instance."""
return ReadReplicaRoutingMiddleware(mock_get_response)
@pytest.fixture
def request_factory():
"""Fixture for Django RequestFactory."""
return RequestFactory()
@pytest.fixture
def mock_view_func():
"""Fixture for a basic mocked view function."""
view = Mock()
view.use_read_replica = True
return view
@pytest.fixture
def get_request(request_factory):
"""Fixture for a GET request."""
return request_factory.get("/api/test/")
@pytest.fixture
def post_request(request_factory):
"""Fixture for a POST request."""
return request_factory.post("/api/test/")
@pytest.mark.unit
class TestReadReplicaRoutingMiddleware:
"""Test cases for ReadReplicaRoutingMiddleware core functionality."""
def test_middleware_initialization(self, middleware, mock_get_response):
"""Test middleware initializes correctly with expected attributes."""
assert middleware.get_response == mock_get_response
assert hasattr(middleware, "READ_ONLY_METHODS")
assert "GET" in middleware.READ_ONLY_METHODS
assert "HEAD" in middleware.READ_ONLY_METHODS
assert "OPTIONS" in middleware.READ_ONLY_METHODS
def test_read_only_methods_constant(self, middleware):
"""Test READ_ONLY_METHODS contains expected HTTP methods."""
expected_methods = {"GET", "HEAD", "OPTIONS"}
assert middleware.READ_ONLY_METHODS == expected_methods
@patch("plane.middleware.db_routing.set_use_read_replica")
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_call_routes_write_methods_to_primary(
self, mock_clear, mock_set, middleware, post_request, mock_get_response
):
"""Test __call__ routes write methods to primary database."""
response = middleware(post_request)
mock_set.assert_called_once_with(False) # Primary database
mock_clear.assert_called_once()
assert response == mock_get_response.return_value
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_call_with_read_methods_waits_for_process_view(
self, mock_clear, middleware, get_request, mock_get_response
):
"""Test __call__ with read methods waits for process_view."""
response = middleware(get_request)
mock_clear.assert_called_once()
assert response == mock_get_response.return_value
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_call_always_cleans_up_context(self, mock_clear, middleware, get_request):
"""Test __call__ always cleans up context."""
middleware(get_request)
mock_clear.assert_called_once()
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_call_cleans_up_context_on_exception(self, mock_clear, middleware, get_request, mock_get_response):
"""Test __call__ cleans up context even if get_response raises."""
mock_get_response.side_effect = Exception("Test exception")
with pytest.raises(Exception, match="Test exception"):
middleware(get_request)
mock_clear.assert_called_once()
@pytest.mark.unit
class TestProcessView:
"""Test cases for process_view method functionality."""
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_read_method_and_replica_true(self, mock_set, middleware, get_request):
"""Test process_view with GET request and use_read_replica=True."""
view_func = Mock()
view_func.use_read_replica = True
result = middleware.process_view(get_request, view_func, (), {})
mock_set.assert_called_once_with(True)
assert result is None
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_read_method_and_replica_false(self, mock_set, middleware, get_request):
"""Test process_view with GET request and use_read_replica=False."""
view_func = Mock()
view_func.use_read_replica = False
result = middleware.process_view(get_request, view_func, (), {})
mock_set.assert_called_once_with(False)
assert result is None
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_read_method_and_no_replica_attribute(self, mock_set, middleware, get_request):
"""Test process_view with GET request and no use_read_replica attr."""
view_func = Mock(spec=[]) # No use_read_replica attribute
result = middleware.process_view(get_request, view_func, (), {})
mock_set.assert_called_once_with(False) # Default to primary
assert result is None
def test_with_write_method_ignores_view_attributes(self, middleware, post_request):
"""Test process_view with write methods ignores view attributes."""
view_func = Mock()
view_func.use_read_replica = True # This should be ignored for POST
result = middleware.process_view(post_request, view_func, (), {})
assert result is None # Should not process for write methods
@pytest.mark.unit
class TestReplicaDecisionLogic:
"""Test cases for replica decision logic methods."""
def test_should_use_read_replica_with_true_attribute(self, middleware):
"""Test _should_use_read_replica returns True for True attribute."""
view_func = Mock()
view_func.use_read_replica = True
result = middleware._should_use_read_replica(view_func)
assert result is True
def test_should_use_read_replica_with_false_attribute(self, middleware):
"""Test _should_use_read_replica returns False for False attribute."""
view_func = Mock()
view_func.use_read_replica = False
result = middleware._should_use_read_replica(view_func)
assert result is False
def test_should_use_read_replica_with_no_attribute_defaults_false(self, middleware):
"""Test _should_use_read_replica defaults to False for missing attr."""
view_func = Mock(spec=[]) # No use_read_replica attribute
result = middleware._should_use_read_replica(view_func)
assert result is False
@pytest.mark.unit
class TestAttributeDetection:
"""Test cases for view attribute detection methods."""
def test_get_use_replica_attribute_function_based_view(self, middleware):
"""Test _get_use_replica_attribute with function-based view."""
# Test with True
view_func = Mock()
view_func.use_read_replica = True
result = middleware._get_use_replica_attribute(view_func)
assert result is True
# Test with False
view_func.use_read_replica = False
result = middleware._get_use_replica_attribute(view_func)
assert result is False
# Test with no attribute
view_func = Mock(spec=[])
result = middleware._get_use_replica_attribute(view_func)
assert result is None
def test_get_use_replica_attribute_django_cbv(self, middleware):
"""Test _get_use_replica_attribute with Django CBV wrapper."""
view_class = Mock()
view_class.use_read_replica = True
view_func = Mock()
view_func.view_class = view_class
# Remove use_read_replica from view_func to ensure it checks view_class
del view_func.use_read_replica
result = middleware._get_use_replica_attribute(view_func)
assert result is True
def test_get_use_replica_attribute_drf_wrapper(self, middleware):
"""Test _get_use_replica_attribute with DRF wrapper."""
# Create a real object to avoid Mock issues
class ViewClass:
use_read_replica = True
class ViewFunc:
cls = ViewClass()
view_func = ViewFunc()
result = middleware._get_use_replica_attribute(view_func)
assert result is True
def test_get_use_replica_attribute_priority_order(self, middleware):
"""Test attribute priority: direct > view_class > cls."""
view_func = Mock()
view_func.use_read_replica = True # Direct attribute (highest priority)
# Add conflicting attributes with lower priority
view_class = Mock()
view_class.use_read_replica = False
view_func.view_class = view_class
cls = Mock()
cls.use_read_replica = False
view_func.cls = cls
result = middleware._get_use_replica_attribute(view_func)
assert result is True # Should use direct attribute
@pytest.mark.parametrize(
"value,expected",
[
(True, True),
(False, False),
(1, True),
(0, False),
("yes", True),
("", False),
([], False),
([1], True),
(None, False),
],
)
def test_should_use_read_replica_truthy_falsy_values(self, middleware, value, expected):
"""Test _should_use_read_replica with various truthy/falsy values."""
# Create a real object to test the attribute handling
class TestView:
pass
view_func = TestView()
view_func.use_read_replica = value
result = middleware._should_use_read_replica(view_func)
assert result == expected
@pytest.mark.unit
class TestExceptionHandling:
"""Test cases for exception handling and cleanup."""
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_process_exception_cleans_up_context(self, mock_clear, middleware, request_factory):
"""Test process_exception cleans up context."""
request = request_factory.get("/api/test/")
exception = Exception("Test exception")
result = middleware.process_exception(request, exception)
mock_clear.assert_called_once()
assert result is None # Don't handle the exception
@patch("plane.middleware.db_routing.set_use_read_replica")
@patch("plane.middleware.db_routing.clear_read_replica_context")
def test_integration_full_request_cycle(self, mock_clear, mock_set, middleware, request_factory, mock_get_response):
"""Test complete request cycle from __call__ through process_view."""
request = request_factory.get("/api/test/")
view_func = Mock()
view_func.use_read_replica = True
# Call middleware and process_view manually
response = middleware(request)
middleware.process_view(request, view_func, (), {})
mock_set.assert_called_once_with(True)
mock_clear.assert_called_once()
assert response == mock_get_response.return_value
@pytest.mark.unit
class TestRealViewIntegration:
"""Test middleware with real Django/DRF view classes."""
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_django_class_based_view(self, mock_set, middleware, request_factory):
"""Test middleware with actual Django CBV."""
class TestView(View):
use_read_replica = True
# Simulate Django's URL resolver creating a view wrapper
view_func = TestView.as_view()
request = request_factory.get("/api/test/")
middleware.process_view(request, view_func, (), {})
mock_set.assert_called_once_with(True)
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_drf_api_view(self, mock_set, middleware, request_factory):
"""Test middleware with DRF APIView."""
class TestAPIView(APIView):
use_read_replica = True
# Simulate DRF's URL pattern creating a view wrapper
view_func = TestAPIView.as_view()
request = request_factory.get("/api/test/")
middleware.process_view(request, view_func, (), {})
mock_set.assert_called_once_with(True)
@patch("plane.middleware.db_routing.set_use_read_replica")
def test_with_drf_viewset(self, mock_set, middleware, request_factory):
"""Test middleware with DRF ViewSet."""
class TestViewSet(ViewSet):
use_read_replica = True
# Simulate DRF router creating viewset action
view_func = TestViewSet.as_view({"get": "list"})
request = request_factory.get("/api/test/")
middleware.process_view(request, view_func, (), {})
mock_set.assert_called_once_with(True)
@pytest.mark.unit
class TestEdgeCases:
"""Test edge cases and error conditions."""
def test_process_view_with_none_view_func(self, middleware, request_factory):
"""Test process_view handles None view_func gracefully."""
request = request_factory.get("/api/test/")
result = middleware.process_view(request, None, (), {})
assert result is None # Should not crash
def test_get_use_replica_attribute_with_attribute_error(self, middleware):
"""Test _get_use_replica_attribute with view that raises AttributeError."""
# Create a view class that raises AttributeError on access
class ProblematicView:
def __getattr__(self, name):
if name == "use_read_replica":
raise AttributeError("Simulated attribute error")
raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
view_func = ProblematicView()
result = middleware._get_use_replica_attribute(view_func)
assert result is None # Should handle gracefully
def test_multiple_exception_calls_are_safe(self, middleware, request_factory):
"""Test that multiple calls to process_exception don't cause issues."""
request = request_factory.get("/api/test/")
exception = Exception("Test exception")
# Call multiple times
result1 = middleware.process_exception(request, exception)
result2 = middleware.process_exception(request, exception)
assert result1 is None # Both should return None safely
assert result2 is None
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,290 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from plane.db.models import IssueComment, Description, Project, Issue, Workspace, State
@pytest.fixture
def workspace(create_user):
"""Create a test workspace"""
return Workspace.objects.create(
name="Test Workspace",
slug="test-workspace",
owner=create_user,
)
@pytest.fixture
def project(workspace, create_user):
"""Create a test project"""
return Project.objects.create(
name="Test Project",
identifier="TP",
workspace=workspace,
created_by=create_user,
)
@pytest.fixture
def state(project):
"""Create a test state"""
return State.objects.create(
name="Todo",
project=project,
group="backlog",
default=True,
)
@pytest.fixture
def issue(workspace, project, state, create_user):
"""Create a test issue"""
return Issue.objects.create(
name="Test Issue",
workspace=workspace,
project=project,
state=state,
created_by=create_user,
)
@pytest.mark.unit
class TestIssueCommentModel:
"""Test the IssueComment model"""
@pytest.mark.django_db
def test_issue_comment_creation_creates_description(self, workspace, project, issue, create_user):
"""Test that creating a comment automatically creates a description"""
# Arrange
comment_html = "<p>This is a test comment</p>"
comment_json = {"type": "doc", "content": [{"type": "paragraph", "text": "This is a test comment"}]}
# Act
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=comment_html,
comment_json=comment_json,
created_by=create_user,
updated_by=create_user,
)
# Assert
assert issue_comment.id is not None
assert issue_comment.comment_stripped == "This is a test comment"
assert issue_comment.description_id is not None
# Verify description was created
description = Description.objects.get(pk=issue_comment.description_id)
assert description is not None
assert description.description_html == comment_html
assert description.description_json == comment_json
assert description.description_stripped == "This is a test comment"
assert description.workspace_id == workspace.id
assert description.project_id == project.id
@pytest.mark.django_db
def test_issue_comment_update_updates_description(self, workspace, project, issue, create_user):
"""Test that updating a comment updates its associated description"""
# Arrange - Create initial comment
initial_html = "<p>Initial comment</p>"
initial_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Initial comment"}]}
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=initial_html,
comment_json=initial_json,
created_by=create_user,
updated_by=create_user,
)
initial_description_id = issue_comment.description_id
# Act - Update the comment
updated_html = "<p>Updated comment</p>"
updated_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Updated comment"}]}
issue_comment.comment_html = updated_html
issue_comment.comment_json = updated_json
issue_comment.save()
# Assert
# Refresh from database
issue_comment.refresh_from_db()
updated_description = Description.objects.get(pk=initial_description_id)
# Verify comment was updated
assert issue_comment.comment_stripped == "Updated comment"
assert issue_comment.description_id == initial_description_id # Same description object
# Verify description was updated
assert updated_description.description_html == updated_html
assert updated_description.description_json == updated_json
assert updated_description.description_stripped == "Updated comment"
@pytest.mark.django_db
def test_issue_comment_update_only_changed_fields_in_description(self, workspace, project, issue, create_user):
"""Test that only changed fields are updated in description"""
# Arrange - Create initial comment
initial_html = "<p>Initial comment</p>"
initial_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Initial comment"}]}
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=initial_html,
comment_json=initial_json,
created_by=create_user,
updated_by=create_user,
)
initial_description_id = issue_comment.description_id
# Act - Update only the HTML (not JSON)
updated_html = "<p>Updated comment only HTML</p>"
issue_comment.comment_html = updated_html
# comment_json remains the same
issue_comment.save()
# Assert
updated_description = Description.objects.get(pk=initial_description_id)
# Verify HTML was updated
assert updated_description.description_html == updated_html
assert updated_description.description_stripped == "Updated comment only HTML"
# Verify JSON remained the same
assert updated_description.description_json == initial_json
@pytest.mark.django_db
def test_issue_comment_no_update_when_content_unchanged(self, workspace, project, issue, create_user):
"""Test that description is not updated when comment content doesn't change"""
# Arrange - Create initial comment
initial_html = "<p>Test comment</p>"
initial_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Test comment"}]}
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=initial_html,
comment_json=initial_json,
created_by=create_user,
updated_by=create_user,
)
initial_description_id = issue_comment.description_id
# Act - Save without changing content
issue_comment.save()
# Assert
updated_description = Description.objects.get(pk=initial_description_id)
# Verify description was not updated (updated_at should be the same)
# Note: This test assumes updated_at is not changed when no fields change
assert updated_description.description_html == initial_html
assert updated_description.description_json == initial_json
assert updated_description.description_stripped == "Test comment"
@pytest.mark.django_db
def test_issue_comment_update_creates_description_if_missing(self, workspace, project, issue, create_user):
"""Test that updating a comment creates description if it doesn't exist (legacy data)"""
# Arrange - Create comment and manually remove description (simulating legacy data)
initial_html = "<p>Legacy comment</p>"
initial_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Legacy comment"}]}
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=initial_html,
comment_json=initial_json,
created_by=create_user,
updated_by=create_user,
)
# Simulate legacy data by removing the description
if issue_comment.description_id:
Description.objects.filter(pk=issue_comment.description_id).delete()
IssueComment.objects.filter(pk=issue_comment.pk).update(description_id=None)
issue_comment.refresh_from_db()
assert issue_comment.description_id is None
# Act - Update the comment
updated_html = "<p>Updated legacy comment</p>"
updated_json = {"type": "doc", "content": [{"type": "paragraph", "text": "Updated legacy comment"}]}
issue_comment.comment_html = updated_html
issue_comment.comment_json = updated_json
issue_comment.save()
# Assert
issue_comment.refresh_from_db()
# Verify description was created
assert issue_comment.description_id is not None
description = Description.objects.get(pk=issue_comment.description_id)
assert description.description_html == updated_html
assert description.description_json == updated_json
assert description.description_stripped == "Updated legacy comment"
@pytest.mark.django_db
def test_issue_comment_strips_html_tags(self, workspace, project, issue, create_user):
"""Test that HTML tags are properly stripped from comment_html"""
# Arrange
comment_html = "<p>This is <strong>bold</strong> and <em>italic</em> text</p>"
comment_json = {"type": "doc", "content": []}
# Act
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=comment_html,
comment_json=comment_json,
created_by=create_user,
updated_by=create_user,
)
# Assert
assert issue_comment.comment_stripped == "This is bold and italic text"
# Verify description has the same stripped content
description = Description.objects.get(pk=issue_comment.description_id)
assert description.description_stripped == "This is bold and italic text"
@pytest.mark.django_db
def test_issue_comment_empty_html_creates_empty_stripped(self, workspace, project, issue, create_user):
"""Test that empty HTML results in empty comment_stripped"""
# Arrange
comment_html = ""
comment_json = {"type": "doc", "content": []}
# Act
issue_comment = IssueComment.objects.create(
workspace=workspace,
project=project,
issue=issue,
comment_html=comment_html,
comment_json=comment_json,
created_by=create_user,
updated_by=create_user,
)
# Assert
assert issue_comment.comment_stripped == ""
# Verify description was created with empty stripped content
description = Description.objects.get(pk=issue_comment.description_id)
assert description.description_stripped is None
@@ -0,0 +1,48 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from uuid import uuid4
from plane.db.models import Workspace, WorkspaceMember
@pytest.mark.unit
class TestWorkspaceModel:
"""Test the Workspace model"""
@pytest.mark.django_db
def test_workspace_creation(self, create_user):
"""Test creating a workspace"""
# Create a workspace
workspace = Workspace.objects.create(
name="Test Workspace", slug="test-workspace", id=uuid4(), owner=create_user
)
# Verify it was created
assert workspace.id is not None
assert workspace.name == "Test Workspace"
assert workspace.slug == "test-workspace"
assert workspace.owner == create_user
@pytest.mark.django_db
def test_workspace_member_creation(self, create_user):
"""Test creating a workspace member"""
# Create a workspace
workspace = Workspace.objects.create(
name="Test Workspace", slug="test-workspace", id=uuid4(), owner=create_user
)
# Create a workspace member
workspace_member = WorkspaceMember.objects.create(
workspace=workspace,
member=create_user,
role=20, # Admin role
)
# Verify it was created
assert workspace_member.id is not None
assert workspace_member.workspace == workspace
assert workspace_member.member == create_user
assert workspace_member.role == 20
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,73 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from plane.db.models import (
Workspace,
Project,
Issue,
User,
IssueAssignee,
WorkspaceMember,
ProjectMember,
)
from plane.app.serializers.workspace import IssueRecentVisitSerializer
from django.utils import timezone
@pytest.mark.unit
class TestIssueRecentVisitSerializer:
"""Test the IssueRecentVisitSerializer"""
def test_issue_recent_visit_serializer_fields(self, db):
"""Test that the serializer includes the correct fields"""
test_user_1 = User.objects.create(email="test_user_1@example.com", first_name="Test", last_name="User")
# To test for deleted issue assignee
test_user_2 = User.objects.create(
email="test_user_2@example.com",
first_name="Other",
last_name="User",
username="some user name",
)
workspace = Workspace.objects.create(name="Test Workspace", slug="test-workspace", owner=test_user_1)
WorkspaceMember.objects.create(member=test_user_2, role=15, workspace=workspace)
project = Project.objects.create(name="Test Project", identifier="test-project", workspace=workspace)
ProjectMember.objects.create(project=project, member=test_user_2)
issue = Issue.objects.create(
name="Test Issue",
workspace=workspace,
project=project,
)
IssueAssignee.objects.create(issue=issue, assignee=test_user_1, project=project)
# Deleted issue assignee
IssueAssignee.objects.create(
issue=issue,
assignee=test_user_2,
project=project,
deleted_at=timezone.now(),
)
serialized_data = IssueRecentVisitSerializer(
issue,
).data
# Check fields are present and correct
assert "name" in serialized_data
assert "assignees" in serialized_data
assert "project_identifier" in serialized_data
assert serialized_data["name"] == "Test Issue"
assert serialized_data["project_identifier"] == "TEST-PROJECT"
# Only including non-deleted issue assignees
assert serialized_data["assignees"] == [test_user_1.id]
@@ -0,0 +1,41 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from plane.app.serializers import LabelSerializer
from plane.db.models import Project, Label
@pytest.mark.unit
class TestLabelSerializer:
"""Test the LabelSerializer"""
@pytest.mark.django_db
def test_label_serializer_create_valid_data(self, db, workspace):
"""Test creating a label with valid data"""
project = Project.objects.create(name="Test Project", identifier="TEST", workspace=workspace)
serializer = LabelSerializer(
data={"name": "Test Label"},
context={"project_id": project.id},
)
assert serializer.is_valid()
assert serializer.errors == {}
serializer.save(project_id=project.id)
label = Label.objects.all().first()
assert label.name == "Test Label"
assert label.project == project
assert label
@pytest.mark.django_db
def test_label_serializer_create_duplicate_name(self, db, workspace):
"""Test creating a label with a duplicate name"""
project = Project.objects.create(name="Test Project", identifier="TEST", workspace=workspace)
Label.objects.create(name="Test Label", project=project)
serializer = LabelSerializer(data={"name": "Test Label"}, context={"project_id": project.id})
assert not serializer.is_valid()
assert serializer.errors == {"name": ["LABEL_NAME_ALREADY_EXISTS"]}
@@ -0,0 +1,54 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from uuid import uuid4
from plane.api.serializers import WorkspaceLiteSerializer
from plane.db.models import Workspace, User
@pytest.mark.unit
class TestWorkspaceLiteSerializer:
"""Test the WorkspaceLiteSerializer"""
def test_workspace_lite_serializer_fields(self, db):
"""Test that the serializer includes the correct fields"""
# Create a user to be the owner
owner = User.objects.create(email="test@example.com", first_name="Test", last_name="User")
# Create a workspace with explicit ID to test serialization
workspace_id = uuid4()
workspace = Workspace.objects.create(name="Test Workspace", slug="test-workspace", id=workspace_id, owner=owner)
# Serialize the workspace
serialized_data = WorkspaceLiteSerializer(workspace).data
# Check fields are present and correct
assert "name" in serialized_data
assert "slug" in serialized_data
assert "id" in serialized_data
assert serialized_data["name"] == "Test Workspace"
assert serialized_data["slug"] == "test-workspace"
assert str(serialized_data["id"]) == str(workspace_id)
def test_workspace_lite_serializer_read_only(self, db):
"""Test that the serializer fields are read-only"""
# Create a user to be the owner
owner = User.objects.create(email="test2@example.com", first_name="Test", last_name="User")
# Create a workspace
workspace = Workspace.objects.create(name="Test Workspace", slug="test-workspace", id=uuid4(), owner=owner)
# Try to update via serializer
serializer = WorkspaceLiteSerializer(workspace, data={"name": "Updated Name", "slug": "updated-slug"})
# Serializer should be valid (since read-only fields are ignored)
assert serializer.is_valid()
# Save should not update the read-only fields
updated_workspace = serializer.save()
assert updated_workspace.name == "Test Workspace"
assert updated_workspace.slug == "test-workspace"
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,206 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import os
from unittest.mock import Mock, patch
import pytest
from plane.settings.storage import S3Storage
@pytest.mark.unit
class TestS3StorageSignedURLExpiration:
"""Test the configurable signed URL expiration in S3Storage"""
@patch.dict(os.environ, {}, clear=True)
@patch("plane.settings.storage.boto3")
def test_default_expiration_without_env_variable(self, mock_boto3):
"""Test that default expiration is 3600 seconds when env variable is not set"""
# Mock the boto3 client
mock_boto3.client.return_value = Mock()
# Create S3Storage instance without SIGNED_URL_EXPIRATION env variable
storage = S3Storage()
# Assert default expiration is 3600
assert storage.signed_url_expiration == 3600
@patch.dict(os.environ, {"SIGNED_URL_EXPIRATION": "30"}, clear=True)
@patch("plane.settings.storage.boto3")
def test_custom_expiration_with_env_variable(self, mock_boto3):
"""Test that expiration is read from SIGNED_URL_EXPIRATION env variable"""
# Mock the boto3 client
mock_boto3.client.return_value = Mock()
# Create S3Storage instance with SIGNED_URL_EXPIRATION=30
storage = S3Storage()
# Assert expiration is 30
assert storage.signed_url_expiration == 30
@patch.dict(os.environ, {"SIGNED_URL_EXPIRATION": "300"}, clear=True)
@patch("plane.settings.storage.boto3")
def test_custom_expiration_multiple_values(self, mock_boto3):
"""Test that expiration works with different custom values"""
# Mock the boto3 client
mock_boto3.client.return_value = Mock()
# Create S3Storage instance with SIGNED_URL_EXPIRATION=300
storage = S3Storage()
# Assert expiration is 300
assert storage.signed_url_expiration == 300
@patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "test-key",
"AWS_SECRET_ACCESS_KEY": "test-secret",
"AWS_S3_BUCKET_NAME": "test-bucket",
"AWS_REGION": "us-east-1",
},
clear=True,
)
@patch("plane.settings.storage.boto3")
def test_generate_presigned_post_uses_default_expiration(self, mock_boto3):
"""Test that generate_presigned_post uses the configured default expiration"""
# Mock the boto3 client and its response
mock_s3_client = Mock()
mock_s3_client.generate_presigned_post.return_value = {
"url": "https://test-url.com",
"fields": {},
}
mock_boto3.client.return_value = mock_s3_client
# Create S3Storage instance
storage = S3Storage()
# Call generate_presigned_post without explicit expiration
storage.generate_presigned_post("test-object", "image/png", 1024)
# Assert that the boto3 method was called with the default expiration (3600)
mock_s3_client.generate_presigned_post.assert_called_once()
call_kwargs = mock_s3_client.generate_presigned_post.call_args[1]
assert call_kwargs["ExpiresIn"] == 3600
@patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "test-key",
"AWS_SECRET_ACCESS_KEY": "test-secret",
"AWS_S3_BUCKET_NAME": "test-bucket",
"AWS_REGION": "us-east-1",
"SIGNED_URL_EXPIRATION": "60",
},
clear=True,
)
@patch("plane.settings.storage.boto3")
def test_generate_presigned_post_uses_custom_expiration(self, mock_boto3):
"""Test that generate_presigned_post uses custom expiration from env variable"""
# Mock the boto3 client and its response
mock_s3_client = Mock()
mock_s3_client.generate_presigned_post.return_value = {
"url": "https://test-url.com",
"fields": {},
}
mock_boto3.client.return_value = mock_s3_client
# Create S3Storage instance with SIGNED_URL_EXPIRATION=60
storage = S3Storage()
# Call generate_presigned_post without explicit expiration
storage.generate_presigned_post("test-object", "image/png", 1024)
# Assert that the boto3 method was called with custom expiration (60)
mock_s3_client.generate_presigned_post.assert_called_once()
call_kwargs = mock_s3_client.generate_presigned_post.call_args[1]
assert call_kwargs["ExpiresIn"] == 60
@patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "test-key",
"AWS_SECRET_ACCESS_KEY": "test-secret",
"AWS_S3_BUCKET_NAME": "test-bucket",
"AWS_REGION": "us-east-1",
},
clear=True,
)
@patch("plane.settings.storage.boto3")
def test_generate_presigned_url_uses_default_expiration(self, mock_boto3):
"""Test that generate_presigned_url uses the configured default expiration"""
# Mock the boto3 client and its response
mock_s3_client = Mock()
mock_s3_client.generate_presigned_url.return_value = "https://test-url.com"
mock_boto3.client.return_value = mock_s3_client
# Create S3Storage instance
storage = S3Storage()
# Call generate_presigned_url without explicit expiration
storage.generate_presigned_url("test-object")
# Assert that the boto3 method was called with the default expiration (3600)
mock_s3_client.generate_presigned_url.assert_called_once()
call_kwargs = mock_s3_client.generate_presigned_url.call_args[1]
assert call_kwargs["ExpiresIn"] == 3600
@patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "test-key",
"AWS_SECRET_ACCESS_KEY": "test-secret",
"AWS_S3_BUCKET_NAME": "test-bucket",
"AWS_REGION": "us-east-1",
"SIGNED_URL_EXPIRATION": "30",
},
clear=True,
)
@patch("plane.settings.storage.boto3")
def test_generate_presigned_url_uses_custom_expiration(self, mock_boto3):
"""Test that generate_presigned_url uses custom expiration from env variable"""
# Mock the boto3 client and its response
mock_s3_client = Mock()
mock_s3_client.generate_presigned_url.return_value = "https://test-url.com"
mock_boto3.client.return_value = mock_s3_client
# Create S3Storage instance with SIGNED_URL_EXPIRATION=30
storage = S3Storage()
# Call generate_presigned_url without explicit expiration
storage.generate_presigned_url("test-object")
# Assert that the boto3 method was called with custom expiration (30)
mock_s3_client.generate_presigned_url.assert_called_once()
call_kwargs = mock_s3_client.generate_presigned_url.call_args[1]
assert call_kwargs["ExpiresIn"] == 30
@patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "test-key",
"AWS_SECRET_ACCESS_KEY": "test-secret",
"AWS_S3_BUCKET_NAME": "test-bucket",
"AWS_REGION": "us-east-1",
"SIGNED_URL_EXPIRATION": "30",
},
clear=True,
)
@patch("plane.settings.storage.boto3")
def test_explicit_expiration_overrides_default(self, mock_boto3):
"""Test that explicit expiration parameter overrides the default"""
# Mock the boto3 client and its response
mock_s3_client = Mock()
mock_s3_client.generate_presigned_url.return_value = "https://test-url.com"
mock_boto3.client.return_value = mock_s3_client
# Create S3Storage instance with SIGNED_URL_EXPIRATION=30
storage = S3Storage()
# Call generate_presigned_url with explicit expiration=120
storage.generate_presigned_url("test-object", expiration=120)
# Assert that the boto3 method was called with explicit expiration (120)
mock_s3_client.generate_presigned_url.assert_called_once()
call_kwargs = mock_s3_client.generate_presigned_url.call_args[1]
assert call_kwargs["ExpiresIn"] == 120
@@ -0,0 +1,4 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
@@ -0,0 +1,257 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from plane.utils.url import (
contains_url,
is_valid_url,
normalize_url_path,
)
@pytest.mark.unit
class TestContainsURL:
"""Test the contains_url function"""
def test_contains_url_with_http_protocol(self):
"""Test contains_url with HTTP protocol URLs"""
assert contains_url("Check out http://example.com") is True
assert contains_url("Visit http://google.com/search") is True
assert contains_url("http://localhost:8000") is True
def test_contains_url_with_https_protocol(self):
"""Test contains_url with HTTPS protocol URLs"""
assert contains_url("Check out https://example.com") is True
assert contains_url("Visit https://google.com/search") is True
assert contains_url("https://secure.example.com") is True
def test_contains_url_with_www_prefix(self):
"""Test contains_url with www prefix"""
assert contains_url("Visit www.example.com") is True
assert contains_url("Check www.google.com") is True
assert contains_url("Go to www.test-site.org") is True
def test_contains_url_with_domain_patterns(self):
"""Test contains_url with domain patterns"""
assert contains_url("Visit example.com") is True
assert contains_url("Check google.org") is True
assert contains_url("Go to test-site.co.uk") is True
assert contains_url("Visit sub.domain.com") is True
def test_contains_url_with_ip_addresses(self):
"""Test contains_url with IP addresses"""
assert contains_url("Connect to 192.168.1.1") is True
assert contains_url("Visit 10.0.0.1") is True
assert contains_url("Check 127.0.0.1") is True
assert contains_url("Go to 8.8.8.8") is True
def test_contains_url_case_insensitive(self):
"""Test contains_url is case insensitive"""
assert contains_url("Check HTTP://EXAMPLE.COM") is True
assert contains_url("Visit WWW.GOOGLE.COM") is True
assert contains_url("Go to Https://Test.Com") is True
def test_contains_url_with_no_urls(self):
"""Test contains_url with text that doesn't contain URLs"""
assert contains_url("This is just plain text") is False
assert contains_url("No URLs here!") is False
assert contains_url("com org net") is False # Just TLD words
assert contains_url("192.168") is False # Incomplete IP
assert contains_url("") is False # Empty string
def test_contains_url_edge_cases(self):
"""Test contains_url with edge cases"""
assert contains_url("example.c") is False # TLD too short
assert contains_url("999.999.999.999") is False # Invalid IP (octets > 255)
assert contains_url("just-a-hyphen") is False # No domain
assert contains_url("www.") is False # Incomplete www - needs at least one char after dot
def test_contains_url_length_limit_under_1000(self):
"""Test contains_url with input under 1000 characters containing URLs"""
# Create a string under 1000 characters with a URL
text_with_url = "a" * 970 + " https://example.com" # 970 + 1 + 19 = 990 chars
assert len(text_with_url) < 1000
assert contains_url(text_with_url) is True
# Test with exactly 1000 characters
text_exact_1000 = "a" * 981 + "https://example.com" # 981 + 19 = 1000 chars
assert len(text_exact_1000) == 1000
assert contains_url(text_exact_1000) is True
def test_contains_url_length_limit_over_1000(self):
"""Test contains_url with input over 1000 characters returns False"""
# Create a string over 1000 characters with a URL
text_with_url = "a" * 982 + "https://example.com" # 982 + 19 = 1001 chars
assert len(text_with_url) > 1000
assert contains_url(text_with_url) is False
# Test with much longer input
long_text_with_url = "a" * 5000 + " https://example.com"
assert contains_url(long_text_with_url) is False
def test_contains_url_length_limit_exactly_1000(self):
"""Test contains_url with input exactly 1000 characters"""
# Test with exactly 1000 characters without URL
text_no_url = "a" * 1000
assert len(text_no_url) == 1000
assert contains_url(text_no_url) is False
# Test with exactly 1000 characters with URL at the end
text_with_url = "a" * 981 + "https://example.com" # 981 + 19 = 1000 chars
assert len(text_with_url) == 1000
assert contains_url(text_with_url) is True
def test_contains_url_line_length_scenarios(self):
"""Test contains_url with realistic line length scenarios"""
# Test with multiline input where total is under 1000 but we test line processing
# Short lines with URL
multiline_short = "Line 1\nLine 2 with https://example.com\nLine 3"
assert contains_url(multiline_short) is True
# Multiple lines under total limit
multiline_text = "a" * 200 + "\n" + "b" * 200 + "https://example.com\n" + "c" * 200
assert len(multiline_text) < 1000
assert contains_url(multiline_text) is True
def test_contains_url_total_length_vs_line_length(self):
"""Test the interaction between total length limit and line processing"""
# Test that total length limit takes precedence
# Even if individual lines would be processed, total > 1000 means immediate False
over_limit_text = "a" * 1001 # No URL, but over total limit
assert contains_url(over_limit_text) is False
# Test that under total limit, line processing works normally
under_limit_with_url = "a" * 900 + "https://example.com" # 919 chars total
assert len(under_limit_with_url) < 1000
assert contains_url(under_limit_with_url) is True
def test_contains_url_multiline_mixed_lengths(self):
"""Test contains_url with multiple lines of different lengths"""
# Test realistic multiline scenario under 1000 chars total
multiline_text = (
"Short line\n"
+ "a" * 400
+ "https://example.com\n" # Line with URL
+ "b" * 300 # Another line
)
assert len(multiline_text) < 1000
assert contains_url(multiline_text) is True
# Test multiline without URLs
multiline_no_url = "Short line\n" + "a" * 400 + "\n" + "b" * 300
assert len(multiline_no_url) < 1000
assert contains_url(multiline_no_url) is False
def test_contains_url_edge_cases_with_length_limits(self):
"""Test contains_url edge cases related to length limits"""
# Empty string
assert contains_url("") is False
# Very short string with URL
assert contains_url("http://a.co") is True
# String with newlines and mixed content
mixed_content = "Line 1\nLine 2 with https://example.com\nLine 3"
assert contains_url(mixed_content) is True
# String with many newlines under total limit
many_newlines = "\n" * 500 + "https://example.com"
assert len(many_newlines) < 1000
assert contains_url(many_newlines) is True
@pytest.mark.unit
class TestIsValidURL:
"""Test the is_valid_url function"""
def test_is_valid_url_with_valid_urls(self):
"""Test is_valid_url with valid URLs"""
assert is_valid_url("https://example.com") is True
assert is_valid_url("http://google.com") is True
assert is_valid_url("https://sub.domain.com/path") is True
assert is_valid_url("http://localhost:8000") is True
assert is_valid_url("https://example.com/path?query=1") is True
assert is_valid_url("ftp://files.example.com") is True
def test_is_valid_url_with_invalid_urls(self):
"""Test is_valid_url with invalid URLs"""
assert is_valid_url("not a url") is False
assert is_valid_url("example.com") is False # No scheme
assert is_valid_url("https://") is False # No netloc
assert is_valid_url("") is False # Empty string
assert is_valid_url("://example.com") is False # No scheme
assert is_valid_url("https:/example.com") is False # Malformed
def test_is_valid_url_with_non_string_input(self):
"""Test is_valid_url with non-string input"""
assert is_valid_url(None) is False
assert is_valid_url([]) is False
assert is_valid_url({}) is False
def test_is_valid_url_with_special_schemes(self):
"""Test is_valid_url with special URL schemes"""
assert is_valid_url("ftp://ftp.example.com") is True
assert is_valid_url("mailto:user@example.com") is False
assert is_valid_url("file:///path/to/file") is False
@pytest.mark.unit
class TestNormalizeURLPath:
"""Test the normalize_url_path function"""
def test_normalize_url_path_with_multiple_slashes(self):
"""Test normalize_url_path with multiple consecutive slashes"""
result = normalize_url_path("https://example.com//foo///bar//baz")
assert result == "https://example.com/foo/bar/baz"
def test_normalize_url_path_with_query_and_fragment(self):
"""Test normalize_url_path preserves query and fragment"""
result = normalize_url_path("https://example.com//foo///bar//baz?x=1&y=2#fragment")
assert result == "https://example.com/foo/bar/baz?x=1&y=2#fragment"
def test_normalize_url_path_with_no_redundant_slashes(self):
"""Test normalize_url_path with already normalized URL"""
url = "https://example.com/foo/bar/baz?x=1#fragment"
result = normalize_url_path(url)
assert result == url
def test_normalize_url_path_with_root_path(self):
"""Test normalize_url_path with root path"""
result = normalize_url_path("https://example.com//")
assert result == "https://example.com/"
def test_normalize_url_path_with_empty_path(self):
"""Test normalize_url_path with empty path"""
result = normalize_url_path("https://example.com")
assert result == "https://example.com"
def test_normalize_url_path_with_complex_path(self):
"""Test normalize_url_path with complex path structure"""
result = normalize_url_path("https://example.com///api//v1///users//123//profile")
assert result == "https://example.com/api/v1/users/123/profile"
def test_normalize_url_path_with_different_schemes(self):
"""Test normalize_url_path with different URL schemes"""
# HTTP
result = normalize_url_path("http://example.com//path")
assert result == "http://example.com/path"
# FTP
result = normalize_url_path("ftp://ftp.example.com//files//document.txt")
assert result == "ftp://ftp.example.com/files/document.txt"
def test_normalize_url_path_with_port(self):
"""Test normalize_url_path with port number"""
result = normalize_url_path("https://example.com:8080//api//v1")
assert result == "https://example.com:8080/api/v1"
def test_normalize_url_path_edge_cases(self):
"""Test normalize_url_path with edge cases"""
# Many consecutive slashes
result = normalize_url_path("https://example.com///////path")
assert result == "https://example.com/path"
# Mixed single and multiple slashes
result = normalize_url_path("https://example.com/a//b/c///d")
assert result == "https://example.com/a/b/c/d"
@@ -0,0 +1,53 @@
# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import uuid
import pytest
from plane.utils.uuid import is_valid_uuid, convert_uuid_to_integer
@pytest.mark.unit
class TestUUIDUtils:
"""Test the UUID utilities"""
def test_is_valid_uuid_with_valid_uuid(self):
"""Test is_valid_uuid with a valid UUID"""
# Generate a valid UUID
valid_uuid = str(uuid.uuid4())
assert is_valid_uuid(valid_uuid) is True
def test_is_valid_uuid_with_invalid_uuid(self):
"""Test is_valid_uuid with invalid UUID strings"""
# Test with different invalid formats
assert is_valid_uuid("not-a-uuid") is False
assert is_valid_uuid("123456789") is False
assert is_valid_uuid("") is False
assert is_valid_uuid("00000000-0000-0000-0000-000000000000") is False # This is a valid UUID but version 1
def test_convert_uuid_to_integer(self):
"""Test convert_uuid_to_integer function"""
# Create a known UUID
test_uuid = uuid.UUID("f47ac10b-58cc-4372-a567-0e02b2c3d479")
# Convert to integer
result = convert_uuid_to_integer(test_uuid)
# Check that the result is an integer
assert isinstance(result, int)
# Ensure consistent results with the same input
assert convert_uuid_to_integer(test_uuid) == result
# Different UUIDs should produce different integers
different_uuid = uuid.UUID("550e8400-e29b-41d4-a716-446655440000")
assert convert_uuid_to_integer(different_uuid) != result
def test_convert_uuid_to_integer_string_input(self):
"""Test convert_uuid_to_integer handles string UUID"""
# Test with a UUID string
test_uuid_str = "f47ac10b-58cc-4372-a567-0e02b2c3d479"
test_uuid = uuid.UUID(test_uuid_str)
# Should get the same result whether passing UUID or string
assert convert_uuid_to_integer(test_uuid) == convert_uuid_to_integer(test_uuid_str)