Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions cloud_storage/cloud_storage/overrides/file.py
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,21 @@ def add_file_version(self, version_id):
"timestamp": get_datetime(),
},
)
if not self.is_new():
# File already exists in DB (filename-conflict path in write_file).
# Frappe's save lifecycle won't persist this file's child tables,
# so we insert the version record directly.
frappe.get_doc(
{
"doctype": "File Version",
"parent": self.name,
"parenttype": "File",
"parentfield": "versions",
"version": str(version_id),
"user": frappe.session.user,
"timestamp": get_datetime(),
}
).insert(ignore_permissions=True)

def remove_file_association(self, dt: str, dn: str) -> None:
if len(self.file_association) <= 1:
Expand Down Expand Up @@ -525,17 +540,19 @@ def upload_file(file: File) -> File:
path = get_file_path(file, client.folder)
file.db_set("file_url", FILE_URL.format(path=path))
content_type = file.content_type or from_buffer(file.content, mime=True)
version_id = None
try:
response = client.put_object(
Body=file.content, Bucket=client.bucket, Key=path, ContentType=content_type
)
if response.get("VersionId"):
file.add_file_version(response.get("VersionId"))
version_id = response.get("VersionId") or file.content_hash
file.associate_files(file.attached_to_doctype, file.attached_to_name)
except S3UploadFailedError:
frappe.throw(_("File Upload Failed. Please try again."))
except Exception as e:
frappe.log_error("File Upload Error", e)
if version_id:
file.add_file_version(version_id)
file.db_set("s3_key", path)
return file

Expand Down
4 changes: 4 additions & 0 deletions cloud_storage/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,4 +87,8 @@ def mocked_s3_client():
)
bucket = "test_bucket"
client.create_bucket(Bucket=bucket)
client.put_bucket_versioning(
Bucket=bucket,
VersioningConfiguration={"Status": "Enabled"},
)
yield _MockedS3Client(client, bucket, "test_folder", 110)
43 changes: 23 additions & 20 deletions cloud_storage/tests/test_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,26 +188,29 @@ def test_save_file_without_S3_and_preview(example_file_record_4, example_file_re
frappe.conf.cloud_storage_settings = old_settings


@mock_s3
def test_file_versioning_with_content_change(example_file_record_5, tmp_path):
frappe.set_user("Administrator")
file1 = create_upload_file(example_file_record_5, file_name="sample.csv")
assert frappe.db.exists("File", file1.name)

modified_csv = tmp_path / "sample.csv"
with open(example_file_record_5) as src, open(modified_csv, "w") as dst:
lines = src.readlines()
dst.writelines(lines)
dst.write("4,5,6\n")

file2 = create_upload_file(modified_csv, file_name="sample.csv")
file2.load_from_db()

assert len(file1.versions) >= 2
# Optionally, check that the latest version is the most recent
latest_version = file1.versions[-1]
assert latest_version.user == "Administrator"
assert latest_version.version is not None
def test_file_versioning_with_content_change(mocked_s3_client, example_file_record_5, tmp_path):
with patch(
"cloud_storage.cloud_storage.overrides.file.get_cloud_storage_client",
return_value=mocked_s3_client,
):
frappe.set_user("Administrator")
file1 = create_upload_file(example_file_record_5, file_name="sample.csv")
assert frappe.db.exists("File", file1.name)

modified_csv = tmp_path / "sample.csv"
with open(example_file_record_5) as src, open(modified_csv, "w") as dst:
lines = src.readlines()
dst.writelines(lines)
dst.write("4,5,6\n")

file2 = create_upload_file(modified_csv, file_name="sample.csv")
file1.load_from_db()

assert len(file1.versions) >= 2
# Optionally, check that the latest version is the most recent
latest_version = file1.versions[-1]
assert latest_version.user == "Administrator"
assert latest_version.version is not None


def test_migration_command(mocked_s3_client, example_file_record_6):
Expand Down
Loading