arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-04-24 15:27:17 +00:00
Code Issues Projects Releases Wiki Activity

github upload support (#41)

Browse Source
This commit is contained in:
Evgenii Alekseev 2021-10-14 02:30:13 +03:00 committed by GitHub
parent 72b26603bf
commit fcb167b1a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 690 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
docs/configuration.md
View File

@ -101,7 +101,20 @@ Group name must refer to architecture, e.g. it should be `html:x86_64` for x86_6
Remote synchronization settings.
* `target` - list of synchronizations to be used, space separated list of strings, required. Allowed values are `rsync`, `s3`.
* `target` - list of synchronizations to be used, space separated list of strings, required. Allowed values are `rsync`, `s3`, `github`.
### `github:*` groups
Group name must refer to architecture, e.g. it should be `github:x86_64` for x86_64 architecture. This feature requires Github key creation (see below).
* `api_key` - created Github API key. In order to create it do the following:
1. Go to [settings page](https://github.com/settings/profile).
2. Switch to [developers settings](https://github.com/settings/apps).
3. Switch to [personal access tokens](https://github.com/settings/tokens).
4. Generate new token. Required scope is `public_repo` (or `repo` for private repository support).
* `owner` - Github repository owner, string, required.
* `repository` - Github repository name, string, required. Repository must be created before any action and must have active branch (e.g. with readme).
* `username` - Github authorization user, string, optional, default is the same as `owner`.
### `rsync:*` groups

2
src/ahriman/application/application.py
View File

@ -109,7 +109,7 @@ class Application:
def add_archive(src: Path) -> None:
dst = self.repository.paths.packages / src.name
shutil.move(src, dst)
shutil.copy(src, dst)
def add_directory(path: Path) -> None:
for full_path in filter(package_like, path.iterdir()):

18
src/ahriman/core/status/web_client.py
View File

@ -111,7 +111,7 @@ class WebClient(Client):
try:
response = self.__session.post(self._login_url, json=payload)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not login as %s: %s", self.user, exception_response_text(e))
except Exception:
self.logger.exception("could not login as %s", self.user)
@ -138,7 +138,7 @@ class WebClient(Client):
try:
response = self.__session.post(self._package_url(package.base), json=payload)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not add %s: %s", package.base, exception_response_text(e))
except Exception:
self.logger.exception("could not add %s", package.base)
@ -158,7 +158,7 @@ class WebClient(Client):
(Package.from_json(package["package"]), BuildStatus.from_json(package["status"]))
for package in status_json
]
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not get %s: %s", base, exception_response_text(e))
except Exception:
self.logger.exception("could not get %s", base)
@ -175,7 +175,7 @@ class WebClient(Client):
status_json = response.json()
return InternalStatus.from_json(status_json)
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not get web service status: %s", exception_response_text(e))
except Exception:
self.logger.exception("could not get web service status")
@ -192,7 +192,7 @@ class WebClient(Client):
status_json = response.json()
return BuildStatus.from_json(status_json)
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not get service status: %s", exception_response_text(e))
except Exception:
self.logger.exception("could not get service status")
@ -205,7 +205,7 @@ class WebClient(Client):
try:
response = self.__session.post(self._reload_auth_url)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not reload auth module: %s", exception_response_text(e))
except Exception:
self.logger.exception("could not reload auth module")
@ -218,7 +218,7 @@ class WebClient(Client):
try:
response = self.__session.delete(self._package_url(base))
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not delete %s: %s", base, exception_response_text(e))
except Exception:
self.logger.exception("could not delete %s", base)
@ -234,7 +234,7 @@ class WebClient(Client):
try:
response = self.__session.post(self._package_url(base), json=payload)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not update %s: %s", base, exception_response_text(e))
except Exception:
self.logger.exception("could not update %s", base)
@ -249,7 +249,7 @@ class WebClient(Client):
try:
response = self.__session.post(self._ahriman_url, json=payload)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
except requests.HTTPError as e:
self.logger.exception("could not update service status: %s", exception_response_text(e))
except Exception:
self.logger.exception("could not update service status")

218
src/ahriman/core/upload/github.py Normal file
View File

@ -0,0 +1,218 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import hashlib
import mimetypes
import requests
from pathlib import Path
from typing import Any, Dict, Iterable, Optional
from ahriman.core.configuration import Configuration
from ahriman.core.upload.upload import Upload
from ahriman.core.util import exception_response_text, walk
from ahriman.models.package import Package
class Github(Upload):
"""
upload files to github releases
:ivar auth: requests authentication tuple
:ivar gh_owner: github repository owner
:ivar gh_repository: github repository name
"""
def __init__(self, architecture: str, configuration: Configuration) -> None:
"""
default constructor
:param architecture: repository architecture
:param configuration: configuration instance
"""
Upload.__init__(self, architecture, configuration)
self.gh_owner = configuration.get("github", "owner")
self.gh_repository = configuration.get("github", "repository")
gh_api_key = configuration.get("github", "api_key")
gh_username = configuration.get("github", "username", fallback=self.gh_owner)
self.auth = (gh_username, gh_api_key)
@staticmethod
def calculate_hash(path: Path) -> str:
"""
calculate file checksum. Github API does not provide hashes itself, so we have to handle it manually
:param path: path to local file
:return: calculated checksum of the file
"""
with path.open("rb") as local_file:
md5 = hashlib.md5(local_file.read()) # nosec
return md5.hexdigest()
@staticmethod
def get_body(local_files: Dict[Path, str]) -> str:
"""
generate release body from the checksums as returned from Github.get_hashes method
:param local_files: map of the paths to its checksum
:return: body to be inserted into release
"""
return "\n".join(f"{file.name} {md5}" for file, md5 in sorted(local_files.items()))
@staticmethod
def get_hashes(release: Dict[str, Any]) -> Dict[str, str]:
"""
get checksums of the content from the repository
:param release: release object
:return: map of the filename to its checksum as it is written in body
"""
body: str = release["body"] or ""
files = {}
for line in body.splitlines():
file, md5 = line.split()
files[file] = md5
return files
def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
"""
github request wrapper
:param method: request method
:param url: request url
:param kwargs: request parameters to be passed as is
:return: request response object
"""
try:
response = requests.request(method, url, auth=self.auth, **kwargs)
response.raise_for_status()
except requests.HTTPError as e:
self.logger.exception("could not perform %s request to %s: %s", method, url, exception_response_text(e))
raise
return response
def asset_remove(self, release: Dict[str, Any], name: str) -> None:
"""
remove asset from the release by name
:param release: release object
:param name: asset name
"""
try:
asset = next(asset for asset in release["assets"] if asset["name"] == name)
self._request("DELETE", asset["url"])
except StopIteration:
self.logger.info("no asset %s found in release %s", name, release["name"])
def asset_upload(self, release: Dict[str, Any], path: Path) -> None:
"""
upload asset to the release
:param release: release object
:param path: path to local file
"""
exists = any(path.name == asset["name"] for asset in release["assets"])
if exists:
self.asset_remove(release, path.name)
(url, _) = release["upload_url"].split("{") # it is parametrized url
(mime, _) = mimetypes.guess_type(path)
headers = {"Content-Type": mime} if mime is not None else {"Content-Type": "application/octet-stream"}
self._request("POST", url, params={"name": path.name}, data=path.open("rb"), headers=headers)
def get_local_files(self, path: Path) -> Dict[Path, str]:
"""
get all local files and their calculated checksums
:param path: local path to sync
:return: map of path objects to its checksum
"""
return {
local_file: self.calculate_hash(local_file)
for local_file in walk(path)
}
def files_remove(self, release: Dict[str, Any], local_files: Dict[Path, str], remote_files: Dict[str, str]) -> None:
"""
remove files from github
:param release: release object
:param local_files: map of local file paths to its checksum
:param remote_files: map of the remote files and its checksum
"""
local_filenames = {local_file.name for local_file in local_files}
for remote_file in remote_files:
if remote_file in local_filenames:
continue
self.asset_remove(release, remote_file)
def files_upload(self, release: Dict[str, Any], local_files: Dict[Path, str], remote_files: Dict[str, str]) -> None:
"""
upload files to github
:param release: release object
:param local_files: map of local file paths to its checksum
:param remote_files: map of the remote files and its checksum
"""
for local_file, checksum in local_files.items():
remote_checksum = remote_files.get(local_file.name)
if remote_checksum == checksum:
continue
self.asset_upload(release, local_file)
def release_create(self) -> Dict[str, Any]:
"""
create empty release
:return: github API release object for the new release
"""
response = self._request("POST", f"https://api.github.com/repos/{self.gh_owner}/{self.gh_repository}/releases",
json={"tag_name": self.architecture, "name": self.architecture})
release: Dict[str, Any] = response.json()
return release
def release_get(self) -> Optional[Dict[str, Any]]:
"""
get release object if any
:return: github API release object if release found and None otherwise
"""
try:
response = self._request(
"GET",
f"https://api.github.com/repos/{self.gh_owner}/{self.gh_repository}/releases/tags/{self.architecture}")
release: Dict[str, Any] = response.json()
return release
except requests.HTTPError as e:
status_code = e.response.status_code if e.response is not None else None
if status_code == 404:
return None
raise
def release_update(self, release: Dict[str, Any], body: str) -> None:
"""
update release
:param release: release object
:param body: new release body
"""
self._request("POST", release["url"], json={"body": body})
def sync(self, path: Path, built_packages: Iterable[Package]) -> None:
"""
sync data to remote server
:param path: local path to sync
:param built_packages: list of packages which has just been built
"""
release = self.release_get()
if release is None:
release = self.release_create()
remote_files = self.get_hashes(release)
local_files = self.get_local_files(path)
self.files_upload(release, local_files, remote_files)
self.files_remove(release, local_files, remote_files)
self.release_update(release, self.get_body(local_files))

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

@ -22,10 +22,11 @@ import hashlib
import mimetypes
from pathlib import Path
from typing import Any, Dict, Generator, Iterable
from typing import Any, Dict, Iterable
from ahriman.core.configuration import Configuration
from ahriman.core.upload.upload import Upload
from ahriman.core.util import walk
from ahriman.models.package import Package
@ -82,7 +83,7 @@ class S3(Upload):
return client.Bucket(configuration.get("s3", "bucket"))
@staticmethod
def remove_files(local_files: Dict[Path, str], remote_objects: Dict[Path, Any]) -> None:
def files_remove(local_files: Dict[Path, str], remote_objects: Dict[Path, Any]) -> None:
"""
remove files which have been removed locally
:param local_files: map of local path object to its checksum
@ -93,19 +94,33 @@ class S3(Upload):
continue
remote_object.delete()
def files_upload(self, path: Path, local_files: Dict[Path, str], remote_objects: Dict[Path, Any]) -> None:
"""
upload changed files to s3
:param path: local path to sync
:param local_files: map of local path object to its checksum
:param remote_objects: map of remote path object to the remote s3 object
"""
for local_file, checksum in local_files.items():
remote_object = remote_objects.get(local_file)
# 0 and -1 elements are " (double quote)
remote_checksum = remote_object.e_tag[1:-1] if remote_object is not None else None
if remote_checksum == checksum:
continue
local_path = path / local_file
remote_path = Path(self.architecture) / local_file
(mime, _) = mimetypes.guess_type(local_path)
extra_args = {"ContentType": mime} if mime is not None else None
self.bucket.upload_file(Filename=str(local_path), Key=str(remote_path), ExtraArgs=extra_args)
def get_local_files(self, path: Path) -> Dict[Path, str]:
"""
get all local files and their calculated checksums
:param path: local path to sync
:return: map of path object to its checksum
"""
# credits to https://stackoverflow.com/a/64915960
def walk(directory_path: Path) -> Generator[Path, None, None]:
for element in directory_path.iterdir():
if element.is_dir():
yield from walk(element)
continue
yield element
return {
local_file.relative_to(path): self.calculate_etag(local_file, self.chunk_size)
for local_file in walk(path)
@ -128,26 +143,5 @@ class S3(Upload):
remote_objects = self.get_remote_objects()
local_files = self.get_local_files(path)
self.upload_files(path, local_files, remote_objects)
self.remove_files(local_files, remote_objects)
def upload_files(self, path: Path, local_files: Dict[Path, str], remote_objects: Dict[Path, Any]) -> None:
"""
upload changed files to s3
:param path: local path to sync
:param local_files: map of local path object to its checksum
:param remote_objects: map of remote path object to the remote s3 object
"""
for local_file, checksum in local_files.items():
remote_object = remote_objects.get(local_file)
# 0 and -1 elements are " (double quote)
remote_checksum = remote_object.e_tag[1:-1] if remote_object is not None else None
if remote_checksum == checksum:
continue
local_path = path / local_file
remote_path = Path(self.architecture) / local_file
(mime, _) = mimetypes.guess_type(local_path)
extra_args = {"ContentType": mime} if mime is not None else None
self.bucket.upload_file(Filename=str(local_path), Key=str(remote_path), ExtraArgs=extra_args)
self.files_upload(path, local_files, remote_objects)
self.files_remove(local_files, remote_objects)

3
src/ahriman/core/upload/upload.py
View File

@ -64,6 +64,9 @@ class Upload:
if provider == UploadSettings.S3:
from ahriman.core.upload.s3 import S3
return S3(architecture, configuration)
if provider == UploadSettings.Github:
from ahriman.core.upload.github import Github
return Github(architecture, configuration)
return cls(architecture, configuration) # should never happen
def run(self, path: Path, built_packages: Iterable[Package]) -> None:

16
src/ahriman/core/util.py
View File

@ -23,7 +23,7 @@ import requests
from logging import Logger
from pathlib import Path
from typing import Optional, Union
from typing import Generator, Optional, Union
from ahriman.core.exceptions import InvalidOption
@ -106,3 +106,17 @@ def pretty_size(size: Optional[float], level: int = 0) -> str:
if size < 1024 or level >= 3:
return f"{size:.1f} {str_level()}"
return pretty_size(size / 1024, level + 1)
def walk(directory_path: Path) -> Generator[Path, None, None]:
"""
list all file paths in given directory
Credits to https://stackoverflow.com/a/64915960
:param directory_path: root directory path
:return: all found files in given directory with full path
"""
for element in directory_path.iterdir():
if element.is_dir():
yield from walk(element)
continue
yield element

4
src/ahriman/models/upload_settings.py
View File

@ -31,11 +31,13 @@ class UploadSettings(Enum):
:cvar Disabled: no sync will be performed, required for testing purpose
:cvar Rsync: sync via rsync
:cvar S3: sync to Amazon S3
:cvar Github: sync to github releases page
"""
Disabled = "disabled" # for testing purpose
Rsync = "rsync"
S3 = "s3"
Github = "github"
@classmethod
def from_option(cls: Type[UploadSettings], value: str) -> UploadSettings:
@ -48,4 +50,6 @@ class UploadSettings(Enum):
return cls.Rsync
if value.lower() in ("s3",):
return cls.S3
if value.lower() in ("github",):
return cls.Github
raise InvalidOption(value)

8
tests/ahriman/application/test_application.py
View File

@ -110,10 +110,10 @@ def test_add_archive(application: Application, package_ahriman: Package, mocker:
must add package from archive
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
move_mock = mocker.patch("shutil.move")
copy_mock = mocker.patch("shutil.copy")
application.add([package_ahriman.base], PackageSource.Archive, False)
move_mock.assert_called_once()
copy_mock.assert_called_once()
def test_add_remote(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
@ -149,11 +149,11 @@ def test_add_directory(application: Application, package_ahriman: Package, mocke
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
iterdir_mock = mocker.patch("pathlib.Path.iterdir",
return_value=[package.filepath for package in package_ahriman.packages.values()])
move_mock = mocker.patch("shutil.move")
copy_mock = mocker.patch("shutil.copy")
application.add([package_ahriman.base], PackageSource.Directory, False)
iterdir_mock.assert_called_once()
move_mock.assert_called_once()
copy_mock.assert_called_once()
def test_add_local(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:

29
tests/ahriman/core/test_util.py
View File

@ -2,10 +2,11 @@ import logging
import pytest
import subprocess
from pathlib import Path
from pytest_mock import MockerFixture
from ahriman.core.exceptions import InvalidOption
from ahriman.core.util import check_output, package_like, pretty_datetime, pretty_size
from ahriman.core.util import check_output, package_like, pretty_datetime, pretty_size, walk
from ahriman.models.package import Package
@ -138,3 +139,29 @@ def test_pretty_size_empty() -> None:
must generate empty string for None value
"""
assert pretty_size(None) == ""
def test_walk(resource_path_root: Path) -> None:
"""
must traverse directory recursively
"""
expected = sorted([
resource_path_root / "core/ahriman.ini",
resource_path_root / "core/logging.ini",
resource_path_root / "models/big_file_checksum",
resource_path_root / "models/empty_file_checksum",
resource_path_root / "models/package_ahriman_srcinfo",
resource_path_root / "models/package_tpacpi-bat-git_srcinfo",
resource_path_root / "models/package_yay_srcinfo",
resource_path_root / "web/templates/build-status/login-modal.jinja2",
resource_path_root / "web/templates/build-status/package-actions-modals.jinja2",
resource_path_root / "web/templates/build-status/package-actions-script.jinja2",
resource_path_root / "web/templates/static/favicon.ico",
resource_path_root / "web/templates/utils/bootstrap-scripts.jinja2",
resource_path_root / "web/templates/utils/style.jinja2",
resource_path_root / "web/templates/build-status.jinja2",
resource_path_root / "web/templates/email-index.jinja2",
resource_path_root / "web/templates/repo-index.jinja2",
])
local_files = list(sorted(walk(resource_path_root)))
assert local_files == expected

44
tests/ahriman/core/upload/conftest.py
View File

@ -1,16 +1,58 @@
import pytest
from collections import namedtuple
from typing import List
from typing import Any, Dict, List
from unittest.mock import MagicMock
from ahriman.core.configuration import Configuration
from ahriman.core.upload.github import Github
from ahriman.core.upload.rsync import Rsync
from ahriman.core.upload.s3 import S3
_s3_object = namedtuple("s3_object", ["key", "e_tag", "delete"])
@pytest.fixture
def github(configuration: Configuration) -> Github:
"""
fixture for github synchronization
:param configuration: configuration fixture
:return: github test instance
"""
return Github("x86_64", configuration)
@pytest.fixture
def github_release() -> Dict[str, Any]:
"""
fixture for the github release object
:return: github test release object
"""
return {
"url": "release_url",
"assets_url": "assets_url",
"upload_url": "upload_url{?name,label}",
"tag_name": "x86_64",
"name": "x86_64",
"assets": [{
"url": "asset_url",
"name": "asset_name",
}],
"body": None,
}
@pytest.fixture
def rsync(configuration: Configuration) -> Rsync:
"""
fixture for rsync synchronization
:param configuration: configuration fixture
:return: rsync test instance
"""
return Rsync("x86_64", configuration)
@pytest.fixture
def s3(configuration: Configuration) -> S3:
"""

266
tests/ahriman/core/upload/test_github.py Normal file
View File

@ -0,0 +1,266 @@
import pytest
import requests
from pathlib import Path
from pytest_mock import MockerFixture
from typing import Any, Dict
from unittest import mock
from unittest.mock import MagicMock
from ahriman.core.upload.github import Github
def test_calculate_hash_empty(resource_path_root: Path) -> None:
"""
must calculate checksum for empty file correctly
"""
path = resource_path_root / "models" / "empty_file_checksum"
assert Github.calculate_hash(path) == "d41d8cd98f00b204e9800998ecf8427e"
def test_calculate_hash_small(resource_path_root: Path) -> None:
"""
must calculate checksum for path which is single chunk
"""
path = resource_path_root / "models" / "package_ahriman_srcinfo"
assert Github.calculate_hash(path) == "a55f82198e56061295d405aeb58f4062"
def test_get_body_get_hashes() -> None:
"""
must generate readable body
"""
source = {Path("c"): "c_md5", Path("a"): "a_md5", Path("b"): "b_md5"}
body = Github.get_body(source)
parsed = Github.get_hashes({"body": body})
assert {fn.name: md5 for fn, md5 in source.items()} == parsed
def test_get_hashes_empty() -> None:
"""
must read empty body
"""
assert Github.get_hashes({"body": None}) == {}
def test_request(github: Github, mocker: MockerFixture) -> None:
"""
must call request method
"""
response_mock = MagicMock()
request_mock = mocker.patch("requests.request", return_value=response_mock)
github._request("GET", "url", arg="arg")
request_mock.assert_called_once_with("GET", "url", auth=github.auth, arg="arg")
response_mock.raise_for_status.assert_called_once()
def test_request_exception(github: Github, mocker: MockerFixture) -> None:
"""
must call request method and log HTTPError exception
"""
mocker.patch("requests.request", side_effect=requests.HTTPError())
with pytest.raises(requests.HTTPError):
github._request("GET", "url", arg="arg")
def test_asset_remove(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must remove asset from the release
"""
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.asset_remove(github_release, "asset_name")
request_mock.assert_called_with("DELETE", "asset_url")
def test_asset_remove_unknown(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must not fail if no asset found
"""
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.asset_remove(github_release, "unknown_asset_name")
request_mock.assert_not_called()
def test_asset_upload(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must upload asset to the repository
"""
mocker.patch("pathlib.Path.open", return_value=b"")
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
remove_mock = mocker.patch("ahriman.core.upload.github.Github.asset_remove")
github.asset_upload(github_release, Path("/root/new.tar.xz"))
request_mock.assert_called_with("POST", "upload_url", params={"name": "new.tar.xz"},
data=b"", headers={"Content-Type": "application/x-tar"})
remove_mock.assert_not_called()
def test_asset_upload_with_removal(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must remove existing file before upload
"""
mocker.patch("pathlib.Path.open", return_value=b"")
mocker.patch("ahriman.core.upload.github.Github._request")
remove_mock = mocker.patch("ahriman.core.upload.github.Github.asset_remove")
github.asset_upload(github_release, Path("asset_name"))
remove_mock.assert_called_with(github_release, "asset_name")
github.asset_upload(github_release, Path("/root/asset_name"))
remove_mock.assert_called_with(github_release, "asset_name")
def test_asset_upload_empty_mimetype(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must upload asset to the repository with empty mime type if cannot guess it
"""
mocker.patch("pathlib.Path.open", return_value=b"")
mocker.patch("ahriman.core.upload.github.Github.asset_remove")
mocker.patch("mimetypes.guess_type", return_value=(None, None))
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.asset_upload(github_release, Path("/root/new.tar.xz"))
request_mock.assert_called_with("POST", "upload_url", params={"name": "new.tar.xz"},
data=b"", headers={"Content-Type": "application/octet-stream"})
def test_get_local_files(github: Github, resource_path_root: Path, mocker: MockerFixture) -> None:
"""
must get all local files recursively
"""
walk_mock = mocker.patch("ahriman.core.util.walk")
github.get_local_files(resource_path_root)
walk_mock.assert_called()
def test_files_remove(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must remove files from the remote
"""
remove_mock = mocker.patch("ahriman.core.upload.github.Github.asset_remove")
github.files_remove(github_release, {Path("a"): "a"}, {"a": "a", "b": "b"})
remove_mock.assert_called_once_with(github_release, "b")
def test_files_remove_empty(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must remove nothing if nothing changed
"""
remove_mock = mocker.patch("ahriman.core.upload.github.Github.asset_remove")
github.files_remove(github_release, {Path("a"): "a"}, {"a": "a"})
remove_mock.assert_not_called()
def test_files_upload(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must upload files to the remote
"""
upload_mock = mocker.patch("ahriman.core.upload.github.Github.asset_upload")
github.files_upload(github_release, {Path("a"): "a", Path("b"): "c", Path("c"): "c"}, {"a": "a", "b": "b"})
upload_mock.assert_has_calls([
mock.call(github_release, Path("b")),
mock.call(github_release, Path("c")),
])
def test_files_upload_empty(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must upload nothing if nothing changed
"""
upload_mock = mocker.patch("ahriman.core.upload.github.Github.asset_upload")
github.files_upload(github_release, {Path("a"): "a"}, {"a": "a"})
upload_mock.assert_not_called()
def test_release_create(github: Github, mocker: MockerFixture) -> None:
"""
must create release
"""
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.release_create()
request_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True),
json={"tag_name": github.architecture, "name": github.architecture})
def test_release_get(github: Github, mocker: MockerFixture) -> None:
"""
must get release
"""
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.release_get()
request_mock.assert_called_once_with("GET", pytest.helpers.anyvar(str, True))
def test_release_get_empty(github: Github, mocker: MockerFixture) -> None:
"""
must return nothing in case of 404 status code
"""
response = requests.Response()
response.status_code = 404
mocker.patch("ahriman.core.upload.github.Github._request", side_effect=requests.HTTPError(response=response))
assert github.release_get() is None
def test_release_get_exception(github: Github, mocker: MockerFixture) -> None:
"""
must re-raise non HTTPError exception
"""
mocker.patch("ahriman.core.upload.github.Github._request", side_effect=Exception())
with pytest.raises(Exception):
github.release_get()
def test_release_get_exception_http_error(github: Github, mocker: MockerFixture) -> None:
"""
must re-raise HTTPError exception with code differs from 404
"""
exception = requests.HTTPError(response=requests.Response())
mocker.patch("ahriman.core.upload.github.Github._request", side_effect=exception)
with pytest.raises(requests.HTTPError):
github.release_get()
def test_release_update(github: Github, github_release: Dict[str, Any], mocker: MockerFixture) -> None:
"""
must update release
"""
request_mock = mocker.patch("ahriman.core.upload.github.Github._request")
github.release_update(github_release, "body")
request_mock.assert_called_once_with("POST", "release_url", json={"body": "body"})
def test_release_sync(github: Github, mocker: MockerFixture) -> None:
"""
must run sync command
"""
release_get_mock = mocker.patch("ahriman.core.upload.github.Github.release_get")
get_hashes_mock = mocker.patch("ahriman.core.upload.github.Github.get_hashes")
get_local_files_mock = mocker.patch("ahriman.core.upload.github.Github.get_local_files")
files_upload_mock = mocker.patch("ahriman.core.upload.github.Github.files_upload")
files_remove_mock = mocker.patch("ahriman.core.upload.github.Github.files_remove")
release_update_mock = mocker.patch("ahriman.core.upload.github.Github.release_update")
github.sync(Path("local"), [])
release_get_mock.assert_called_once()
get_hashes_mock.assert_called_once()
get_local_files_mock.assert_called_once()
files_upload_mock.assert_called_once()
files_remove_mock.assert_called_once()
release_update_mock.assert_called_once()
def test_release_sync_create_release(github: Github, mocker: MockerFixture) -> None:
"""
must create release in case if it does not exist
"""
mocker.patch("ahriman.core.upload.github.Github.release_get", return_value=None)
mocker.patch("ahriman.core.upload.github.Github.get_hashes")
mocker.patch("ahriman.core.upload.github.Github.get_local_files")
mocker.patch("ahriman.core.upload.github.Github.files_upload")
mocker.patch("ahriman.core.upload.github.Github.files_remove")
mocker.patch("ahriman.core.upload.github.Github.release_update")
release_create_mock = mocker.patch("ahriman.core.upload.github.Github.release_create")
github.sync(Path("local"), [])
release_create_mock.assert_called_once()

7
tests/ahriman/core/upload/test_rsync.py
View File

@ -1,16 +1,13 @@
from pathlib import Path
from pytest_mock import MockerFixture
from ahriman.core.configuration import Configuration
from ahriman.core.upload.rsync import Rsync
def test_sync(configuration: Configuration, mocker: MockerFixture) -> None:
def test_sync(rsync: Rsync, mocker: MockerFixture) -> None:
"""
must run sync command
"""
check_output_mock = mocker.patch("ahriman.core.upload.rsync.Rsync._check_output")
upload = Rsync("x86_64", configuration)
upload.sync(Path("path"), [])
rsync.sync(Path("path"), [])
check_output_mock.assert_called_once()

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

@ -34,7 +34,7 @@ def test_calculate_etag_small(resource_path_root: Path) -> None:
assert S3.calculate_etag(path, _chunk_size) == "a55f82198e56061295d405aeb58f4062"
def test_remove_files(s3_remote_objects: List[Any]) -> None:
def test_files_remove(s3_remote_objects: List[Any]) -> None:
"""
must remove remote objects
"""
@ -43,35 +43,49 @@ def test_remove_files(s3_remote_objects: List[Any]) -> None:
}
remote_objects = {Path(item.key): item for item in s3_remote_objects}
S3.remove_files(local_files, remote_objects)
S3.files_remove(local_files, remote_objects)
remote_objects[Path("x86_64/a")].delete.assert_called_once()
def test_get_local_files(s3: S3, resource_path_root: Path) -> None:
def test_files_upload(s3: S3, s3_remote_objects: List[Any], mocker: MockerFixture) -> None:
"""
must upload changed files
"""
def mimetype(path: Path) -> Tuple[Optional[str], None]:
return ("text/html", None) if path.name == "b" else (None, None)
root = Path("path")
local_files = {
Path(item.key.replace("a", "d")): item.e_tag.replace("b", "d").replace("\"", "")
for item in s3_remote_objects
}
remote_objects = {Path(item.key): item for item in s3_remote_objects}
mocker.patch("mimetypes.guess_type", side_effect=mimetype)
upload_mock = s3.bucket = MagicMock()
s3.files_upload(root, local_files, remote_objects)
upload_mock.upload_file.assert_has_calls(
[
mock.call(
Filename=str(root / s3.architecture / "b"),
Key=f"{s3.architecture}/{s3.architecture}/b",
ExtraArgs={"ContentType": "text/html"}),
mock.call(
Filename=str(root / s3.architecture / "d"),
Key=f"{s3.architecture}/{s3.architecture}/d",
ExtraArgs=None),
],
any_order=True)
def test_get_local_files(s3: S3, resource_path_root: Path, mocker: MockerFixture) -> None:
"""
must get all local files recursively
"""
expected = sorted([
Path("core/ahriman.ini"),
Path("core/logging.ini"),
Path("models/big_file_checksum"),
Path("models/empty_file_checksum"),
Path("models/package_ahriman_srcinfo"),
Path("models/package_tpacpi-bat-git_srcinfo"),
Path("models/package_yay_srcinfo"),
Path("web/templates/build-status/login-modal.jinja2"),
Path("web/templates/build-status/package-actions-modals.jinja2"),
Path("web/templates/build-status/package-actions-script.jinja2"),
Path("web/templates/static/favicon.ico"),
Path("web/templates/utils/bootstrap-scripts.jinja2"),
Path("web/templates/utils/style.jinja2"),
Path("web/templates/build-status.jinja2"),
Path("web/templates/email-index.jinja2"),
Path("web/templates/repo-index.jinja2"),
])
local_files = list(sorted(s3.get_local_files(resource_path_root).keys()))
assert local_files == expected
walk_mock = mocker.patch("ahriman.core.util.walk")
s3.get_local_files(resource_path_root)
walk_mock.assert_called()
def test_get_remote_objects(s3: S3, s3_remote_objects: List[Any]) -> None:
@ -92,43 +106,11 @@ def test_sync(s3: S3, mocker: MockerFixture) -> None:
"""
local_files_mock = mocker.patch("ahriman.core.upload.s3.S3.get_local_files")
remote_objects_mock = mocker.patch("ahriman.core.upload.s3.S3.get_remote_objects")
remove_files_mock = mocker.patch("ahriman.core.upload.s3.S3.remove_files")
upload_files_mock = mocker.patch("ahriman.core.upload.s3.S3.upload_files")
remove_files_mock = mocker.patch("ahriman.core.upload.s3.S3.files_remove")
upload_files_mock = mocker.patch("ahriman.core.upload.s3.S3.files_upload")
s3.sync(Path("root"), [])
local_files_mock.assert_called_once()
remote_objects_mock.assert_called_once()
remove_files_mock.assert_called_once()
upload_files_mock.assert_called_once()
def test_upload_files(s3: S3, s3_remote_objects: List[Any], mocker: MockerFixture) -> None:
"""
must upload changed files
"""
def mimetype(path: Path) -> Tuple[Optional[str], None]:
return ("text/html", None) if path.name == "b" else (None, None)
root = Path("path")
local_files = {
Path(item.key.replace("a", "d")): item.e_tag.replace("b", "d").replace("\"", "")
for item in s3_remote_objects
}
remote_objects = {Path(item.key): item for item in s3_remote_objects}
mocker.patch("mimetypes.guess_type", side_effect=mimetype)
upload_mock = s3.bucket = MagicMock()
s3.upload_files(root, local_files, remote_objects)
upload_mock.upload_file.assert_has_calls(
[
mock.call(
Filename=str(root / s3.architecture / "b"),
Key=f"{s3.architecture}/{s3.architecture}/b",
ExtraArgs={"ContentType": "text/html"}),
mock.call(
Filename=str(root / s3.architecture / "d"),
Key=f"{s3.architecture}/{s3.architecture}/d",
ExtraArgs=None),
],
any_order=True)

9
tests/ahriman/core/upload/test_upload.py
View File

@ -44,3 +44,12 @@ def test_upload_s3(configuration: Configuration, mocker: MockerFixture) -> None:
upload_mock = mocker.patch("ahriman.core.upload.s3.S3.sync")
Upload.load("x86_64", configuration, UploadSettings.S3.name).run(Path("path"), [])
upload_mock.assert_called_once()
def test_upload_github(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must upload via github
"""
upload_mock = mocker.patch("ahriman.core.upload.github.Github.sync")
Upload.load("x86_64", configuration, UploadSettings.Github.name).run(Path("path"), [])
upload_mock.assert_called_once()

3
tests/ahriman/models/test_upload_settings.py
View File

@ -21,3 +21,6 @@ def test_from_option_valid() -> None:
assert UploadSettings.from_option("s3") == UploadSettings.S3
assert UploadSettings.from_option("S3") == UploadSettings.S3
assert UploadSettings.from_option("github") == UploadSettings.Github
assert UploadSettings.from_option("GitHub") == UploadSettings.Github

5
tests/testresources/core/ahriman.ini
View File

@ -61,6 +61,11 @@ bucket = bucket
region = eu-central-1
secret_key =
[github]
api_key =
owner = arcan1s
repository = ahriman
[web]
debug = no
debug_check_host = no