mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-06-28 06:41:43 +00:00
feat: add separated web client for ahriman web services
This commit is contained in:
21
tests/ahriman/core/http/conftest.py
Normal file
21
tests/ahriman/core/http/conftest.py
Normal file
@ -0,0 +1,21 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.http import SyncAhrimanClient
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ahriman_client(configuration: Configuration) -> SyncAhrimanClient:
|
||||
"""
|
||||
ahriman web client fixture
|
||||
|
||||
Args:
|
||||
configuration(Configuration): configuration fixture
|
||||
|
||||
Returns:
|
||||
SyncAhrimanClient: ahriman web client test instance
|
||||
"""
|
||||
configuration.set("web", "port", "8080")
|
||||
_, repository_id = configuration.check_loaded()
|
||||
return WebClient(repository_id, configuration)
|
81
tests/ahriman/core/http/test_sync_ahriman_client.py
Normal file
81
tests/ahriman/core/http/test_sync_ahriman_client.py
Normal file
@ -0,0 +1,81 @@
|
||||
import pytest
|
||||
import requests
|
||||
import requests_unixsocket
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.http import SyncAhrimanClient
|
||||
from ahriman.models.user import User
|
||||
|
||||
|
||||
def test_session(ahriman_client: SyncAhrimanClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create normal requests session
|
||||
"""
|
||||
login_mock = mocker.patch("ahriman.core.http.SyncAhrimanClient._login")
|
||||
|
||||
assert isinstance(ahriman_client.session, requests.Session)
|
||||
assert not isinstance(ahriman_client.session, requests_unixsocket.Session)
|
||||
login_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_session_unix_socket(ahriman_client: SyncAhrimanClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create unix socket session
|
||||
"""
|
||||
login_mock = mocker.patch("ahriman.core.http.SyncAhrimanClient._login")
|
||||
ahriman_client.address = "http+unix://path"
|
||||
|
||||
assert isinstance(ahriman_client.session, requests_unixsocket.Session)
|
||||
login_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_login(ahriman_client: SyncAhrimanClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must login user
|
||||
"""
|
||||
ahriman_client.auth = (user.username, user.password)
|
||||
requests_mock = mocker.patch("ahriman.core.http.SyncAhrimanClient.make_request")
|
||||
payload = {
|
||||
"username": user.username,
|
||||
"password": user.password
|
||||
}
|
||||
session = requests.Session()
|
||||
|
||||
ahriman_client._login(session)
|
||||
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True), json=payload, session=session)
|
||||
|
||||
|
||||
def test_login_failed(ahriman_client: SyncAhrimanClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during login
|
||||
"""
|
||||
ahriman_client.user = user
|
||||
mocker.patch("requests.Session.request", side_effect=Exception())
|
||||
ahriman_client._login(requests.Session())
|
||||
|
||||
|
||||
def test_login_failed_http_error(ahriman_client: SyncAhrimanClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress HTTP exception happened during login
|
||||
"""
|
||||
ahriman_client.user = user
|
||||
mocker.patch("requests.Session.request", side_effect=requests.HTTPError())
|
||||
ahriman_client._login(requests.Session())
|
||||
|
||||
|
||||
def test_login_skip(ahriman_client: SyncAhrimanClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip login if no user set
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.Session.request")
|
||||
ahriman_client._login(requests.Session())
|
||||
requests_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_login_url(ahriman_client: SyncAhrimanClient) -> None:
|
||||
"""
|
||||
must generate login url correctly
|
||||
"""
|
||||
assert ahriman_client._login_url().startswith(ahriman_client.address)
|
||||
assert ahriman_client._login_url().endswith("/api/v1/login")
|
@ -2,7 +2,6 @@ import json
|
||||
import logging
|
||||
import pytest
|
||||
import requests
|
||||
import requests_unixsocket
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
@ -12,29 +11,6 @@ from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
from ahriman.models.log_record_id import LogRecordId
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.user import User
|
||||
|
||||
|
||||
def test_session(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create normal requests session
|
||||
"""
|
||||
login_mock = mocker.patch("ahriman.core.status.web_client.WebClient._login")
|
||||
|
||||
assert isinstance(web_client.session, requests.Session)
|
||||
assert not isinstance(web_client.session, requests_unixsocket.Session)
|
||||
login_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_session_unix_socket(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create unix socket session
|
||||
"""
|
||||
login_mock = mocker.patch("ahriman.core.status.web_client.WebClient._login")
|
||||
web_client.address = "http+unix://path"
|
||||
|
||||
assert isinstance(web_client.session, requests_unixsocket.Session)
|
||||
login_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_parse_address(configuration: Configuration) -> None:
|
||||
@ -55,57 +31,6 @@ def test_parse_address(configuration: Configuration) -> None:
|
||||
assert WebClient.parse_address(configuration) == ("status", "http://localhost:8082")
|
||||
|
||||
|
||||
def test_login(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must login user
|
||||
"""
|
||||
web_client.auth = (user.username, user.password)
|
||||
requests_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request")
|
||||
payload = {
|
||||
"username": user.username,
|
||||
"password": user.password
|
||||
}
|
||||
session = requests.Session()
|
||||
|
||||
web_client._login(session)
|
||||
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True), json=payload, session=session)
|
||||
|
||||
|
||||
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.request", side_effect=Exception())
|
||||
web_client._login(requests.Session())
|
||||
|
||||
|
||||
def test_login_failed_http_error(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress HTTP exception happened during login
|
||||
"""
|
||||
web_client.user = user
|
||||
mocker.patch("requests.Session.request", side_effect=requests.HTTPError())
|
||||
web_client._login(requests.Session())
|
||||
|
||||
|
||||
def test_login_skip(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip login if no user set
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.Session.request")
|
||||
web_client._login(requests.Session())
|
||||
requests_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_login_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate login url correctly
|
||||
"""
|
||||
assert web_client._login_url().startswith(web_client.address)
|
||||
assert web_client._login_url().endswith("/api/v1/login")
|
||||
|
||||
|
||||
def test_status_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate package status url correctly
|
||||
@ -377,11 +302,13 @@ def test_status_update(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
requests_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request")
|
||||
|
||||
web_client.status_update(BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True),
|
||||
params=web_client.repository_id.query(),
|
||||
json={
|
||||
"status": BuildStatusEnum.Unknown.value,
|
||||
})
|
||||
requests_mock.assert_called_once_with(
|
||||
"POST", pytest.helpers.anyvar(str, True),
|
||||
params=web_client.repository_id.query(),
|
||||
json={
|
||||
"status": BuildStatusEnum.Unknown.value,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_status_update_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
|
Reference in New Issue
Block a user