169 lines
9.8 KiB
Python
169 lines
9.8 KiB
Python
"""Deletion contract: admin may remove shared spaces, never private spaces.
|
|
|
|
Run: USE_SQLITE=true python manage.py test drive.test_protected_deletions
|
|
"""
|
|
import tempfile
|
|
import uuid
|
|
from pathlib import Path
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIClient
|
|
from .models import DriveItem, DriveSpace, SpaceGrant, ArchivedDriveAccount
|
|
from .spaces import ensure_space, ensure_private_space, create_space
|
|
|
|
|
|
class ProtectedDeletionTests(TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.storage = Path(tmp.name)
|
|
override = override_settings(DRIVE_STORAGE_ROOT=self.storage,
|
|
FILE_UPLOAD_TEMP_DIR=str(self.storage / '.uploads'))
|
|
override.enable()
|
|
self.addCleanup(override.disable)
|
|
User = get_user_model()
|
|
self.admin = User.objects.create_superuser('test-admin', 'admin@example.org', 'VeryStrongPass123!')
|
|
self.alice = User.objects.create_user('test-alice', password='VeryStrongPass123!')
|
|
self.root = ensure_space(self.admin)
|
|
self.admin_private = ensure_private_space(self.admin)
|
|
self.assigned = create_space(self.alice, self.root, 'Alice partagé')
|
|
self.private = ensure_private_space(self.alice)
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(user=self.admin)
|
|
|
|
def private_file(self):
|
|
item = DriveItem.objects.create(owner=self.alice, space=self.private,
|
|
name='mon-secret.txt', storage_key=str(uuid.uuid4()),
|
|
storage_location=f'{self.private.relative_path}/mon-secret.txt', size=6)
|
|
path = self.storage / item.storage_location
|
|
path.write_bytes(b'secret')
|
|
return item, path
|
|
|
|
def test_admin_cannot_browse_or_delete_any_other_private_space(self):
|
|
item, path = self.private_file()
|
|
result = self.client.get('/api/drive/spaces/')
|
|
self.assertEqual(result.status_code, 200, result.data)
|
|
self.assertNotIn(self.private.pk, [s['id'] for s in result.data['spaces']])
|
|
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code, 403)
|
|
self.assertEqual(self.client.get('/api/drive/stats/', {'space':self.private.pk}).status_code, 403)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{item.pk}/link/').status_code, 403)
|
|
self.assertEqual(self.client.delete(f'/api/drive/spaces/{self.private.pk}/',
|
|
{'confirm_name':self.private.display_name}, format='json').status_code, 403)
|
|
self.client.force_authenticate(user=self.alice)
|
|
self.assertEqual(self.client.delete(f'/api/drive/spaces/{self.private.pk}/',
|
|
{'confirm_name':self.private.display_name}, format='json').status_code, 403)
|
|
self.assertTrue(path.is_file())
|
|
|
|
def test_admin_can_delete_assigned_shared_space_without_private_loss(self):
|
|
secret, path = self.private_file()
|
|
shared_path = self.storage / self.assigned.relative_path
|
|
self.assertTrue(shared_path.is_dir())
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
res = self.client.delete(f'/api/drive/spaces/{self.assigned.pk}/',
|
|
{'confirm_name':self.assigned.display_name}, format='json')
|
|
self.assertEqual(res.status_code, 200, res.data)
|
|
self.assertFalse(DriveSpace.objects.filter(pk=self.assigned.pk).exists())
|
|
self.assertFalse(shared_path.exists())
|
|
self.private.refresh_from_db()
|
|
self.assertEqual(self.private.parent_id, self.root.pk)
|
|
self.assertTrue(path.exists())
|
|
self.assertTrue(DriveItem.objects.filter(pk=secret.pk).exists())
|
|
self.client.force_authenticate(user=self.alice)
|
|
self.assertIsNone(ensure_space(self.alice))
|
|
self.assertIsNone(self.client.get('/api/auth/me/').data['space_id'])
|
|
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code, 200)
|
|
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.root.pk}).status_code, 403)
|
|
|
|
def test_admin_root_reset_keeps_root_and_private_files(self):
|
|
item, private_path = self.private_file()
|
|
root_folder = DriveItem.objects.create(owner=self.admin, space=self.root,
|
|
name='public-folder', is_folder=True)
|
|
public_path = self.storage / 'public-folder'
|
|
public_path.mkdir()
|
|
shared_path = self.storage / self.assigned.relative_path
|
|
self.assertTrue(shared_path.is_dir())
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
res = self.client.delete(f'/api/drive/spaces/{self.root.pk}/',
|
|
{'confirm_name':self.root.display_name}, format='json')
|
|
self.assertEqual(res.status_code, 200, res.data)
|
|
self.assertTrue(res.data['root_reset'])
|
|
self.assertTrue(DriveSpace.objects.filter(pk=self.root.pk).exists())
|
|
self.assertFalse(DriveSpace.objects.filter(pk=self.assigned.pk).exists())
|
|
self.assertFalse(DriveItem.objects.filter(pk=root_folder.pk).exists())
|
|
self.assertFalse(shared_path.exists())
|
|
self.assertFalse(public_path.exists())
|
|
self.assertTrue(private_path.exists())
|
|
self.assertTrue(DriveItem.objects.filter(pk=item.pk).exists())
|
|
self.assertTrue((self.storage/'privates').is_dir())
|
|
|
|
def test_user_deletion_removes_private_directory_and_keeps_shared_files(self):
|
|
item, path = self.private_file()
|
|
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.admin.pk}/',
|
|
{'confirm_username':self.admin.username}, format='json').status_code, 403)
|
|
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
|
{'confirm_username':'wrong'}, format='json').status_code, 400)
|
|
shared = DriveItem.objects.create(owner=self.alice, space=self.root,
|
|
name='communal.txt', storage_key=str(uuid.uuid4()),
|
|
storage_location='communal.txt', size=4)
|
|
shared_path = self.storage / 'communal.txt'
|
|
shared_path.write_bytes(b'team')
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
|
{'confirm_username':self.alice.username}, format='json').status_code, 204)
|
|
self.alice.refresh_from_db()
|
|
self.assertFalse(self.alice.is_active)
|
|
self.assertFalse(self.alice.has_usable_password())
|
|
self.assertTrue(ArchivedDriveAccount.objects.filter(user=self.alice).exists())
|
|
self.assertNotIn(self.alice.pk, [u['id'] for u in self.client.get('/api/drive/admin/users/').data['users']])
|
|
self.assertFalse(path.exists())
|
|
self.assertFalse((self.storage / self.private.relative_path).exists())
|
|
self.assertFalse(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
|
self.assertFalse(DriveItem.objects.filter(pk=item.pk).exists())
|
|
self.assertTrue(shared_path.exists())
|
|
self.assertTrue(DriveItem.objects.filter(pk=shared.pk).exists())
|
|
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
|
{'confirm_username':self.alice.username}, format='json').status_code, 404)
|
|
|
|
def test_non_admin_cannot_delete_other_users(self):
|
|
self.client.force_authenticate(user=self.alice)
|
|
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.admin.pk}/',
|
|
{'confirm_username':self.admin.username}, format='json').status_code, 403)
|
|
|
|
def test_admin_can_revoke_space_grants_without_removing_private_storage(self):
|
|
from .models import SpaceGrant
|
|
# Access granted explicitly by invitation/existing-account grant.
|
|
grant = SpaceGrant.objects.create(user=self.alice, space=self.root,
|
|
can_write=True, can_invite=True)
|
|
rsp = self.client.get(f'/api/drive/admin/users/{self.alice.pk}/access/')
|
|
self.assertEqual(rsp.status_code, 200, rsp.data)
|
|
self.assertTrue(any(a['kind'] == 'grant' and a['space_id'] == self.root.pk
|
|
for a in rsp.data['accesses']))
|
|
self.assertNotIn(self.private.pk, [a['space_id'] for a in rsp.data['accesses']])
|
|
rsp = self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/access/grant/{self.root.pk}/')
|
|
self.assertEqual(rsp.status_code, 204)
|
|
self.assertFalse(SpaceGrant.objects.filter(pk=grant.pk).exists())
|
|
self.assertTrue(DriveSpace.objects.filter(pk=self.root.pk).exists())
|
|
self.assertTrue(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
|
|
|
# Older accounts are linked directly to their assigned shared space.
|
|
rsp = self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/access/owner/{self.assigned.pk}/')
|
|
self.assertEqual(rsp.status_code, 204)
|
|
self.assigned.refresh_from_db()
|
|
self.assertIsNone(self.assigned.user_id)
|
|
self.assertIsNone(ensure_space(self.alice))
|
|
self.client.force_authenticate(user=self.alice)
|
|
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.root.pk}).status_code,403)
|
|
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code,200)
|
|
self.assertEqual(self.client.get(f'/api/drive/admin/users/{self.alice.pk}/access/').status_code,403)
|
|
|
|
def test_account_delete_rejects_private_folder_outside_privates(self):
|
|
self.private.relative_path = f'{self.assigned.relative_path}/legacy'
|
|
self.private.save(update_fields=['relative_path'])
|
|
original = self.storage / self.private.relative_path
|
|
original.mkdir(parents=True, exist_ok=True)
|
|
rsp=self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
|
{'confirm_username':self.alice.username},format='json')
|
|
self.assertEqual(rsp.status_code,409)
|
|
self.assertTrue(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
|
self.assertTrue(original.exists())
|