mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-07-24 19:29:57 +00:00
Auth support (#25)
* initial auth implementation * add create user parser * add tests * update dependencies list * add login annd logout to index also improve auth * realworld fixes * add method set_option to Configuration and also use it everywhere * split CreateUser handler to additional read method * check user duplicate on auth mapping read * generate salt by using passlib instead of random.choice * case-insensetive usernames * update dependencies * update configuration reference * improve tests * fix codefactor errors * hide fields if authorization is enabled, but no auth supplied * add settings object for auth provider * readme update
This commit is contained in:
GitHub
13
tests/ahriman/core/auth/conftest.py
Normal file
13
tests/ahriman/core/auth/conftest.py
Normal file
@ -0,0 +1,13 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.auth.mapping_auth import MappingAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mapping_auth(configuration: Configuration) -> MappingAuth:
|
||||
"""
|
||||
auth provider fixture
|
||||
:return: auth service instance
|
||||
"""
|
||||
return MappingAuth(configuration)
|
||||
81
tests/ahriman/core/auth/test_auth.py
Normal file
81
tests/ahriman/core/auth/test_auth.py
Normal file
@ -0,0 +1,81 @@
|
||||
from ahriman.core.auth.auth import Auth
|
||||
from ahriman.core.auth.mapping_auth import MappingAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.models.user import User
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
|
||||
def test_load_dummy(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load dummy validator if authorization is not enabled
|
||||
"""
|
||||
configuration.set_option("auth", "target", "disabled")
|
||||
auth = Auth.load(configuration)
|
||||
assert isinstance(auth, Auth)
|
||||
|
||||
|
||||
def test_load_dummy_empty(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load dummy validator if no option set
|
||||
"""
|
||||
auth = Auth.load(configuration)
|
||||
assert isinstance(auth, Auth)
|
||||
|
||||
|
||||
def test_load_mapping(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load mapping validator if option set
|
||||
"""
|
||||
configuration.set_option("auth", "target", "configuration")
|
||||
auth = Auth.load(configuration)
|
||||
assert isinstance(auth, MappingAuth)
|
||||
|
||||
|
||||
def test_check_credentials(auth: Auth, user: User) -> None:
|
||||
"""
|
||||
must pass any credentials
|
||||
"""
|
||||
assert auth.check_credentials(user.username, user.password)
|
||||
assert auth.check_credentials(None, "")
|
||||
assert auth.check_credentials("", None)
|
||||
assert auth.check_credentials(None, None)
|
||||
|
||||
|
||||
def test_is_safe_request(auth: Auth) -> None:
|
||||
"""
|
||||
must validate safe request
|
||||
"""
|
||||
# login and logout are always safe
|
||||
assert auth.is_safe_request("/login")
|
||||
assert auth.is_safe_request("/logout")
|
||||
|
||||
auth.allowed_paths.add("/safe")
|
||||
auth.allowed_paths_groups.add("/unsafe/safe")
|
||||
|
||||
assert auth.is_safe_request("/safe")
|
||||
assert not auth.is_safe_request("/unsafe")
|
||||
assert auth.is_safe_request("/unsafe/safe")
|
||||
assert auth.is_safe_request("/unsafe/safe/suffix")
|
||||
|
||||
|
||||
def test_is_safe_request_empty(auth: Auth) -> None:
|
||||
"""
|
||||
must not allow requests without path
|
||||
"""
|
||||
assert not auth.is_safe_request(None)
|
||||
assert not auth.is_safe_request("")
|
||||
|
||||
|
||||
def test_known_username(auth: Auth, user: User) -> None:
|
||||
"""
|
||||
must allow any username
|
||||
"""
|
||||
assert auth.known_username(user.username)
|
||||
|
||||
|
||||
def test_verify_access(auth: Auth, user: User) -> None:
|
||||
"""
|
||||
must allow any access
|
||||
"""
|
||||
assert auth.verify_access(user.username, user.access, None)
|
||||
assert auth.verify_access(user.username, UserAccess.Write, None)
|
||||
102
tests/ahriman/core/auth/test_helpers.py
Normal file
102
tests/ahriman/core/auth/test_helpers.py
Normal file
@ -0,0 +1,102 @@
|
||||
import importlib
|
||||
import sys
|
||||
|
||||
import ahriman.core.auth.helpers as helpers
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
|
||||
def test_import_aiohttp_security() -> None:
|
||||
"""
|
||||
must import aiohttp_security correctly
|
||||
"""
|
||||
assert helpers._has_aiohttp_security
|
||||
|
||||
|
||||
def test_import_aiohttp_security_missing(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must set missing flag if no aiohttp_security module found
|
||||
"""
|
||||
mocker.patch.dict(sys.modules, {"aiohttp_security": None})
|
||||
importlib.reload(helpers)
|
||||
assert not helpers._has_aiohttp_security
|
||||
|
||||
|
||||
async def test_authorized_userid_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call authorized_userid from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
authorized_userid_mock = mocker.patch("aiohttp_security.authorized_userid")
|
||||
await helpers.authorized_userid()
|
||||
authorized_userid_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_authorized_userid_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call authorized_userid from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
authorized_userid_mock = mocker.patch("aiohttp_security.authorized_userid")
|
||||
await helpers.authorized_userid()
|
||||
authorized_userid_mock.assert_called_once()
|
||||
|
||||
|
||||
async def test_check_authorized_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call check_authorized from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
check_authorized_mock = mocker.patch("aiohttp_security.check_authorized")
|
||||
await helpers.check_authorized()
|
||||
check_authorized_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_check_authorized_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call check_authorized from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
check_authorized_mock = mocker.patch("aiohttp_security.check_authorized")
|
||||
await helpers.check_authorized()
|
||||
check_authorized_mock.assert_called_once()
|
||||
|
||||
|
||||
async def test_forget_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call forget from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
await helpers.forget()
|
||||
forget_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_forget_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call forget from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
await helpers.forget()
|
||||
forget_mock.assert_called_once()
|
||||
|
||||
|
||||
async def test_remember_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call remember from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
await helpers.remember()
|
||||
remember_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_remember_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call remember from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
await helpers.remember()
|
||||
remember_mock.assert_called_once()
|
||||
121
tests/ahriman/core/auth/test_mapping_auth.py
Normal file
121
tests/ahriman/core/auth/test_mapping_auth.py
Normal file
@ -0,0 +1,121 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.auth.mapping_auth import MappingAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.exceptions import DuplicateUser
|
||||
from ahriman.models.user import User
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
|
||||
def test_get_users(mapping_auth: MappingAuth, configuration: Configuration) -> None:
|
||||
"""
|
||||
must return valid user list
|
||||
"""
|
||||
user_write = User("user_write", "pwd_write", UserAccess.Write)
|
||||
write_section = Configuration.section_name("auth", user_write.access.value)
|
||||
configuration.set_option(write_section, user_write.username, user_write.password)
|
||||
user_read = User("user_read", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user_read.access.value)
|
||||
configuration.set_option(read_section, user_read.username, user_read.password)
|
||||
user_read = User("user_read", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user_read.access.value)
|
||||
configuration.set_option(read_section, user_read.username, user_read.password)
|
||||
|
||||
users = mapping_auth.get_users(configuration)
|
||||
expected = {user_write.username: user_write, user_read.username: user_read}
|
||||
assert users == expected
|
||||
|
||||
|
||||
def test_get_users_normalized(mapping_auth: MappingAuth, configuration: Configuration) -> None:
|
||||
"""
|
||||
must return user list with normalized usernames in keys
|
||||
"""
|
||||
user = User("UsEr", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user.access.value)
|
||||
configuration.set_option(read_section, user.username, user.password)
|
||||
|
||||
users = mapping_auth.get_users(configuration)
|
||||
expected = user.username.lower()
|
||||
assert expected in users
|
||||
assert users[expected].username == expected
|
||||
|
||||
|
||||
def test_get_users_duplicate(mapping_auth: MappingAuth, configuration: Configuration, user: User) -> None:
|
||||
"""
|
||||
must raise exception on duplicate username
|
||||
"""
|
||||
write_section = Configuration.section_name("auth", UserAccess.Write.value)
|
||||
configuration.set_option(write_section, user.username, user.password)
|
||||
read_section = Configuration.section_name("auth", UserAccess.Read.value)
|
||||
configuration.set_option(read_section, user.username, user.password)
|
||||
|
||||
with pytest.raises(DuplicateUser):
|
||||
mapping_auth.get_users(configuration)
|
||||
|
||||
|
||||
def test_check_credentials(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must return true for valid credentials
|
||||
"""
|
||||
current_password = user.password
|
||||
user.password = user.hash_password(user.password, mapping_auth.salt)
|
||||
mapping_auth._users[user.username] = user
|
||||
assert mapping_auth.check_credentials(user.username, current_password)
|
||||
assert not mapping_auth.check_credentials(user.username, user.password) # here password is hashed so it is invalid
|
||||
|
||||
|
||||
def test_check_credentials_empty(mapping_auth: MappingAuth) -> None:
|
||||
"""
|
||||
must reject on empty credentials
|
||||
"""
|
||||
assert not mapping_auth.check_credentials(None, "")
|
||||
assert not mapping_auth.check_credentials("", None)
|
||||
assert not mapping_auth.check_credentials(None, None)
|
||||
|
||||
|
||||
def test_check_credentials_unknown(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must reject on unknown user
|
||||
"""
|
||||
assert not mapping_auth.check_credentials(user.username, user.password)
|
||||
|
||||
|
||||
def test_get_user(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must return user from storage by username
|
||||
"""
|
||||
mapping_auth._users[user.username] = user
|
||||
assert mapping_auth.get_user(user.username) == user
|
||||
|
||||
|
||||
def test_get_user_normalized(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must return user from storage by username case-insensitive
|
||||
"""
|
||||
mapping_auth._users[user.username] = user
|
||||
assert mapping_auth.get_user(user.username.upper()) == user
|
||||
|
||||
|
||||
def test_get_user_unknown(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must return None in case if no user found
|
||||
"""
|
||||
assert mapping_auth.get_user(user.username) is None
|
||||
|
||||
|
||||
def test_known_username(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must allow only known users
|
||||
"""
|
||||
mapping_auth._users[user.username] = user
|
||||
assert mapping_auth.known_username(user.username)
|
||||
assert not mapping_auth.known_username(user.password)
|
||||
|
||||
|
||||
def test_verify_access(mapping_auth: MappingAuth, user: User) -> None:
|
||||
"""
|
||||
must verify user access
|
||||
"""
|
||||
mapping_auth._users[user.username] = user
|
||||
assert mapping_auth.verify_access(user.username, user.access, None)
|
||||
assert not mapping_auth.verify_access(user.username, UserAccess.Write, None)
|
||||
@ -2,6 +2,7 @@ import pytest
|
||||
|
||||
from ahriman.core.alpm.pacman import Pacman
|
||||
from ahriman.core.alpm.repo import Repo
|
||||
from ahriman.core.auth.auth import Auth
|
||||
from ahriman.core.build_tools.task import Task
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.tree import Leaf
|
||||
|
||||
@ -23,8 +23,8 @@ def test_send_auth(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must send an email with attachment with auth
|
||||
"""
|
||||
configuration.set("email", "user", "username")
|
||||
configuration.set("email", "password", "password")
|
||||
configuration.set_option("email", "user", "username")
|
||||
configuration.set_option("email", "password", "password")
|
||||
smtp_mock = mocker.patch("smtplib.SMTP")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -36,7 +36,7 @@ def test_send_auth_no_password(configuration: Configuration, mocker: MockerFixtu
|
||||
"""
|
||||
must send an email with attachment without auth if no password supplied
|
||||
"""
|
||||
configuration.set("email", "user", "username")
|
||||
configuration.set_option("email", "user", "username")
|
||||
smtp_mock = mocker.patch("smtplib.SMTP")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -48,7 +48,7 @@ def test_send_auth_no_user(configuration: Configuration, mocker: MockerFixture)
|
||||
"""
|
||||
must send an email with attachment without auth if no user supplied
|
||||
"""
|
||||
configuration.set("email", "password", "password")
|
||||
configuration.set_option("email", "password", "password")
|
||||
smtp_mock = mocker.patch("smtplib.SMTP")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -60,7 +60,7 @@ def test_send_ssl_tls(configuration: Configuration, mocker: MockerFixture) -> No
|
||||
"""
|
||||
must send an email with attachment with ssl/tls
|
||||
"""
|
||||
configuration.set("email", "ssl", "ssl")
|
||||
configuration.set_option("email", "ssl", "ssl")
|
||||
smtp_mock = mocker.patch("smtplib.SMTP_SSL")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -75,7 +75,7 @@ def test_send_starttls(configuration: Configuration, mocker: MockerFixture) -> N
|
||||
"""
|
||||
must send an email with attachment with starttls
|
||||
"""
|
||||
configuration.set("email", "ssl", "starttls")
|
||||
configuration.set_option("email", "ssl", "starttls")
|
||||
smtp_mock = mocker.patch("smtplib.SMTP")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -109,7 +109,7 @@ def test_generate_no_empty(configuration: Configuration, package_ahriman: Packag
|
||||
"""
|
||||
must not generate report with built packages if no_empty_report is set
|
||||
"""
|
||||
configuration.set("email", "no_empty_report", "yes")
|
||||
configuration.set_option("email", "no_empty_report", "yes")
|
||||
send_mock = mocker.patch("ahriman.core.report.email.Email._send")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
@ -122,7 +122,7 @@ def test_generate_no_empty_with_built(configuration: Configuration, package_ahri
|
||||
"""
|
||||
must generate report with built packages if no_empty_report is set
|
||||
"""
|
||||
configuration.set("email", "no_empty_report", "yes")
|
||||
configuration.set_option("email", "no_empty_report", "yes")
|
||||
send_mock = mocker.patch("ahriman.core.report.email.Email._send")
|
||||
|
||||
report = Email("x86_64", configuration)
|
||||
|
||||
@ -2,6 +2,7 @@ import pytest
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.status.client import Client
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
@ -40,9 +41,11 @@ def client() -> Client:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def web_client() -> WebClient:
|
||||
def web_client(configuration: Configuration) -> WebClient:
|
||||
"""
|
||||
fixture for web client
|
||||
:param configuration: configuration fixture
|
||||
:return: web client test instance
|
||||
"""
|
||||
return WebClient("localhost", 8080)
|
||||
configuration.set("web", "port", 8080)
|
||||
return WebClient(configuration)
|
||||
|
||||
@ -17,10 +17,18 @@ def test_load_dummy_client(configuration: Configuration) -> None:
|
||||
|
||||
def test_load_full_client(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load full client if no settings set
|
||||
must load full client if settings set
|
||||
"""
|
||||
configuration.set("web", "host", "localhost")
|
||||
configuration.set("web", "port", "8080")
|
||||
configuration.set_option("web", "host", "localhost")
|
||||
configuration.set_option("web", "port", "8080")
|
||||
assert isinstance(Client.load(configuration), WebClient)
|
||||
|
||||
|
||||
def test_load_full_client_from_address(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load full client if settings set
|
||||
"""
|
||||
configuration.set_option("web", "address", "http://localhost:8080")
|
||||
assert isinstance(Client.load(configuration), WebClient)
|
||||
|
||||
|
||||
|
||||
@ -5,41 +5,97 @@ import requests
|
||||
from pytest_mock import MockerFixture
|
||||
from requests import Response
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.user import User
|
||||
|
||||
|
||||
def test_ahriman_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate service status url correctly
|
||||
"""
|
||||
assert web_client._ahriman_url().startswith(f"http://{web_client.host}:{web_client.port}")
|
||||
assert web_client._ahriman_url().endswith("/api/v1/ahriman")
|
||||
|
||||
|
||||
def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must generate package status correctly
|
||||
"""
|
||||
assert web_client._package_url(package_ahriman.base).startswith(f"http://{web_client.host}:{web_client.port}")
|
||||
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
|
||||
assert web_client._ahriman_url.startswith(web_client.address)
|
||||
assert web_client._ahriman_url.endswith("/api/v1/ahriman")
|
||||
|
||||
|
||||
def test_status_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate service status url correctly
|
||||
"""
|
||||
assert web_client._status_url().startswith(f"http://{web_client.host}:{web_client.port}")
|
||||
assert web_client._status_url().endswith("/api/v1/status")
|
||||
assert web_client._status_url.startswith(web_client.address)
|
||||
assert web_client._status_url.endswith("/api/v1/status")
|
||||
|
||||
|
||||
def test_parse_address(configuration: Configuration) -> None:
|
||||
"""
|
||||
must extract address correctly
|
||||
"""
|
||||
configuration.set_option("web", "host", "localhost")
|
||||
configuration.set_option("web", "port", "8080")
|
||||
assert WebClient.parse_address(configuration) == "http://localhost:8080"
|
||||
|
||||
configuration.set_option("web", "address", "http://localhost:8081")
|
||||
assert WebClient.parse_address(configuration) == "http://localhost:8081"
|
||||
|
||||
|
||||
def test_login(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must login user
|
||||
"""
|
||||
web_client.user = user
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
payload = {
|
||||
"username": user.username,
|
||||
"password": user.password
|
||||
}
|
||||
|
||||
web_client._login()
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json=payload)
|
||||
|
||||
|
||||
def test_login_failed(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during login
|
||||
"""
|
||||
web_client.user = user
|
||||
mocker.patch("requests.Session.post", side_effect=Exception())
|
||||
web_client._login()
|
||||
|
||||
|
||||
def test_login_failed_http_error(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during login
|
||||
"""
|
||||
web_client.user = user
|
||||
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
|
||||
web_client._login()
|
||||
|
||||
|
||||
def test_login_skip(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip login if no user set
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
web_client._login()
|
||||
requests_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must generate package status correctly
|
||||
"""
|
||||
assert web_client._package_url(package_ahriman.base).startswith(web_client.address)
|
||||
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
|
||||
|
||||
|
||||
def test_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process package addition
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
payload = pytest.helpers.get_package_status(package_ahriman)
|
||||
|
||||
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
@ -50,7 +106,7 @@ def test_add_failed(web_client: WebClient, package_ahriman: Package, mocker: Moc
|
||||
"""
|
||||
must suppress any exception happened during addition
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
mocker.patch("requests.Session.post", side_effect=Exception())
|
||||
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
@ -58,7 +114,7 @@ def test_add_failed_http_error(web_client: WebClient, package_ahriman: Package,
|
||||
"""
|
||||
must suppress any exception happened during addition
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
|
||||
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
@ -71,7 +127,7 @@ def test_get_all(web_client: WebClient, package_ahriman: Package, mocker: Mocker
|
||||
response_obj._content = json.dumps(response).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get(None)
|
||||
requests_mock.assert_called_once()
|
||||
@ -83,7 +139,7 @@ def test_get_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=Exception())
|
||||
mocker.patch("requests.Session.get", side_effect=Exception())
|
||||
assert web_client.get(None) == []
|
||||
|
||||
|
||||
@ -91,7 +147,7 @@ def test_get_failed_http_error(web_client: WebClient, mocker: MockerFixture) ->
|
||||
"""
|
||||
must suppress any exception happened during status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
|
||||
assert web_client.get(None) == []
|
||||
|
||||
|
||||
@ -104,7 +160,7 @@ def test_get_single(web_client: WebClient, package_ahriman: Package, mocker: Moc
|
||||
response_obj._content = json.dumps(response).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get(package_ahriman.base)
|
||||
requests_mock.assert_called_once()
|
||||
@ -120,7 +176,7 @@ def test_get_internal(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
response_obj._content = json.dumps(InternalStatus(architecture="x86_64").view()).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get_internal()
|
||||
requests_mock.assert_called_once()
|
||||
@ -131,7 +187,7 @@ def test_get_internal_failed(web_client: WebClient, mocker: MockerFixture) -> No
|
||||
"""
|
||||
must suppress any exception happened during web service status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=Exception())
|
||||
mocker.patch("requests.Session.get", side_effect=Exception())
|
||||
assert web_client.get_internal() == InternalStatus()
|
||||
|
||||
|
||||
@ -139,7 +195,7 @@ def test_get_internal_failed_http_error(web_client: WebClient, mocker: MockerFix
|
||||
"""
|
||||
must suppress any exception happened during web service status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
|
||||
assert web_client.get_internal() == InternalStatus()
|
||||
|
||||
|
||||
@ -151,7 +207,7 @@ def test_get_self(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
response_obj._content = json.dumps(BuildStatus().view()).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get_self()
|
||||
requests_mock.assert_called_once()
|
||||
@ -162,7 +218,7 @@ def test_get_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during service status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=Exception())
|
||||
mocker.patch("requests.Session.get", side_effect=Exception())
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
@ -170,7 +226,7 @@ def test_get_self_failed_http_error(web_client: WebClient, mocker: MockerFixture
|
||||
"""
|
||||
must suppress any exception happened during service status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
@ -178,7 +234,7 @@ def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerF
|
||||
"""
|
||||
must process package removal
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.delete")
|
||||
requests_mock = mocker.patch("requests.Session.delete")
|
||||
|
||||
web_client.remove(package_ahriman.base)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True))
|
||||
@ -188,7 +244,7 @@ def test_remove_failed(web_client: WebClient, package_ahriman: Package, mocker:
|
||||
"""
|
||||
must suppress any exception happened during removal
|
||||
"""
|
||||
mocker.patch("requests.delete", side_effect=Exception())
|
||||
mocker.patch("requests.Session.delete", side_effect=Exception())
|
||||
web_client.remove(package_ahriman.base)
|
||||
|
||||
|
||||
@ -196,7 +252,7 @@ def test_remove_failed_http_error(web_client: WebClient, package_ahriman: Packag
|
||||
"""
|
||||
must suppress any exception happened during removal
|
||||
"""
|
||||
mocker.patch("requests.delete", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.delete", side_effect=requests.exceptions.HTTPError())
|
||||
web_client.remove(package_ahriman.base)
|
||||
|
||||
|
||||
@ -204,7 +260,7 @@ def test_update(web_client: WebClient, package_ahriman: Package, mocker: MockerF
|
||||
"""
|
||||
must process package update
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
|
||||
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json={"status": BuildStatusEnum.Unknown.value})
|
||||
@ -214,7 +270,7 @@ def test_update_failed(web_client: WebClient, package_ahriman: Package, mocker:
|
||||
"""
|
||||
must suppress any exception happened during update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
mocker.patch("requests.Session.post", side_effect=Exception())
|
||||
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
@ -222,7 +278,7 @@ def test_update_failed_http_error(web_client: WebClient, package_ahriman: Packag
|
||||
"""
|
||||
must suppress any exception happened during update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
|
||||
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
@ -230,7 +286,7 @@ def test_update_self(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process service update
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
|
||||
web_client.update_self(BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json={"status": BuildStatusEnum.Unknown.value})
|
||||
@ -240,7 +296,7 @@ def test_update_self_failed(web_client: WebClient, mocker: MockerFixture) -> Non
|
||||
"""
|
||||
must suppress any exception happened during service update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
mocker.patch("requests.Session.post", side_effect=Exception())
|
||||
web_client.update_self(BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
@ -248,5 +304,5 @@ def test_update_self_failed_http_error(web_client: WebClient, mocker: MockerFixt
|
||||
"""
|
||||
must suppress any exception happened during service update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=requests.exceptions.HTTPError())
|
||||
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
|
||||
web_client.update_self(BuildStatusEnum.Unknown)
|
||||
|
||||
@ -9,7 +9,7 @@ def test_from_path(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load configuration
|
||||
"""
|
||||
read_mock = mocker.patch("configparser.RawConfigParser.read")
|
||||
read_mock = mocker.patch("ahriman.core.configuration.Configuration.read")
|
||||
load_includes_mock = mocker.patch("ahriman.core.configuration.Configuration.load_includes")
|
||||
load_logging_mock = mocker.patch("ahriman.core.configuration.Configuration.load_logging")
|
||||
path = Path("path")
|
||||
@ -33,7 +33,7 @@ def test_absolute_path_for_absolute(configuration: Configuration) -> None:
|
||||
must not change path for absolute path in settings
|
||||
"""
|
||||
path = Path("/a/b/c")
|
||||
configuration.set("build", "path", str(path))
|
||||
configuration.set_option("build", "path", str(path))
|
||||
assert configuration.getpath("build", "path") == path
|
||||
|
||||
|
||||
@ -42,7 +42,7 @@ def test_absolute_path_for_relative(configuration: Configuration) -> None:
|
||||
must prepend root path to relative path
|
||||
"""
|
||||
path = Path("a")
|
||||
configuration.set("build", "path", str(path))
|
||||
configuration.set_option("build", "path", str(path))
|
||||
result = configuration.getpath("build", "path")
|
||||
assert result.is_absolute()
|
||||
assert result.parent == configuration.path.parent
|
||||
@ -61,8 +61,7 @@ def test_dump_architecture_specific(configuration: Configuration) -> None:
|
||||
dump must contain architecture specific settings
|
||||
"""
|
||||
section = configuration.section_name("build", "x86_64")
|
||||
configuration.add_section(section)
|
||||
configuration.set(section, "archbuild_flags", "hello flag")
|
||||
configuration.set_option(section, "archbuild_flags", "hello flag")
|
||||
configuration.merge_sections("x86_64")
|
||||
|
||||
dump = configuration.dump()
|
||||
@ -76,7 +75,7 @@ def test_getlist(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return list of string correctly
|
||||
"""
|
||||
configuration.set("build", "test_list", "a b c")
|
||||
configuration.set_option("build", "test_list", "a b c")
|
||||
assert configuration.getlist("build", "test_list") == ["a", "b", "c"]
|
||||
|
||||
|
||||
@ -85,7 +84,7 @@ def test_getlist_empty(configuration: Configuration) -> None:
|
||||
must return list of string correctly for non-existing option
|
||||
"""
|
||||
assert configuration.getlist("build", "test_list") == []
|
||||
configuration.set("build", "test_list", "")
|
||||
configuration.set_option("build", "test_list", "")
|
||||
assert configuration.getlist("build", "test_list") == []
|
||||
|
||||
|
||||
@ -93,7 +92,7 @@ def test_getlist_single(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return list of strings for single string
|
||||
"""
|
||||
configuration.set("build", "test_list", "a")
|
||||
configuration.set_option("build", "test_list", "a")
|
||||
assert configuration.getlist("build", "test_list") == ["a"]
|
||||
|
||||
|
||||
@ -101,7 +100,7 @@ def test_load_includes_missing(configuration: Configuration) -> None:
|
||||
"""
|
||||
must not fail if not include directory found
|
||||
"""
|
||||
configuration.set("settings", "include", "path")
|
||||
configuration.set_option("settings", "include", "path")
|
||||
configuration.load_includes()
|
||||
|
||||
|
||||
@ -144,8 +143,23 @@ def test_merge_sections_missing(configuration: Configuration) -> None:
|
||||
"""
|
||||
section = configuration.section_name("build", "x86_64")
|
||||
configuration.remove_section("build")
|
||||
configuration.add_section(section)
|
||||
configuration.set(section, "key", "value")
|
||||
configuration.set_option(section, "key", "value")
|
||||
|
||||
configuration.merge_sections("x86_64")
|
||||
assert configuration.get("build", "key") == "value"
|
||||
|
||||
|
||||
def test_set_option(configuration: Configuration) -> None:
|
||||
"""
|
||||
must set option correctly
|
||||
"""
|
||||
configuration.set_option("settings", "option", "value")
|
||||
assert configuration.get("settings", "option") == "value"
|
||||
|
||||
|
||||
def test_set_option_new_section(configuration: Configuration) -> None:
|
||||
"""
|
||||
must set option correctly even if no section found
|
||||
"""
|
||||
configuration.set_option("section", "option", "value")
|
||||
assert configuration.get("section", "option") == "value"
|
||||
|
||||
@ -61,6 +61,8 @@ def test_get_local_files(s3: S3, resource_path_root: Path) -> None:
|
||||
Path("models/package_yay_srcinfo"),
|
||||
Path("web/templates/search-line.jinja2"),
|
||||
Path("web/templates/build-status.jinja2"),
|
||||
Path("web/templates/login-form.jinja2"),
|
||||
Path("web/templates/login-form-hide.jinja2"),
|
||||
Path("web/templates/repo-index.jinja2"),
|
||||
Path("web/templates/sorttable.jinja2"),
|
||||
Path("web/templates/style.jinja2"),
|
||||
|
||||
Reference in New Issue
Block a user