jcv-dev
551b42eab5
Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
101 lines
4 KiB
Python
101 lines
4 KiB
Python
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
from infrasynth.features.models import FeatureFlag
|
|
from infrasynth.files.models import FileCategory, ProcessingPipeline, StoredFile
|
|
from infrasynth.files.services import FileService
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
FILES_URL = "/api/v1/files/files/"
|
|
CATEGORIES_URL = "/api/v1/files/categories/"
|
|
PIPELINES_URL = "/api/v1/files/pipelines/"
|
|
|
|
|
|
@pytest.fixture
|
|
def stored_file(user, media_root):
|
|
return FileService().upload(
|
|
SimpleUploadedFile("doc.pdf", b"%PDF-1.4", content_type="application/pdf"),
|
|
filename="doc.pdf",
|
|
user=user,
|
|
)
|
|
|
|
|
|
class TestStoredFileViewSet:
|
|
def test_list_files(self, authenticated_client, stored_file):
|
|
response = authenticated_client.get(FILES_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["original_filename"] == "doc.pdf"
|
|
|
|
def test_list_requires_auth(self, api_client, stored_file):
|
|
assert api_client.get(FILES_URL).status_code == 401
|
|
|
|
def test_retrieve_file(self, authenticated_client, stored_file):
|
|
response = authenticated_client.get(f"{FILES_URL}{stored_file.id}/")
|
|
assert response.status_code == 200
|
|
assert response.data["id"] == stored_file.id
|
|
assert response.data["mime_type"] == "application/pdf"
|
|
|
|
def test_download_action(self, authenticated_client, stored_file):
|
|
response = authenticated_client.get(f"{FILES_URL}{stored_file.id}/download/")
|
|
assert response.status_code == 200
|
|
assert "attachment" in response["Content-Disposition"]
|
|
assert b"".join(response.streaming_content) == b"%PDF-1.4"
|
|
|
|
def test_download_requires_auth(self, api_client, stored_file):
|
|
assert api_client.get(f"{FILES_URL}{stored_file.id}/download/").status_code == 401
|
|
|
|
def test_delete_file(self, authenticated_client, stored_file):
|
|
response = authenticated_client.delete(f"{FILES_URL}{stored_file.id}/")
|
|
assert response.status_code == 204
|
|
assert not StoredFile.objects.filter(pk=stored_file.pk).exists()
|
|
|
|
def test_disabled_feature_returns_404(self, authenticated_client):
|
|
FeatureFlag.objects.create(slug="files", name="Files", is_active=False)
|
|
assert authenticated_client.get(FILES_URL).status_code == 404
|
|
|
|
|
|
class TestFileCategoryViewSet:
|
|
def test_list_categories(self, authenticated_client):
|
|
FileCategory.objects.create(slug="docs", name="Docs", storage_path="docs")
|
|
response = authenticated_client.get(f"{CATEGORIES_URL}")
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["slug"] == "docs"
|
|
|
|
def test_create_category(self, authenticated_client):
|
|
response = authenticated_client.post(
|
|
f"{CATEGORIES_URL}",
|
|
{"slug": "img", "name": "Images", "storage_path": "img"},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 201
|
|
assert FileCategory.objects.count() == 1
|
|
|
|
def test_requires_auth(self, api_client):
|
|
assert api_client.get(f"{CATEGORIES_URL}").status_code == 401
|
|
|
|
|
|
class TestProcessingPipelineViewSet:
|
|
def test_list_pipelines(self, authenticated_client):
|
|
ProcessingPipeline.objects.create(name="Resize", slug="resize", steps=[])
|
|
response = authenticated_client.get(f"{PIPELINES_URL}")
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
|
|
def test_create_pipeline(self, authenticated_client):
|
|
response = authenticated_client.post(
|
|
f"{PIPELINES_URL}",
|
|
{
|
|
"name": "Optimize",
|
|
"slug": "optimize",
|
|
"steps": [{"type": "optimize"}],
|
|
},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 201
|
|
assert ProcessingPipeline.objects.filter(slug="optimize").exists()
|
|
|
|
def test_requires_auth(self, api_client):
|
|
assert api_client.get(f"{PIPELINES_URL}").status_code == 401
|