add userfile module
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
j3d1 2023-07-01 20:45:38 +02:00
parent 68a9d520f2
commit 46a6b4e3d4
6 changed files with 187 additions and 0 deletions

View file

@ -48,6 +48,7 @@ INSTALLED_APPS = [
'drf_yasg',
'authentication',
'hostadmin',
'files',
'toolshed',
]

View file

View file

@ -0,0 +1,26 @@
# Generated by Django 4.2.2 on 2023-07-01 18:22
from django.db import migrations, models
import files.models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='File',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('file', models.FileField(unique=True, upload_to=files.models.hash_upload)),
('mime_type', models.CharField(max_length=255)),
('hash', models.CharField(max_length=64, unique=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
),
]

View file

55
backend/files/models.py Normal file
View file

@ -0,0 +1,55 @@
from django.core.files.base import ContentFile
from django.db import models, IntegrityError
from django.db.models import Model
from authentication.models import ToolshedUser
def hash_upload(instance, filename):
return f"{instance.hash[:2]}/{instance.hash[2:4]}/{instance.hash[4:6]}/{instance.hash[6:]}"
class FileManager(models.Manager):
def get_or_create(self, **kwargs):
if 'data' in kwargs and type(kwargs['data']) == str:
import base64
from hashlib import sha256
content = base64.b64decode(kwargs['data'], validate=True)
kwargs.pop('data')
content_hash = sha256(content).hexdigest()
kwargs['file'] = ContentFile(content, content_hash)
kwargs['hash'] = content_hash
else:
raise ValueError('data must be a base64 encoded string or file and hash must be provided')
try:
return self.get(hash=kwargs['hash']), False
except self.model.DoesNotExist:
return self.create(**kwargs), True
def create(self, **kwargs):
if 'data' in kwargs and type(kwargs['data']) == str:
import base64
from hashlib import sha256
content = base64.b64decode(kwargs['data'], validate=True)
kwargs.pop('data')
content_hash = sha256(content).hexdigest()
kwargs['file'] = ContentFile(content, content_hash)
kwargs['hash'] = content_hash
elif 'file' in kwargs and 'hash' in kwargs and type(kwargs['file']) == ContentFile:
pass
else:
raise ValueError('data must be a base64 encoded string or file and hash must be provided')
if not self.filter(hash=kwargs['hash']).exists():
return super().create(**kwargs)
else:
raise IntegrityError('File with this hash already exists')
class File(Model):
file = models.FileField(upload_to=hash_upload, null=False, blank=False, unique=True)
mime_type = models.CharField(max_length=255, null=False, blank=False)
hash = models.CharField(max_length=64, null=False, blank=False, unique=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
objects = FileManager()

105
backend/files/tests.py Normal file
View file

@ -0,0 +1,105 @@
from django.core.files.base import ContentFile
from django.core.files.storage import DefaultStorage
from django.db import IntegrityError, transaction
from django.test import Client
from authentication.tests import SignatureAuthClient, ToolshedTestCase
from nacl.hash import sha256
from nacl.encoding import HexEncoder
import base64
from files.models import File
anonymous_client = Client()
client = SignatureAuthClient()
def rmdir(storage, path):
dirs, files = storage.listdir(path)
for file in files:
storage.delete(path + file)
for dir in dirs:
rmdir(storage, path + dir + "/")
storage.delete(path)
def countdir(storage, path):
dirs, files = storage.listdir(path)
count = len(files)
for dir in dirs:
count += countdir(storage, path + dir + "/")
return count
class FilesTestMixin:
def prepare_files(self):
rmdir(DefaultStorage(), '')
self.f['test_content1'] = b'testcontent1'
self.f['hash1'] = sha256(self.f['test_content1'], encoder=HexEncoder).decode('utf-8')
self.f['encoded_content1'] = base64.b64encode(self.f['test_content1']).decode('utf-8')
self.f['test_file1'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content1'])
self.f['test_content2'] = b'testcontent2'
self.f['hash2'] = sha256(self.f['test_content2'], encoder=HexEncoder).decode('utf-8')
self.f['encoded_content2'] = base64.b64encode(self.f['test_content2']).decode('utf-8')
self.f['test_file2'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content2'])
self.f['test_content3'] = b'testcontent3'
self.f['hash3'] = sha256(self.f['test_content3'], encoder=HexEncoder).decode('utf-8')
self.f['encoded_content3'] = base64.b64encode(self.f['test_content3']).decode('utf-8')
self.f['test_file3'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content3'])
self.f['test_content4'] = b'testcontent4'
self.f['hash4'] = sha256(self.f['test_content4'], encoder=HexEncoder).decode('utf-8')
self.f['encoded_content4'] = base64.b64encode(self.f['test_content4']).decode('utf-8')
class FilesTestCase(FilesTestMixin, ToolshedTestCase):
def setUp(self):
super().setUp()
self.prepare_files()
def test_file_list(self):
self.assertEqual(File.objects.count(), 3)
self.assertEqual(countdir(DefaultStorage(), ''), 3)
def test_file_upload(self):
File.objects.create(mime_type='text/plain', data=self.f['encoded_content4'])
self.assertEqual(File.objects.count(), 4)
self.assertEqual(countdir(DefaultStorage(), ''), 4)
self.assertEqual(File.objects.get(id=4).file.read(), self.f['test_content4'])
self.assertEqual(File.objects.get(id=4).file.name,
f"{self.f['hash4'][:2]}/{self.f['hash4'][2:4]}/{self.f['hash4'][4:6]}/{self.f['hash4'][6:]}")
def test_file_upload_fail(self):
with transaction.atomic():
with self.assertRaises(ValueError):
File.objects.create(file=ContentFile(self.f['test_content4']), mime_type='text/plain')
self.assertEqual(File.objects.count(), 3)
self.assertEqual(countdir(DefaultStorage(), ''), 3)
def test_file_upload_duplicate(self):
with transaction.atomic():
with self.assertRaises(IntegrityError):
File.objects.create(mime_type='text/plain', data=self.f['encoded_content3'])
self.assertEqual(File.objects.count(), 3)
self.assertEqual(countdir(DefaultStorage(), ''), 3)
def test_file_upload_get_or_create(self):
file, created = File.objects.get_or_create(data=self.f['encoded_content3'])
self.assertEqual(File.objects.count(), 3)
self.assertEqual(countdir(DefaultStorage(), ''), 3)
self.assertFalse(created)
self.assertEqual(file.file.read(), self.f['test_content3'])
self.assertEqual(file.file.name,
f"{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}")
file, created = File.objects.get_or_create(data=self.f['encoded_content4'])
self.assertEqual(File.objects.count(), 4)
self.assertEqual(countdir(DefaultStorage(), ''), 4)
self.assertTrue(created)
self.assertEqual(file.file.read(), self.f['test_content4'])
self.assertEqual(file.file.name,
f"{self.f['hash4'][:2]}/{self.f['hash4'][2:4]}/{self.f['hash4'][4:6]}/{self.f['hash4'][6:]}")
def test_file_upload_get_or_create_fail(self):
with transaction.atomic():
with self.assertRaises(ValueError):
File.objects.get_or_create(hash=self.f['hash3'])
self.assertEqual(File.objects.count(), 3)
self.assertEqual(countdir(DefaultStorage(), ''), 3)