# Django Concurrent Test
A production-ready Django package for safe and configurable concurrent testing with isolated databases, timing analytics, and concurrency simulation middleware.
## 🚀 Features
- **🔒 Secure Database Templating**: Zero-config parallel testing with isolated database instances
- **⚡ Concurrent Test Execution**: ThreadPoolExecutor and asyncio-based concurrency
- **📊 Timing Analytics**: Comprehensive test timing analysis and benchmarking
- **🛡️ Concurrency Safety**: Middleware for detecting race conditions and state mutations
- **🔧 Runtime Configuration**: Dynamic worker scaling and timeout management
- **📈 Performance Monitoring**: Connection pooling, resource monitoring, and metrics
- **🎯 DRF Integration**: Optional Django REST Framework compatibility
- **📋 JUnit XML Output**: CI/CD friendly test reporting
- **🔍 Telemetry-Free**: No data collection or external dependencies
## 📦 Installation
```bash
pip install django-concurrent-test
```
## 🏃♂️ Quick Start
### Basic Usage
```python
# settings.py
INSTALLED_APPS = [
# ... your apps
'django_concurrent_test',
]
# Enable concurrent testing
DJANGO_ENABLE_CONCURRENT = True
```
### Command Line Usage
```bash
# Run tests with concurrent execution
pytest --concurrent
# Specify number of workers
pytest --concurrent --workers 4
# Set timeouts
pytest --concurrent --timeout 300 --test-timeout 60
# Export timing data
pytest --concurrent --export-timings results.json
# Import previous timings and export to CSV
pytest --concurrent --import-timings results.json --export-timings-csv results.csv
```
## 🔧 Configuration
### Environment Variables
```bash
# Enable concurrent testing
export DJANGO_ENABLE_CONCURRENT=True
# Configure workers (auto-detected if not set)
export DJANGO_TEST_WORKERS=4
# Set timeouts
export DJANGO_TEST_TIMEOUT=300
export DJANGO_TEST_BENCHMARK=True
```
### Django Settings
```python
# settings.py
CONCURRENT_TEST = {
'ENABLED': True,
'WORKERS': 4, # Auto-detected if not set
'TIMEOUT': 300,
'TEST_TIMEOUT': 60,
'WORKER_TIMEOUT': 120,
'BENCHMARK': True,
'EXPORT_TIMINGS': 'test_timings.json',
}
```
## 🛡️ Concurrency Safety
### Using `assert_concurrent_safety`
Test your functions for concurrent execution safety:
```python
from django_concurrent_test.middleware import assert_concurrent_safety
def test_user_creation():
"""Test that user creation is safe for concurrent execution."""
def create_user():
from django.contrib.auth.models import User
return User.objects.create_user(
username=f'user_{time.time()}',
email='test@example.com'
)
# This will run the function concurrently and check for race conditions
assert_concurrent_safety(create_user, max_workers=4, iterations=10)
```
### Using `simulate_concurrent_requests`
Simulate concurrent request scenarios:
```python
from django_concurrent_test.middleware import simulate_concurrent_requests
from django.test import RequestFactory
def test_api_endpoint_concurrency():
"""Test API endpoint under concurrent load."""
factory = RequestFactory()
def make_request():
request = factory.get('/api/users/')
response = your_view_function(request)
return response.status_code
# Simulate 10 concurrent requests
results = simulate_concurrent_requests(make_request, num_requests=10)
# Check results
successful = [r for r in results if r['status'] == 'success']
assert len(successful) == 10
```
## 🔌 Middleware Integration
### Auto-Registration
The middleware can be auto-registered during pytest sessions:
```python
# conftest.py
import pytest
from django_concurrent_test.middleware import auto_register_middleware
@pytest.fixture(scope='session', autouse=True)
def setup_concurrent_middleware():
"""Auto-register concurrent testing middleware."""
added_middleware = auto_register_middleware()
if added_middleware:
print(f"Auto-registered middleware: {added_middleware}")
```
### Manual Configuration
Add middleware to your Django settings:
```python
# settings.py
MIDDLEWARE = [
# ... existing middleware
'django_concurrent_test.middleware.ConcurrentSafetyMiddleware',
'django_concurrent_test.middleware.StateMutationMiddleware',
'django_concurrent_test.middleware.ConcurrencySimulationMiddleware',
]
```
### Runtime Configuration
Configure middleware behavior at runtime:
```python
from django_concurrent_test.middleware import (
set_test_override,
concurrent_test_context
)
# Adjust middleware behavior
set_test_override('delay_range', (0.2, 0.8))
set_test_override('probability', 0.5)
# Use context manager for temporary changes
with concurrent_test_context():
# All middleware uses testing configuration
run_tests()
```
## 🍰 DRF Integration
### Testing DRF Viewsets
```python
from django_concurrent_test.middleware import assert_concurrent_safety
from rest_framework.test import APITestCase
from rest_framework import status
class UserViewSetTest(APITestCase):
def test_concurrent_user_creation(self):
"""Test concurrent user creation via DRF."""
def create_user_via_api():
data = {
'username': f'user_{time.time()}',
'email': 'test@example.com',
'password': 'testpass123'
}
response = self.client.post('/api/users/', data)
return response.status_code
# Test concurrent API calls
assert_concurrent_safety(create_user_via_api, max_workers=4, iterations=5)
```
### Testing DRF Serializers
```python
from django_concurrent_test.middleware import assert_concurrent_safety
from rest_framework import serializers
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['username', 'email']
def test_serializer_concurrency():
"""Test serializer validation under concurrent load."""
def validate_user_data():
data = {
'username': f'user_{time.time()}',
'email': 'test@example.com'
}
serializer = UserSerializer(data=data)
return serializer.is_valid()
assert_concurrent_safety(validate_user_data, max_workers=4, iterations=10)
```
## 📊 Timing Analytics
### Export and Import Timing Data
```python
from django_concurrent_test.timing_utils import (
load_timings,
save_timings,
filter_timings,
export_timings_to_csv
)
# Load timing data
timings = load_timings('test_timings.json')
# Filter slow tests
slow_tests = filter_timings(timings, min_duration=5.0)
# Export to CSV
export_timings_to_csv(slow_tests, 'slow_tests.csv')
# Analyze timing data
for test_name, timing_data in slow_tests.items():
print(f"{test_name}: {timing_data['duration']:.2f}s")
```
### Benchmark Analysis
```python
from django_concurrent_test.timing_utils import get_slowest_tests, get_fastest_tests
# Get performance insights
slowest = get_slowest_tests(timings, count=5)
fastest = get_fastest_tests(timings, count=5)
print("Slowest tests:")
for test_name, duration in slowest:
print(f" {test_name}: {duration:.2f}s")
print("Fastest tests:")
for test_name, duration in fastest:
print(f" {test_name}: {duration:.2f}s")
```
## 🔒 Security Features
### Environment Validation
```python
from django_concurrent_test.security import (
security_context,
get_safe_worker_count,
validate_environment
)
# Validate environment before testing
with security_context():
worker_count = get_safe_worker_count()
print(f"Safe worker count: {worker_count}")
# Manual validation
validate_environment()
```
### Resource Monitoring
```python
from django_concurrent_test.security import check_system_resources
# Check system resources
resources = check_system_resources()
print(f"Available memory: {resources['memory_available_gb']:.1f}GB")
print(f"CPU cores: {resources['cpu_count']}")
print(f"Safe worker count: {resources['safe_worker_count']}")
```
## 🧪 Advanced Testing
### Custom Test Runner
```python
from django_concurrent_test.runner import ConcurrentTestRunner
class CustomConcurrentRunner(ConcurrentTestRunner):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.custom_metrics = {}
def run_tests(self, test_labels, **kwargs):
# Custom pre-test setup
self.setup_custom_environment()
# Run tests with concurrent execution
failures = super().run_tests(test_labels, **kwargs)
# Custom post-test cleanup
self.cleanup_custom_environment()
return failures
```
### Database Isolation Testing
```python
from django_concurrent_test.db import verify_database_isolation
def test_database_isolation():
"""Verify that worker databases are properly isolated."""
# Run concurrent tests
runner = ConcurrentTestRunner()
failures = runner.run_tests(['myapp.tests'])
# Verify isolation
worker_connections = runner.get_worker_connections()
isolation_verified = verify_database_isolation(worker_connections)
assert isolation_verified, "Database isolation verification failed"
```
## 📈 Performance Monitoring
### Connection Pool Statistics
```python
from django_concurrent_test.db import get_connection_pool_stats
# Get connection pool metrics
stats = get_connection_pool_stats()
print(f"Active connections: {stats.get('active', 0)}")
print(f"Pooled connections: {stats.get('pooled', 0)}")
print(f"Connection hits: {stats.get('hits', 0)}")
print(f"Connection misses: {stats.get('misses', 0)}")
```
### Memory-Based Scaling
The package automatically scales worker count based on available memory:
```python
from django_concurrent_test.runner import ConcurrentTestRunner
# Memory-based scaling is automatic
runner = ConcurrentTestRunner()
# Worker count will be calculated based on available memory
```
## 🚨 Error Handling
### Timeout Management
```python
from django_concurrent_test.exceptions import TestTimeoutException
def test_with_timeout():
"""Test with custom timeout handling."""
try:
# Run test with timeout
result = run_test_with_timeout(test_function, timeout=30)
assert result is not None
except TestTimeoutException:
pytest.fail("Test timed out")
```
### Database Error Recovery
```python
from django_concurrent_test.db import wait_for_database_ready
def test_database_recovery():
"""Test database connection recovery."""
# Wait for database to be ready
ready = wait_for_database_ready('test_db', timeout=30)
assert ready, "Database not ready within timeout"
```
## 🔧 Development
### Running Tests
```bash
# Run package tests
pytest tests/
# Run with coverage
pytest --cov=django_concurrent_test tests/
# Run with concurrent execution
pytest --concurrent tests/
```
### Building Documentation
```bash
# Build package
python -m build
# Upload to PyPI
twine upload dist/*
```
## 📋 Requirements
- Python 3.9+
- Django 3.2+
- PostgreSQL 10+ or MySQL 5.7+ (for production features)
- SQLite (for development)
## 🤝 Contributing
1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests for new functionality
5. Ensure all tests pass
6. Submit a pull request
## 📞 Contact
- **Email**: ranaehtashamali1@gmail.com
- **Phone**: +923224712517
- **GitHub**: [@RanaEhtashamAli](https://github.com/RanaEhtashamAli)
## 📄 License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## 🙏 Acknowledgments
- Django team for the excellent testing framework
- PostgreSQL and MySQL communities for database support
- All contributors who have helped improve this package
## 📞 Support
- **Issues**: [GitHub Issues](https://github.com/RanaEhtashamAli/django-concurrent-test/issues)
- **Documentation**: [Read the Docs](https://django-concurrent-test.readthedocs.io/)
- **Discussions**: [GitHub Discussions](https://github.com/RanaEhtashamAli/django-concurrent-test/discussions)
- **Email**: ranaehtashamali1@gmail.com
- **Phone**: +923224712517
---
**Made with ❤️ for the Django community**
Raw data
{
"_id": null,
"home_page": "https://github.com/example/django-concurrent-test",
"name": "django-concurrent-test",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "Rana Ehtasham Ali <ranaehtashamali1@gmail.com>",
"keywords": "django, testing, concurrent, parallel, database, postgresql, mysql, middleware, analytics, benchmarking",
"author": "Django Concurrent Test Team",
"author_email": "Rana Ehtasham Ali <ranaehtashamali1@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/07/2b/2183bf69dd7b72319980cea9cbff412622039458061d1fdb93fe5917142e/django_concurrent_test-1.0.0.tar.gz",
"platform": null,
"description": "# Django Concurrent Test\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\nA production-ready Django package for safe and configurable concurrent testing with isolated databases, timing analytics, and concurrency simulation middleware.\r\n\r\n## \ud83d\ude80 Features\r\n\r\n- **\ud83d\udd12 Secure Database Templating**: Zero-config parallel testing with isolated database instances\r\n- **\u26a1 Concurrent Test Execution**: ThreadPoolExecutor and asyncio-based concurrency\r\n- **\ud83d\udcca Timing Analytics**: Comprehensive test timing analysis and benchmarking\r\n- **\ud83d\udee1\ufe0f Concurrency Safety**: Middleware for detecting race conditions and state mutations\r\n- **\ud83d\udd27 Runtime Configuration**: Dynamic worker scaling and timeout management\r\n- **\ud83d\udcc8 Performance Monitoring**: Connection pooling, resource monitoring, and metrics\r\n- **\ud83c\udfaf DRF Integration**: Optional Django REST Framework compatibility\r\n- **\ud83d\udccb JUnit XML Output**: CI/CD friendly test reporting\r\n- **\ud83d\udd0d Telemetry-Free**: No data collection or external dependencies\r\n\r\n## \ud83d\udce6 Installation\r\n\r\n```bash\r\npip install django-concurrent-test\r\n```\r\n\r\n## \ud83c\udfc3\u200d\u2642\ufe0f Quick Start\r\n\r\n### Basic Usage\r\n\r\n```python\r\n# settings.py\r\nINSTALLED_APPS = [\r\n # ... your apps\r\n 'django_concurrent_test',\r\n]\r\n\r\n# Enable concurrent testing\r\nDJANGO_ENABLE_CONCURRENT = True\r\n```\r\n\r\n### Command Line Usage\r\n\r\n```bash\r\n# Run tests with concurrent execution\r\npytest --concurrent\r\n\r\n# Specify number of workers\r\npytest --concurrent --workers 4\r\n\r\n# Set timeouts\r\npytest --concurrent --timeout 300 --test-timeout 60\r\n\r\n# Export timing data\r\npytest --concurrent --export-timings results.json\r\n\r\n# Import previous timings and export to CSV\r\npytest --concurrent --import-timings results.json --export-timings-csv results.csv\r\n```\r\n\r\n## \ud83d\udd27 Configuration\r\n\r\n### Environment Variables\r\n\r\n```bash\r\n# Enable concurrent testing\r\nexport DJANGO_ENABLE_CONCURRENT=True\r\n\r\n# Configure workers (auto-detected if not set)\r\nexport DJANGO_TEST_WORKERS=4\r\n\r\n# Set timeouts\r\nexport DJANGO_TEST_TIMEOUT=300\r\nexport DJANGO_TEST_BENCHMARK=True\r\n```\r\n\r\n### Django Settings\r\n\r\n```python\r\n# settings.py\r\nCONCURRENT_TEST = {\r\n 'ENABLED': True,\r\n 'WORKERS': 4, # Auto-detected if not set\r\n 'TIMEOUT': 300,\r\n 'TEST_TIMEOUT': 60,\r\n 'WORKER_TIMEOUT': 120,\r\n 'BENCHMARK': True,\r\n 'EXPORT_TIMINGS': 'test_timings.json',\r\n}\r\n```\r\n\r\n## \ud83d\udee1\ufe0f Concurrency Safety\r\n\r\n### Using `assert_concurrent_safety`\r\n\r\nTest your functions for concurrent execution safety:\r\n\r\n```python\r\nfrom django_concurrent_test.middleware import assert_concurrent_safety\r\n\r\ndef test_user_creation():\r\n \"\"\"Test that user creation is safe for concurrent execution.\"\"\"\r\n \r\n def create_user():\r\n from django.contrib.auth.models import User\r\n return User.objects.create_user(\r\n username=f'user_{time.time()}',\r\n email='test@example.com'\r\n )\r\n \r\n # This will run the function concurrently and check for race conditions\r\n assert_concurrent_safety(create_user, max_workers=4, iterations=10)\r\n```\r\n\r\n### Using `simulate_concurrent_requests`\r\n\r\nSimulate concurrent request scenarios:\r\n\r\n```python\r\nfrom django_concurrent_test.middleware import simulate_concurrent_requests\r\nfrom django.test import RequestFactory\r\n\r\ndef test_api_endpoint_concurrency():\r\n \"\"\"Test API endpoint under concurrent load.\"\"\"\r\n \r\n factory = RequestFactory()\r\n \r\n def make_request():\r\n request = factory.get('/api/users/')\r\n response = your_view_function(request)\r\n return response.status_code\r\n \r\n # Simulate 10 concurrent requests\r\n results = simulate_concurrent_requests(make_request, num_requests=10)\r\n \r\n # Check results\r\n successful = [r for r in results if r['status'] == 'success']\r\n assert len(successful) == 10\r\n```\r\n\r\n## \ud83d\udd0c Middleware Integration\r\n\r\n### Auto-Registration\r\n\r\nThe middleware can be auto-registered during pytest sessions:\r\n\r\n```python\r\n# conftest.py\r\nimport pytest\r\nfrom django_concurrent_test.middleware import auto_register_middleware\r\n\r\n@pytest.fixture(scope='session', autouse=True)\r\ndef setup_concurrent_middleware():\r\n \"\"\"Auto-register concurrent testing middleware.\"\"\"\r\n added_middleware = auto_register_middleware()\r\n if added_middleware:\r\n print(f\"Auto-registered middleware: {added_middleware}\")\r\n```\r\n\r\n### Manual Configuration\r\n\r\nAdd middleware to your Django settings:\r\n\r\n```python\r\n# settings.py\r\nMIDDLEWARE = [\r\n # ... existing middleware\r\n 'django_concurrent_test.middleware.ConcurrentSafetyMiddleware',\r\n 'django_concurrent_test.middleware.StateMutationMiddleware',\r\n 'django_concurrent_test.middleware.ConcurrencySimulationMiddleware',\r\n]\r\n```\r\n\r\n### Runtime Configuration\r\n\r\nConfigure middleware behavior at runtime:\r\n\r\n```python\r\nfrom django_concurrent_test.middleware import (\r\n set_test_override, \r\n concurrent_test_context\r\n)\r\n\r\n# Adjust middleware behavior\r\nset_test_override('delay_range', (0.2, 0.8))\r\nset_test_override('probability', 0.5)\r\n\r\n# Use context manager for temporary changes\r\nwith concurrent_test_context():\r\n # All middleware uses testing configuration\r\n run_tests()\r\n```\r\n\r\n## \ud83c\udf70 DRF Integration\r\n\r\n### Testing DRF Viewsets\r\n\r\n```python\r\nfrom django_concurrent_test.middleware import assert_concurrent_safety\r\nfrom rest_framework.test import APITestCase\r\nfrom rest_framework import status\r\n\r\nclass UserViewSetTest(APITestCase):\r\n def test_concurrent_user_creation(self):\r\n \"\"\"Test concurrent user creation via DRF.\"\"\"\r\n \r\n def create_user_via_api():\r\n data = {\r\n 'username': f'user_{time.time()}',\r\n 'email': 'test@example.com',\r\n 'password': 'testpass123'\r\n }\r\n response = self.client.post('/api/users/', data)\r\n return response.status_code\r\n \r\n # Test concurrent API calls\r\n assert_concurrent_safety(create_user_via_api, max_workers=4, iterations=5)\r\n```\r\n\r\n### Testing DRF Serializers\r\n\r\n```python\r\nfrom django_concurrent_test.middleware import assert_concurrent_safety\r\nfrom rest_framework import serializers\r\n\r\nclass UserSerializer(serializers.ModelSerializer):\r\n class Meta:\r\n model = User\r\n fields = ['username', 'email']\r\n\r\ndef test_serializer_concurrency():\r\n \"\"\"Test serializer validation under concurrent load.\"\"\"\r\n \r\n def validate_user_data():\r\n data = {\r\n 'username': f'user_{time.time()}',\r\n 'email': 'test@example.com'\r\n }\r\n serializer = UserSerializer(data=data)\r\n return serializer.is_valid()\r\n \r\n assert_concurrent_safety(validate_user_data, max_workers=4, iterations=10)\r\n```\r\n\r\n## \ud83d\udcca Timing Analytics\r\n\r\n### Export and Import Timing Data\r\n\r\n```python\r\nfrom django_concurrent_test.timing_utils import (\r\n load_timings, \r\n save_timings, \r\n filter_timings,\r\n export_timings_to_csv\r\n)\r\n\r\n# Load timing data\r\ntimings = load_timings('test_timings.json')\r\n\r\n# Filter slow tests\r\nslow_tests = filter_timings(timings, min_duration=5.0)\r\n\r\n# Export to CSV\r\nexport_timings_to_csv(slow_tests, 'slow_tests.csv')\r\n\r\n# Analyze timing data\r\nfor test_name, timing_data in slow_tests.items():\r\n print(f\"{test_name}: {timing_data['duration']:.2f}s\")\r\n```\r\n\r\n### Benchmark Analysis\r\n\r\n```python\r\nfrom django_concurrent_test.timing_utils import get_slowest_tests, get_fastest_tests\r\n\r\n# Get performance insights\r\nslowest = get_slowest_tests(timings, count=5)\r\nfastest = get_fastest_tests(timings, count=5)\r\n\r\nprint(\"Slowest tests:\")\r\nfor test_name, duration in slowest:\r\n print(f\" {test_name}: {duration:.2f}s\")\r\n\r\nprint(\"Fastest tests:\")\r\nfor test_name, duration in fastest:\r\n print(f\" {test_name}: {duration:.2f}s\")\r\n```\r\n\r\n## \ud83d\udd12 Security Features\r\n\r\n### Environment Validation\r\n\r\n```python\r\nfrom django_concurrent_test.security import (\r\n security_context, \r\n get_safe_worker_count,\r\n validate_environment\r\n)\r\n\r\n# Validate environment before testing\r\nwith security_context():\r\n worker_count = get_safe_worker_count()\r\n print(f\"Safe worker count: {worker_count}\")\r\n\r\n# Manual validation\r\nvalidate_environment()\r\n```\r\n\r\n### Resource Monitoring\r\n\r\n```python\r\nfrom django_concurrent_test.security import check_system_resources\r\n\r\n# Check system resources\r\nresources = check_system_resources()\r\nprint(f\"Available memory: {resources['memory_available_gb']:.1f}GB\")\r\nprint(f\"CPU cores: {resources['cpu_count']}\")\r\nprint(f\"Safe worker count: {resources['safe_worker_count']}\")\r\n```\r\n\r\n## \ud83e\uddea Advanced Testing\r\n\r\n### Custom Test Runner\r\n\r\n```python\r\nfrom django_concurrent_test.runner import ConcurrentTestRunner\r\n\r\nclass CustomConcurrentRunner(ConcurrentTestRunner):\r\n def __init__(self, *args, **kwargs):\r\n super().__init__(*args, **kwargs)\r\n self.custom_metrics = {}\r\n \r\n def run_tests(self, test_labels, **kwargs):\r\n # Custom pre-test setup\r\n self.setup_custom_environment()\r\n \r\n # Run tests with concurrent execution\r\n failures = super().run_tests(test_labels, **kwargs)\r\n \r\n # Custom post-test cleanup\r\n self.cleanup_custom_environment()\r\n \r\n return failures\r\n```\r\n\r\n### Database Isolation Testing\r\n\r\n```python\r\nfrom django_concurrent_test.db import verify_database_isolation\r\n\r\ndef test_database_isolation():\r\n \"\"\"Verify that worker databases are properly isolated.\"\"\"\r\n \r\n # Run concurrent tests\r\n runner = ConcurrentTestRunner()\r\n failures = runner.run_tests(['myapp.tests'])\r\n \r\n # Verify isolation\r\n worker_connections = runner.get_worker_connections()\r\n isolation_verified = verify_database_isolation(worker_connections)\r\n \r\n assert isolation_verified, \"Database isolation verification failed\"\r\n```\r\n\r\n## \ud83d\udcc8 Performance Monitoring\r\n\r\n### Connection Pool Statistics\r\n\r\n```python\r\nfrom django_concurrent_test.db import get_connection_pool_stats\r\n\r\n# Get connection pool metrics\r\nstats = get_connection_pool_stats()\r\nprint(f\"Active connections: {stats.get('active', 0)}\")\r\nprint(f\"Pooled connections: {stats.get('pooled', 0)}\")\r\nprint(f\"Connection hits: {stats.get('hits', 0)}\")\r\nprint(f\"Connection misses: {stats.get('misses', 0)}\")\r\n```\r\n\r\n### Memory-Based Scaling\r\n\r\nThe package automatically scales worker count based on available memory:\r\n\r\n```python\r\nfrom django_concurrent_test.runner import ConcurrentTestRunner\r\n\r\n# Memory-based scaling is automatic\r\nrunner = ConcurrentTestRunner()\r\n# Worker count will be calculated based on available memory\r\n```\r\n\r\n## \ud83d\udea8 Error Handling\r\n\r\n### Timeout Management\r\n\r\n```python\r\nfrom django_concurrent_test.exceptions import TestTimeoutException\r\n\r\ndef test_with_timeout():\r\n \"\"\"Test with custom timeout handling.\"\"\"\r\n try:\r\n # Run test with timeout\r\n result = run_test_with_timeout(test_function, timeout=30)\r\n assert result is not None\r\n except TestTimeoutException:\r\n pytest.fail(\"Test timed out\")\r\n```\r\n\r\n### Database Error Recovery\r\n\r\n```python\r\nfrom django_concurrent_test.db import wait_for_database_ready\r\n\r\ndef test_database_recovery():\r\n \"\"\"Test database connection recovery.\"\"\"\r\n \r\n # Wait for database to be ready\r\n ready = wait_for_database_ready('test_db', timeout=30)\r\n assert ready, \"Database not ready within timeout\"\r\n```\r\n\r\n## \ud83d\udd27 Development\r\n\r\n### Running Tests\r\n\r\n```bash\r\n# Run package tests\r\npytest tests/\r\n\r\n# Run with coverage\r\npytest --cov=django_concurrent_test tests/\r\n\r\n# Run with concurrent execution\r\npytest --concurrent tests/\r\n```\r\n\r\n### Building Documentation\r\n\r\n```bash\r\n# Build package\r\npython -m build\r\n\r\n# Upload to PyPI\r\ntwine upload dist/*\r\n```\r\n\r\n## \ud83d\udccb Requirements\r\n\r\n- Python 3.9+\r\n- Django 3.2+\r\n- PostgreSQL 10+ or MySQL 5.7+ (for production features)\r\n- SQLite (for development)\r\n\r\n## \ud83e\udd1d Contributing\r\n\r\n1. Fork the repository\r\n2. Create a feature branch\r\n3. Make your changes\r\n4. Add tests for new functionality\r\n5. Ensure all tests pass\r\n6. Submit a pull request\r\n\r\n## \ud83d\udcde Contact\r\n\r\n- **Email**: ranaehtashamali1@gmail.com\r\n- **Phone**: +923224712517\r\n- **GitHub**: [@RanaEhtashamAli](https://github.com/RanaEhtashamAli)\r\n\r\n## \ud83d\udcc4 License\r\n\r\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\r\n\r\n## \ud83d\ude4f Acknowledgments\r\n\r\n- Django team for the excellent testing framework\r\n- PostgreSQL and MySQL communities for database support\r\n- All contributors who have helped improve this package\r\n\r\n## \ud83d\udcde Support\r\n\r\n- **Issues**: [GitHub Issues](https://github.com/RanaEhtashamAli/django-concurrent-test/issues)\r\n- **Documentation**: [Read the Docs](https://django-concurrent-test.readthedocs.io/)\r\n- **Discussions**: [GitHub Discussions](https://github.com/RanaEhtashamAli/django-concurrent-test/discussions)\r\n- **Email**: ranaehtashamali1@gmail.com\r\n- **Phone**: +923224712517\r\n\r\n---\r\n\r\n**Made with \u2764\ufe0f for the Django community** \r\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Production-ready Django package for safe and configurable concurrent testing with isolated databases, timing analytics, and concurrency simulation middleware",
"version": "1.0.0",
"project_urls": {
"Bug Tracker": "https://github.com/RanaEhtashamAli/django-concurrent-test/issues",
"Changelog": "https://github.com/RanaEhtashamAli/django-concurrent-test/blob/main/CHANGELOG.md",
"Documentation": "https://django-concurrent-test.readthedocs.io",
"Homepage": "https://github.com/RanaEhtashamAli/django-concurrent-test",
"Repository": "https://github.com/RanaEhtashamAli/django-concurrent-test"
},
"split_keywords": [
"django",
" testing",
" concurrent",
" parallel",
" database",
" postgresql",
" mysql",
" middleware",
" analytics",
" benchmarking"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "0076fbc0cdb7c71dd52334308a9b2af5682f9440ae61c06f95e41d31259ec95f",
"md5": "ab164c997e6c0c60f1b7b7d1cd677de1",
"sha256": "6cd969e2164b92b25c4aec6c2f4f14b0210b923de54484e0c0584748a52caa49"
},
"downloads": -1,
"filename": "django_concurrent_test-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "ab164c997e6c0c60f1b7b7d1cd677de1",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 40676,
"upload_time": "2025-07-14T00:25:22",
"upload_time_iso_8601": "2025-07-14T00:25:22.722198Z",
"url": "https://files.pythonhosted.org/packages/00/76/fbc0cdb7c71dd52334308a9b2af5682f9440ae61c06f95e41d31259ec95f/django_concurrent_test-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "072b2183bf69dd7b72319980cea9cbff412622039458061d1fdb93fe5917142e",
"md5": "64f3018c73fa4a9b616e739177b98387",
"sha256": "c60d472d98cc6fa00c857ebc19925bb9f41a467bdddd1aad300e762284fc0ee4"
},
"downloads": -1,
"filename": "django_concurrent_test-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "64f3018c73fa4a9b616e739177b98387",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 74474,
"upload_time": "2025-07-14T00:25:24",
"upload_time_iso_8601": "2025-07-14T00:25:24.559709Z",
"url": "https://files.pythonhosted.org/packages/07/2b/2183bf69dd7b72319980cea9cbff412622039458061d1fdb93fe5917142e/django_concurrent_test-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-14 00:25:24",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "example",
"github_project": "django-concurrent-test",
"github_not_found": true,
"lcname": "django-concurrent-test"
}
JSON
