mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-04-24 15:27:17 +00:00
rename watcher methods
This commit is contained in:
parent
8eb80731d4
commit
50775c3f0a
@ -30,13 +30,14 @@ class FilteredAccessLogger(AccessLogger):
|
||||
LOG_PATH_REGEX(re.Pattern): (class attribute) regex for logs uri
|
||||
"""
|
||||
|
||||
# official packages have only ``[A-Za-z0-9_.+-]`` regex
|
||||
LOG_PATH_REGEX = re.compile(r"^/api/v1/packages/[A-Za-z0-9_.+%-]+/logs$")
|
||||
LOG_PATH_REGEX = re.compile(r"^/api/v1/packages/[^/]+/logs$")
|
||||
# technically process id is uuid, but we might change it later
|
||||
PROCESS_PATH_REGEX = re.compile(r"^/api/v1/service/process/[^/]+$")
|
||||
|
||||
@staticmethod
|
||||
def is_logs_post(request: BaseRequest) -> bool:
|
||||
"""
|
||||
check if request looks lie logs posting
|
||||
check if request looks like logs posting
|
||||
|
||||
Args:
|
||||
request(BaseRequest): http reqeust descriptor
|
||||
@ -46,6 +47,19 @@ class FilteredAccessLogger(AccessLogger):
|
||||
"""
|
||||
return request.method == "POST" and FilteredAccessLogger.LOG_PATH_REGEX.match(request.path) is not None
|
||||
|
||||
@staticmethod
|
||||
def is_process_get(request: BaseRequest) -> bool:
|
||||
"""
|
||||
check if request looks like process status request
|
||||
|
||||
Args:
|
||||
request(BaseRequest): http reqeust descriptor
|
||||
|
||||
Returns:
|
||||
bool: True in case if request looks like process status request and False otherwise
|
||||
"""
|
||||
return request.method == "GET" and FilteredAccessLogger.PROCESS_PATH_REGEX.match(request.path) is not None
|
||||
|
||||
def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
|
||||
"""
|
||||
access log with enabled filter by request path
|
||||
@ -55,6 +69,7 @@ class FilteredAccessLogger(AccessLogger):
|
||||
response(StreamResponse): streaming response object
|
||||
time(float):
|
||||
"""
|
||||
if self.is_logs_post(request):
|
||||
if self.is_logs_post(request) \
|
||||
or self.is_process_get(request):
|
||||
return
|
||||
AccessLogger.log(self, request, response, time)
|
||||
|
||||
@ -71,7 +71,60 @@ class Watcher(LazyLogging):
|
||||
"""
|
||||
return list(self.known.values())
|
||||
|
||||
def get(self, package_base: str) -> tuple[Package, BuildStatus]:
|
||||
def load(self) -> None:
|
||||
"""
|
||||
load packages from local repository. In case if last status is known, it will use it
|
||||
"""
|
||||
for package in self.repository.packages():
|
||||
# get status of build or assign unknown
|
||||
if (current := self.known.get(package.base)) is not None:
|
||||
_, status = current
|
||||
else:
|
||||
status = BuildStatus()
|
||||
self.known[package.base] = (package, status)
|
||||
|
||||
for package, status in self.database.packages_get():
|
||||
if package.base in self.known:
|
||||
self.known[package.base] = (package, status)
|
||||
|
||||
def logs_get(self, package_base: str) -> str:
|
||||
"""
|
||||
extract logs for the package base
|
||||
|
||||
Args:
|
||||
package_base(str): package base
|
||||
|
||||
Returns:
|
||||
str: package logs
|
||||
"""
|
||||
return self.database.logs_get(package_base)
|
||||
|
||||
def logs_remove(self, package_base: str, current_process_id: int | None) -> None:
|
||||
"""
|
||||
remove package related logs
|
||||
|
||||
Args:
|
||||
package_base(str): package base
|
||||
current_process_id(int | None): current process id
|
||||
"""
|
||||
self.database.logs_remove(package_base, current_process_id)
|
||||
|
||||
def logs_update(self, log_record_id: LogRecordId, created: float, record: str) -> None:
|
||||
"""
|
||||
make new log record into database
|
||||
|
||||
Args:
|
||||
log_record_id(LogRecordId): log record id
|
||||
created(float): log created record
|
||||
record(str): log record
|
||||
"""
|
||||
if self._last_log_record_id != log_record_id:
|
||||
# there is new log record, so we remove old ones
|
||||
self.logs_remove(log_record_id.package_base, log_record_id.process_id)
|
||||
self._last_log_record_id = log_record_id
|
||||
self.database.logs_insert(log_record_id, created, record)
|
||||
|
||||
def package_get(self, package_base: str) -> tuple[Package, BuildStatus]:
|
||||
"""
|
||||
get current package base build status
|
||||
|
||||
@ -89,35 +142,7 @@ class Watcher(LazyLogging):
|
||||
except KeyError:
|
||||
raise UnknownPackageError(package_base)
|
||||
|
||||
def get_logs(self, package_base: str) -> str:
|
||||
"""
|
||||
extract logs for the package base
|
||||
|
||||
Args:
|
||||
package_base(str): package base
|
||||
|
||||
Returns:
|
||||
str: package logs
|
||||
"""
|
||||
return self.database.logs_get(package_base)
|
||||
|
||||
def load(self) -> None:
|
||||
"""
|
||||
load packages from local repository. In case if last status is known, it will use it
|
||||
"""
|
||||
for package in self.repository.packages():
|
||||
# get status of build or assign unknown
|
||||
if (current := self.known.get(package.base)) is not None:
|
||||
_, status = current
|
||||
else:
|
||||
status = BuildStatus()
|
||||
self.known[package.base] = (package, status)
|
||||
|
||||
for package, status in self.database.packages_get():
|
||||
if package.base in self.known:
|
||||
self.known[package.base] = (package, status)
|
||||
|
||||
def remove(self, package_base: str) -> None:
|
||||
def package_remove(self, package_base: str) -> None:
|
||||
"""
|
||||
remove package base from known list if any
|
||||
|
||||
@ -126,19 +151,9 @@ class Watcher(LazyLogging):
|
||||
"""
|
||||
self.known.pop(package_base, None)
|
||||
self.database.package_remove(package_base)
|
||||
self.remove_logs(package_base, None)
|
||||
self.logs_remove(package_base, None)
|
||||
|
||||
def remove_logs(self, package_base: str, current_process_id: int | None) -> None:
|
||||
"""
|
||||
remove package related logs
|
||||
|
||||
Args:
|
||||
package_base(str): package base
|
||||
current_process_id(int | None): current process id
|
||||
"""
|
||||
self.database.logs_remove(package_base, current_process_id)
|
||||
|
||||
def update(self, package_base: str, status: BuildStatusEnum, package: Package | None) -> None:
|
||||
def package_update(self, package_base: str, status: BuildStatusEnum, package: Package | None) -> None:
|
||||
"""
|
||||
update package status and description
|
||||
|
||||
@ -159,22 +174,7 @@ class Watcher(LazyLogging):
|
||||
self.known[package_base] = (package, full_status)
|
||||
self.database.package_update(package, full_status)
|
||||
|
||||
def update_logs(self, log_record_id: LogRecordId, created: float, record: str) -> None:
|
||||
"""
|
||||
make new log record into database
|
||||
|
||||
Args:
|
||||
log_record_id(LogRecordId): log record id
|
||||
created(float): log created record
|
||||
record(str): log record
|
||||
"""
|
||||
if self._last_log_record_id != log_record_id:
|
||||
# there is new log record, so we remove old ones
|
||||
self.remove_logs(log_record_id.package_base, log_record_id.process_id)
|
||||
self._last_log_record_id = log_record_id
|
||||
self.database.logs_insert(log_record_id, created, record)
|
||||
|
||||
def update_self(self, status: BuildStatusEnum) -> None:
|
||||
def status_update(self, status: BuildStatusEnum) -> None:
|
||||
"""
|
||||
update service status
|
||||
|
||||
|
||||
@ -63,7 +63,7 @@ class LogsView(BaseView):
|
||||
HTTPNoContent: on success response
|
||||
"""
|
||||
package_base = self.request.match_info["package"]
|
||||
self.service.remove_logs(package_base, None)
|
||||
self.service.logs_remove(package_base, None)
|
||||
|
||||
raise HTTPNoContent()
|
||||
|
||||
@ -95,10 +95,10 @@ class LogsView(BaseView):
|
||||
package_base = self.request.match_info["package"]
|
||||
|
||||
try:
|
||||
_, status = self.service.get(package_base)
|
||||
_, status = self.service.package_get(package_base)
|
||||
except UnknownPackageError:
|
||||
raise HTTPNotFound()
|
||||
logs = self.service.get_logs(package_base)
|
||||
logs = self.service.logs_get(package_base)
|
||||
|
||||
response = {
|
||||
"package_base": package_base,
|
||||
@ -141,6 +141,6 @@ class LogsView(BaseView):
|
||||
except Exception as e:
|
||||
raise HTTPBadRequest(reason=str(e))
|
||||
|
||||
self.service.update_logs(LogRecordId(package_base, process_id), created, record)
|
||||
self.service.logs_update(LogRecordId(package_base, process_id), created, record)
|
||||
|
||||
raise HTTPNoContent()
|
||||
|
||||
@ -64,7 +64,7 @@ class PackageView(BaseView):
|
||||
HTTPNoContent: on success response
|
||||
"""
|
||||
package_base = self.request.match_info["package"]
|
||||
self.service.remove(package_base)
|
||||
self.service.package_remove(package_base)
|
||||
|
||||
raise HTTPNoContent()
|
||||
|
||||
@ -96,7 +96,7 @@ class PackageView(BaseView):
|
||||
package_base = self.request.match_info["package"]
|
||||
|
||||
try:
|
||||
package, status = self.service.get(package_base)
|
||||
package, status = self.service.package_get(package_base)
|
||||
except UnknownPackageError:
|
||||
raise HTTPNotFound()
|
||||
|
||||
@ -142,7 +142,7 @@ class PackageView(BaseView):
|
||||
raise HTTPBadRequest(reason=str(e))
|
||||
|
||||
try:
|
||||
self.service.update(package_base, status, package)
|
||||
self.service.package_update(package_base, status, package)
|
||||
except UnknownPackageError:
|
||||
raise HTTPBadRequest(reason=f"Package {package_base} is unknown, but no package body set")
|
||||
|
||||
|
||||
@ -102,6 +102,6 @@ class StatusView(BaseView):
|
||||
except Exception as e:
|
||||
raise HTTPBadRequest(reason=str(e))
|
||||
|
||||
self.service.update_self(status)
|
||||
self.service.status_update(status)
|
||||
|
||||
raise HTTPNoContent()
|
||||
|
||||
@ -43,6 +43,33 @@ def test_is_logs_post() -> None:
|
||||
assert not FilteredAccessLogger.is_logs_post(request)
|
||||
|
||||
|
||||
def test_is_process_get() -> None:
|
||||
"""
|
||||
must correctly define if request belongs to process get
|
||||
"""
|
||||
request = MagicMock()
|
||||
|
||||
request.method = "GET"
|
||||
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00"
|
||||
assert FilteredAccessLogger.is_process_get(request)
|
||||
|
||||
request.method = "POST"
|
||||
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00"
|
||||
assert not FilteredAccessLogger.is_process_get(request)
|
||||
|
||||
request.method = "GET"
|
||||
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00/some/random/path"
|
||||
assert not FilteredAccessLogger.is_process_get(request)
|
||||
|
||||
request.method = "GET"
|
||||
request.path = "/api/v1/service/process"
|
||||
assert not FilteredAccessLogger.is_process_get(request)
|
||||
|
||||
request.method = "GET"
|
||||
request.path = "/api/v1/service/process/"
|
||||
assert not FilteredAccessLogger.is_process_get(request)
|
||||
|
||||
|
||||
def test_log(filtered_access_logger: FilteredAccessLogger, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must emit log record
|
||||
|
||||
@ -22,33 +22,6 @@ def test_force_no_report(configuration: Configuration, database: SQLite, mocker:
|
||||
load_mock.assert_called_once_with("x86_64", configuration, database, report=False)
|
||||
|
||||
|
||||
def test_get(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return package status
|
||||
"""
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
package, status = watcher.get(package_ahriman.base)
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must fail on unknown package
|
||||
"""
|
||||
with pytest.raises(UnknownPackageError):
|
||||
watcher.get(package_ahriman.base)
|
||||
|
||||
|
||||
def test_get_logs(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return package logs
|
||||
"""
|
||||
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_get")
|
||||
watcher.get_logs(package_ahriman.base)
|
||||
logs_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly load packages
|
||||
@ -77,109 +50,136 @@ def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFi
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
def test_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
def test_logs_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove package base
|
||||
must return package logs
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
|
||||
logs_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
assert not watcher.known
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
logs_mock.assert_called_once_with(package_ahriman.base, None)
|
||||
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_get")
|
||||
watcher.logs_get(package_ahriman.base)
|
||||
logs_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
def test_remove_logs(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
def test_logs_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove package logs
|
||||
"""
|
||||
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_remove")
|
||||
watcher.remove_logs(package_ahriman.base, 42)
|
||||
watcher.logs_remove(package_ahriman.base, 42)
|
||||
logs_mock.assert_called_once_with(package_ahriman.base, 42)
|
||||
|
||||
|
||||
def test_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on unknown base removal
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
def test_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status only for known package
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Success, None)
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
def test_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must fail on unknown package status update only
|
||||
"""
|
||||
with pytest.raises(UnknownPackageError):
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, None)
|
||||
|
||||
|
||||
def test_update_logs_new(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
def test_logs_update_new(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create package logs record for new package
|
||||
"""
|
||||
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
|
||||
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
|
||||
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
|
||||
|
||||
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.process_id)
|
||||
assert watcher._last_log_record_id != log_record_id
|
||||
|
||||
watcher.update_logs(log_record_id, 42.01, "log record")
|
||||
watcher.logs_update(log_record_id, 42.01, "log record")
|
||||
delete_mock.assert_called_once_with(package_ahriman.base, log_record_id.process_id)
|
||||
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
|
||||
|
||||
assert watcher._last_log_record_id == log_record_id
|
||||
|
||||
|
||||
def test_update_logs_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
def test_logs_update_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create package logs record for current package
|
||||
"""
|
||||
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
|
||||
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
|
||||
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
|
||||
|
||||
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.process_id)
|
||||
watcher._last_log_record_id = log_record_id
|
||||
|
||||
watcher.update_logs(log_record_id, 42.01, "log record")
|
||||
watcher.logs_update(log_record_id, 42.01, "log record")
|
||||
delete_mock.assert_not_called()
|
||||
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
|
||||
|
||||
|
||||
def test_update_self(watcher: Watcher) -> None:
|
||||
def test_package_get(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return package status
|
||||
"""
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
package, status = watcher.package_get(package_ahriman.base)
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_package_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must fail on unknown package
|
||||
"""
|
||||
with pytest.raises(UnknownPackageError):
|
||||
watcher.package_get(package_ahriman.base)
|
||||
|
||||
|
||||
def test_package_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove package base
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
|
||||
logs_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.package_remove(package_ahriman.base)
|
||||
assert not watcher.known
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
logs_mock.assert_called_once_with(package_ahriman.base, None)
|
||||
|
||||
|
||||
def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on unknown base removal
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
|
||||
|
||||
watcher.package_remove(package_ahriman.base)
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
def test_package_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
|
||||
|
||||
watcher.package_update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_package_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status only for known package
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.package_update(package_ahriman.base, BuildStatusEnum.Success, None)
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
def test_package_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must fail on unknown package status update only
|
||||
"""
|
||||
with pytest.raises(UnknownPackageError):
|
||||
watcher.package_update(package_ahriman.base, BuildStatusEnum.Unknown, None)
|
||||
|
||||
|
||||
def test_status_update(watcher: Watcher) -> None:
|
||||
"""
|
||||
must update service status
|
||||
"""
|
||||
watcher.update_self(BuildStatusEnum.Success)
|
||||
watcher.status_update(BuildStatusEnum.Success)
|
||||
assert watcher.status.status == BuildStatusEnum.Success
|
||||
|
||||
@ -75,7 +75,7 @@ async def test_post_exception_inside(client: TestClient, mocker: MockerFixture)
|
||||
exception handler must handle 500 errors
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.status_update", side_effect=Exception())
|
||||
response_schema = pytest.helpers.schema_response(StatusView.post, code=500)
|
||||
|
||||
response = await client.post("/api/v1/status", json=payload)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user