我在 django 中创建了一个模型来上传文件,它应该将其保存在媒体存储中以供 celery 脚本使用。 它应该创建一个带有id的目录,并且每次我上传相同的文件时,都应该先删除它(我不知道为什么我不能覆盖它,但如果它先删除它就足够了)。 当实体被删除时,目录也应该被删除。
我正在使用 python3.12、django 4.2 和 django Rest Framework3.14。
代码有效:
def upload_to(instance, filename):
instance_id = instance.id or uuid.uuid4().hex
directory = f'sample_{instance_id}'
file_path = os.path.join(directory, filename)
full_path = os.path.join(settings.MEDIA_ROOT, file_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
return file_path
class Sample(models.Model):
name = models.CharField(max_length=30)
test_suite = models.FileField(upload_to=upload_to)
def delete(self, *args, **kwargs):
if self.test_suite:
if os.path.isfile(self.test_suite.path):
os.remove(self.test_suite.path)
super().delete(*args, **kwargs)
def save(self, *args, **kwargs):
if self.pk is None:
saved_test_suite = self.test_suite
self.test_suite = None
super().save(*args, **kwargs)
self.test_suite = saved_test_suite
if 'force_insert' in kwargs:
kwargs.pop('force_insert')
elif self.pk:
try:
old_instance = Sample.objects.get(pk=self.pk)
if (
old_instance.test_suite
and self.test_suite
and old_instance.test_suite != self.test_suite
):
if os.path.isfile(old_instance.test_suite.path):
os.remove(old_instance.test_suite.path)
except Sample.DoesNotExist:
pass
if 'uuid' in self.test_suite.name:
new_path = self.test_suite.name.replace(
self.test_suite.name.split('/')[0],
f'dir_{self.pk}'
)
old_path = self.test_suite.path
new_full_path = os.path.join(settings.MEDIA_ROOT, new_path)
os.renames(old_path, new_full_path)
self.test_suite.name = new_path
super().save(update_fields=['test_suite'])
super().save(*args, **kwargs)
这些是测试:
@pytest.fixture
def create_test_file():
def _create_test_file(filename: str) -> SimpleUploadedFile:
return SimpleUploadedFile(filename, b'test file content')
return _create_test_file
@pytest.mark.django_db
def test_sample_creation(tmp_path, create_test_file):
settings.MEDIA_ROOT = tmp_path
sample = Sample.objects.create(
name='test_sample_name',
test_suite=create_test_file(filename='test_sample.sample'),
)
expected_directory = tmp_path / f'sample_{sample.pk}'
expected_file_path = expected_directory / 'test_sample.sample'
assert expected_directory.is_dir()
assert expected_file_path.is_file()
assert sample.name == 'test_sample_name'
assert sample.test_suite.name.endswith('test_sample.sample')
assert f'sample_{sample.pk}' in sample.test_suite.name
@pytest.mark.django_db
def test_sample_update(tmp_path, create_test_file):
settings.MEDIA_ROOT = tmp_path
sample = Sample.objects.create(
name='test_sample_name',
test_suite=create_test_file('test_sample.sample'),
)
new_file = create_test_file(filename='updated_test_sample.sample')
sample.test_suite = new_file
sample.save()
sample.refresh_from_db()
expected_directory = tmp_path / f'sample_{sample.pk}'
assert expected_directory.is_dir()
expected_file_path = expected_directory / 'updated_test_sample.sample'
# import pdb; pdb.set_trace()
assert expected_file_path.is_file()
assert sample.test_suite.name == f'sample_{sample.pk}/updated_test_sample.sample'
old_file_path = expected_directory / 'test_sample.sample'
assert not old_file_path.exists()
错误是:
> assert expected_file_path.is_file()
E AssertionError: assert False
E + where False = <bound method Path.is_file of PosixPath('/tmp/pytest-of-usuername/pytest-84/test_sample_update0/sample_1/updated_test_sample.sample')>()
E + where <bound method Path.is_file of PosixPath('/tmp/pytest-of-usuername/pytest-84/test_sample_update0/sample_1/updated_test_sample.sample')> = PosixPath('/tmp/pytest-of-usuername/pytest-84/test_sample_update0/sample_1/updated_test_sample.sample').is_file
最奇怪的部分是,如果我单独执行“test_sample_update”,则有效,如果我用“test_sample_update”更改“test_sample_creation”,则失败的是“test_sample_creation”。所以它应该是测试之间保存的某种数据,但我找不到它。
如果我调试注释了“set_trace”的位置:
(Pdb++) expected_file_path
PosixPath('/tmp/pytest-of-username/pytest-75/test_sample_update0/sample_1/updated_test_sample.sample')
(Pdb++) expected_file_path.is_file()
False
单独执行“更新”时
(Pdb++) expected_file_path
PosixPath('/tmp/pytest-of-username/pytest-77/test_sample_update0/sample_1/updated_test_sample.sample')
(Pdb++) expected_file_path.is_file()
True
旧文件正在从与运行测试的目录不同的目录中删除。 要修复它:
old_file_path = Path(sample.test_suite.path)
if old_file_path.exists():
old_file_path.unlink()
new_file_path = Path(sample.test_suite.path)
这应该可以解决问题,因为:
它使用 Django 模型中的实际路径而不是构建它们 它使用正确的路径正确清理旧文件。