mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-07-29 13:49:57 +00:00
implement support of unix socket for server
This feature can be used for unauthorized access to apis - e.g. for reporting service if it is run on the same machine. Since now it becomes recommended way for the interprocess communication, thus some options (e.g. creating user with as-service flag) are no longer available now
This commit is contained in:
@ -43,6 +43,14 @@ def test_load_full_client_from_address(configuration: Configuration) -> None:
|
||||
assert isinstance(Client.load(configuration, report=True), WebClient)
|
||||
|
||||
|
||||
def test_load_full_client_from_unix_socket(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load full client by using unix socket
|
||||
"""
|
||||
configuration.set_option("web", "unix_socket", "/var/lib/ahriman/ahriman-web.sock")
|
||||
assert isinstance(Client.load(configuration, report=True), WebClient)
|
||||
|
||||
|
||||
def test_add(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must process package addition without errors
|
||||
|
@ -2,6 +2,7 @@ import json
|
||||
import logging
|
||||
import pytest
|
||||
import requests
|
||||
import requests_unixsocket
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from requests import Response
|
||||
@ -36,10 +37,36 @@ def test_parse_address(configuration: Configuration) -> None:
|
||||
"""
|
||||
configuration.set_option("web", "host", "localhost")
|
||||
configuration.set_option("web", "port", "8080")
|
||||
assert WebClient.parse_address(configuration) == "http://localhost:8080"
|
||||
assert WebClient.parse_address(configuration) == ("http://localhost:8080", False)
|
||||
|
||||
configuration.set_option("web", "address", "http://localhost:8081")
|
||||
assert WebClient.parse_address(configuration) == "http://localhost:8081"
|
||||
assert WebClient.parse_address(configuration) == ("http://localhost:8081", False)
|
||||
|
||||
configuration.set_option("web", "unix_socket", "/run/ahriman.sock")
|
||||
assert WebClient.parse_address(configuration) == ("http+unix://%2Frun%2Fahriman.sock", True)
|
||||
|
||||
|
||||
def test_create_session(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create normal requests session
|
||||
"""
|
||||
login_mock = mocker.patch("ahriman.core.status.web_client.WebClient._login")
|
||||
|
||||
session = web_client._create_session(use_unix_socket=False)
|
||||
assert isinstance(session, requests.Session)
|
||||
assert not isinstance(session, requests_unixsocket.Session)
|
||||
login_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_create_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")
|
||||
|
||||
session = web_client._create_session(use_unix_socket=True)
|
||||
assert isinstance(session, requests_unixsocket.Session)
|
||||
login_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_login(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
|
||||
|
@ -1,6 +1,5 @@
|
||||
import pytest
|
||||
|
||||
from collections import namedtuple
|
||||
from typing import Any, Dict, List
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
@ -10,9 +9,6 @@ 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:
|
||||
"""
|
||||
@ -78,12 +74,22 @@ def s3(configuration: Configuration) -> S3:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def s3_remote_objects() -> List[_s3_object]:
|
||||
def s3_remote_objects() -> List[MagicMock]:
|
||||
"""
|
||||
fixture for boto3 like S3 objects
|
||||
|
||||
Returns:
|
||||
List[_s3_object]: boto3 like S3 objects test instance
|
||||
List[MagicMock]: boto3 like S3 objects test instance
|
||||
"""
|
||||
delete_mock = MagicMock()
|
||||
return list(map(lambda item: _s3_object(f"x86_64/{item}", f"\"{item}\"", delete_mock), ["a", "b", "c"]))
|
||||
|
||||
result = []
|
||||
for item in ["a", "b", "c"]:
|
||||
s3_object = MagicMock()
|
||||
s3_object.key = f"x86_64/{item}"
|
||||
s3_object.e_tag = f"\"{item}\""
|
||||
s3_object.delete = delete_mock
|
||||
|
||||
result.append(s3_object)
|
||||
|
||||
return result
|
||||
|
Reference in New Issue
Block a user