arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-10-31 05:43:41 +00:00
Code Issues Projects Releases Wiki Activity

guess mime type for local files

Browse Source
This commit is contained in:
Evgenii Alekseev
2021-08-18 05:04:26 +03:00
parent 8f55cc600e
commit 343768e015
2 changed files with 24 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/ahriman/core/upload/s3.py
View File

@ -19,6 +19,7 @@
# #
import boto3 # type: ignore import boto3 # type: ignore
import hashlib import hashlib
import mimetypes
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Generator, Iterable from typing import Any, Dict, Generator, Iterable
@ -122,8 +123,13 @@ class S3(Upload):
remote_checksum = remote_object.e_tag[1:-1] if remote_object is not None else None remote_checksum = remote_object.e_tag[1:-1] if remote_object is not None else None
if remote_checksum == checksum: if remote_checksum == checksum:
continue continue
local_path = path / local_file
remote_path = Path(self.architecture) / local_file remote_path = Path(self.architecture) / local_file
self.bucket.upload_file(str(path / local_file), str(remote_path)) (mime, _) = mimetypes.guess_type(local_path)
extra_args = {"Content-Type": mime} if mime is not None else None
self.bucket.upload_file(Filename=str(local_path), Key=str(remote_path), ExtraArgs=extra_args)
# remove files which were removed locally # remove files which were removed locally
for local_file, remote_object in remote_objects.items(): for local_file, remote_object in remote_objects.items():

24
tests/ahriman/core/upload/test_s3.py
View File

@ -1,6 +1,6 @@
from pathlib import Path from pathlib import Path
from pytest_mock import MockerFixture from pytest_mock import MockerFixture
from typing import Any, List from typing import Any, List, Optional, Tuple
from unittest import mock from unittest import mock
from unittest.mock import MagicMock from unittest.mock import MagicMock
@ -74,15 +74,17 @@ def test_sync(s3: S3, s3_remote_objects: List[Any], mocker: MockerFixture) -> No
""" """
must run sync command must run sync command
""" """
def mimetype(path: Path) -> Tuple[Optional[str], None]:
return ("text/html", None) if path.name == "b" else (None, None)
root = Path("path") root = Path("path")
local_files = { local_files = {
Path(item.key.replace("a", "d")): item.e_tag.replace("b", "d").replace("\"", "") Path(item.key.replace("a", "d")): item.e_tag.replace("b", "d").replace("\"", "")
for item in s3_remote_objects for item in s3_remote_objects
} }
remote_objects = {Path(item.key): item for item in s3_remote_objects} remote_objects = {Path(item.key): item for item in s3_remote_objects}
print(local_files)
print(remote_objects)
mocker.patch("mimetypes.guess_type", side_effect=mimetype)
local_files_mock = mocker.patch("ahriman.core.upload.s3.S3.get_local_files", return_value=local_files) local_files_mock = mocker.patch("ahriman.core.upload.s3.S3.get_local_files", return_value=local_files)
remote_objects_mock = mocker.patch("ahriman.core.upload.s3.S3.get_remote_objects", return_value=remote_objects) remote_objects_mock = mocker.patch("ahriman.core.upload.s3.S3.get_remote_objects", return_value=remote_objects)
upload_mock = s3.bucket = MagicMock() upload_mock = s3.bucket = MagicMock()
@ -91,8 +93,16 @@ def test_sync(s3: S3, s3_remote_objects: List[Any], mocker: MockerFixture) -> No
local_files_mock.assert_called_once() local_files_mock.assert_called_once()
remote_objects_mock.assert_called_once() remote_objects_mock.assert_called_once()
upload_mock.upload_file.assert_has_calls([ upload_mock.upload_file.assert_has_calls(
mock.call(str(root / s3.architecture / "b"), f"{s3.architecture}/{s3.architecture}/b"), [
mock.call(str(root / s3.architecture / "d"), f"{s3.architecture}/{s3.architecture}/d"), mock.call(
], any_order=True) Filename=str(root / s3.architecture / "b"),
Key=f"{s3.architecture}/{s3.architecture}/b",
ExtraArgs={"Content-Type": "text/html"}),
mock.call(
Filename=str(root / s3.architecture / "d"),
Key=f"{s3.architecture}/{s3.architecture}/d",
ExtraArgs=None),
],
any_order=True)
remote_objects[Path("x86_64/a")].delete.assert_called_once() remote_objects[Path("x86_64/a")].delete.assert_called_once()