2022-08-25 17:03:56 +00:00
|
|
|
# pylint: disable=redefined-outer-name
|
|
|
|
# pylint: disable=unused-argument
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
from selfprivacy_api.jobs import Jobs, JobStatus
|
2022-11-23 16:29:50 +00:00
|
|
|
import selfprivacy_api.jobs as jobsmodule
|
2022-08-25 17:03:56 +00:00
|
|
|
|
|
|
|
|
|
|
|
def test_jobs(authorized_client, jobs_file, shared_datadir):
|
|
|
|
jobs = Jobs()
|
2022-11-16 13:54:54 +00:00
|
|
|
jobs.reset()
|
2022-08-25 17:03:56 +00:00
|
|
|
assert jobs.get_jobs() == []
|
|
|
|
|
|
|
|
test_job = jobs.add(
|
|
|
|
type_id="test",
|
|
|
|
name="Test job",
|
|
|
|
description="This is a test job.",
|
|
|
|
status=JobStatus.CREATED,
|
|
|
|
status_text="Status text",
|
|
|
|
progress=0,
|
|
|
|
)
|
|
|
|
|
|
|
|
assert jobs.get_jobs() == [test_job]
|
|
|
|
|
|
|
|
jobs.update(
|
|
|
|
job=test_job,
|
|
|
|
status=JobStatus.RUNNING,
|
|
|
|
status_text="Status text",
|
|
|
|
progress=50,
|
|
|
|
)
|
|
|
|
|
|
|
|
assert jobs.get_jobs() == [test_job]
|
|
|
|
|
2022-11-23 16:29:50 +00:00
|
|
|
backup = jobsmodule.JOB_EXPIRATION_SECONDS
|
|
|
|
jobsmodule.JOB_EXPIRATION_SECONDS = 0
|
|
|
|
|
|
|
|
jobs.update(
|
|
|
|
job=test_job,
|
|
|
|
status=JobStatus.FINISHED,
|
|
|
|
status_text="Yaaay!",
|
|
|
|
progress=100,
|
|
|
|
)
|
|
|
|
|
|
|
|
assert jobs.get_jobs() == []
|
|
|
|
jobsmodule.JOB_EXPIRATION_SECONDS = backup
|
|
|
|
|
2022-08-25 17:03:56 +00:00
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def mock_subprocess_run(mocker):
|
|
|
|
mock = mocker.patch("subprocess.run", autospec=True)
|
|
|
|
return mock
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def mock_shutil_move(mocker):
|
|
|
|
mock = mocker.patch("shutil.move", autospec=True)
|
|
|
|
return mock
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def mock_shutil_chown(mocker):
|
|
|
|
mock = mocker.patch("shutil.chown", autospec=True)
|
|
|
|
return mock
|