SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2025-01-31 05:06:41 +00:00
Code Issues Projects Releases Wiki Activity

style: fix black

Browse source
This commit is contained in:
dettlaff 2024-04-10 19:07:33 +04:00
parent d182e365d6
commit a4b1c058f7
2 changed files with 22 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
selfprivacy_api/jobs/nix_collect_garbage.py
View file

@ -11,11 +11,11 @@ class ShellException(Exception):
"""Shell-related errors""" """Shell-related errors"""
COMPLETED_WITH_ERROR = ( COMPLETED_WITH_ERROR = "Error occurred, please report this to the support chat."
"Error occurred, please report this to the support chat." RESULT_WAS_NOT_FOUND_ERROR = (
) "We are sorry, garbage collection result was not found. "
RESULT_WAS_NOT_FOUND_ERROR = "We are sorry, garbage collection result was not found. " \
"Something went wrong, please report this to the support chat." "Something went wrong, please report this to the support chat."
)
CLEAR_COMPLETED = "Garbage collection completed." CLEAR_COMPLETED = "Garbage collection completed."
@ -74,7 +74,6 @@ def process_stream(job: Job, stream: Iterable[bytes], total_dead_packages: int)
percent = int((completed_packages / total_dead_packages) * 100) percent = int((completed_packages / total_dead_packages) * 100)
if percent - prev_progress >= 5: if percent - prev_progress >= 5:
Jobs.update( Jobs.update(
job=job, job=job,
status=JobStatus.RUNNING, status=JobStatus.RUNNING,
@ -109,7 +108,6 @@ def calculate_and_clear_dead_paths(job: Job):
) )
if dead_packages == 0: if dead_packages == 0:
Jobs.update( Jobs.update(
job=job, job=job,
status=JobStatus.FINISHED, status=JobStatus.FINISHED,

24
tests/test_graphql/test_nix_collect_garbage.py
View file

@ -6,7 +6,12 @@ import pytest
from selfprivacy_api.utils.huey import huey from selfprivacy_api.utils.huey import huey
from selfprivacy_api.jobs import JobStatus, Jobs, Job from selfprivacy_api.jobs import JobStatus, Jobs, Job
from tests.test_graphql.common import get_data, assert_errorcode, assert_ok, assert_empty from tests.test_graphql.common import (
get_data,
assert_errorcode,
assert_ok,
assert_empty,
)
from selfprivacy_api.jobs.nix_collect_garbage import ( from selfprivacy_api.jobs.nix_collect_garbage import (
get_dead_packages, get_dead_packages,
@ -57,7 +62,11 @@ log_event = []
@pytest.fixture @pytest.fixture
def mock_delete_old_gens_and_return_dead_report(mocker): def mock_delete_old_gens_and_return_dead_report(mocker):
mock = mocker.patch("selfprivacy_api.jobs.nix_collect_garbage.delete_old_gens_and_return_dead_report", autospec=True, return_value=OUTPUT_PRINT_DEAD) mock = mocker.patch(
"selfprivacy_api.jobs.nix_collect_garbage.delete_old_gens_and_return_dead_report",
autospec=True,
return_value=OUTPUT_PRINT_DEAD,
)
return mock return mock
@ -74,7 +83,7 @@ def test_parse_line():
) )
output = parse_line(job, txt) output = parse_line(job, txt)
assert output.result == '425.51 MiB have been cleared' assert output.result == "425.51 MiB have been cleared"
assert output.status == JobStatus.FINISHED assert output.status == JobStatus.FINISHED
assert output.error is None assert output.error is None
@ -91,7 +100,6 @@ def test_parse_line_with_blank_line():
output = parse_line(job, txt) output = parse_line(job, txt)
def test_get_dead_packages(): def test_get_dead_packages():
assert get_dead_packages(OUTPUT_PRINT_DEAD) == (5, 20.0) assert get_dead_packages(OUTPUT_PRINT_DEAD) == (5, 20.0)
@ -127,7 +135,9 @@ mutation CollectGarbage {
""" """
def test_graphql_nix_collect_garbage(authorized_client, mock_delete_old_gens_and_return_dead_report): def test_graphql_nix_collect_garbage(
authorized_client, mock_delete_old_gens_and_return_dead_report
):
assert huey.immediate is True assert huey.immediate is True
response = authorized_client.post( response = authorized_client.post(
"/graphql", "/graphql",
@ -143,7 +153,9 @@ def test_graphql_nix_collect_garbage(authorized_client, mock_delete_old_gens_and
assert output["job"]["error"] is None assert output["job"]["error"] is None
def test_graphql_nix_collect_garbage_not_authorized_client(client, mock_delete_old_gens_and_return_dead_report): def test_graphql_nix_collect_garbage_not_authorized_client(
client, mock_delete_old_gens_and_return_dead_report
):
assert huey.immediate is True assert huey.immediate is True
response = client.post( response = client.post(
"/graphql", "/graphql",