mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-29 15:31:28 +00:00
Fix jobs file
This commit is contained in:
parent
64425ac443
commit
039dd2f80e
|
@ -89,7 +89,7 @@ class Jobs:
|
|||
Reset the jobs list.
|
||||
"""
|
||||
with WriteUserData(UserDataFiles.JOBS) as user_data:
|
||||
user_data = []
|
||||
user_data["jobs"] = []
|
||||
|
||||
def add(
|
||||
self,
|
||||
|
@ -116,10 +116,11 @@ class Jobs:
|
|||
)
|
||||
with WriteUserData(UserDataFiles.JOBS) as user_data:
|
||||
try:
|
||||
user_data.append(json.loads(job.json()))
|
||||
if "jobs" not in user_data:
|
||||
user_data["jobs"] = []
|
||||
user_data["jobs"].append(json.loads(job.json()))
|
||||
except json.decoder.JSONDecodeError:
|
||||
user_data = []
|
||||
user_data.append(json.loads(job.json()))
|
||||
user_data["jobs"] = [json.loads(job.json())]
|
||||
return job
|
||||
|
||||
def remove(self, job: Job) -> None:
|
||||
|
@ -127,7 +128,9 @@ class Jobs:
|
|||
Remove a job from the jobs list.
|
||||
"""
|
||||
with WriteUserData(UserDataFiles.JOBS) as user_data:
|
||||
user_data = [x for x in user_data if x["uid"] != job.uid.urn]
|
||||
if "jobs" not in user_data:
|
||||
user_data["jobs"] = []
|
||||
user_data["jobs"] = [x for x in user_data["jobs"] if x["uid"] != str(job.uid)]
|
||||
|
||||
def update(
|
||||
self,
|
||||
|
@ -159,8 +162,10 @@ class Jobs:
|
|||
job.finished_at = datetime.datetime.now()
|
||||
|
||||
with WriteUserData(UserDataFiles.JOBS) as user_data:
|
||||
user_data = [x for x in user_data if x["uid"] != job.uid.urn]
|
||||
user_data.append(json.loads(job.json()))
|
||||
if "jobs" not in user_data:
|
||||
user_data["jobs"] = []
|
||||
user_data["jobs"] = [x for x in user_data["jobs"] if x["uid"] != str(job.uid)]
|
||||
user_data["jobs"].append(json.loads(job.json()))
|
||||
|
||||
return job
|
||||
|
||||
|
@ -169,7 +174,9 @@ class Jobs:
|
|||
Get a job from the jobs list.
|
||||
"""
|
||||
with ReadUserData(UserDataFiles.JOBS) as user_data:
|
||||
for job in user_data:
|
||||
if "jobs" not in user_data:
|
||||
user_data["jobs"] = []
|
||||
for job in user_data["jobs"]:
|
||||
if job["uid"] == id:
|
||||
return Job(**job)
|
||||
return None
|
||||
|
@ -180,6 +187,8 @@ class Jobs:
|
|||
"""
|
||||
with ReadUserData(UserDataFiles.JOBS) as user_data:
|
||||
try:
|
||||
return [Job(**job) for job in user_data]
|
||||
if "jobs" not in user_data:
|
||||
user_data["jobs"] = []
|
||||
return [Job(**job) for job in user_data["jobs"]]
|
||||
except json.decoder.JSONDecodeError:
|
||||
return []
|
||||
|
|
|
@ -41,7 +41,7 @@ class WriteUserData(object):
|
|||
# Make sure file exists
|
||||
if not os.path.isfile(JOBS_FILE):
|
||||
with open(JOBS_FILE, "w", encoding="utf-8") as jobs_file:
|
||||
jobs_file.write("[]")
|
||||
jobs_file.write("{}")
|
||||
self.userdata_file = open(JOBS_FILE, "r+", encoding="utf-8")
|
||||
else:
|
||||
raise ValueError("Unknown file type")
|
||||
|
@ -72,7 +72,7 @@ class ReadUserData(object):
|
|||
# Make sure file exists
|
||||
if not os.path.isfile(JOBS_FILE):
|
||||
with open(JOBS_FILE, "w", encoding="utf-8") as jobs_file:
|
||||
jobs_file.write("[]")
|
||||
jobs_file.write("{}")
|
||||
self.userdata_file = open(JOBS_FILE, "r", encoding="utf-8")
|
||||
else:
|
||||
raise ValueError("Unknown file type")
|
||||
|
|
|
@ -13,6 +13,13 @@ def tokens_file(mocker, shared_datadir):
|
|||
)
|
||||
return mock
|
||||
|
||||
@pytest.fixture
|
||||
def jobs_file(mocker, shared_datadir):
|
||||
"""Mock tokens file."""
|
||||
mock = mocker.patch(
|
||||
"selfprivacy_api.utils.JOBS_FILE", shared_datadir / "jobs.json"
|
||||
)
|
||||
return mock
|
||||
|
||||
@pytest.fixture
|
||||
def huey_database(mocker, shared_datadir):
|
||||
|
@ -24,14 +31,14 @@ def huey_database(mocker, shared_datadir):
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tokens_file, huey_database):
|
||||
def client(tokens_file, huey_database, jobs_file):
|
||||
from selfprivacy_api.app import app
|
||||
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authorized_client(tokens_file, huey_database):
|
||||
def authorized_client(tokens_file, huey_database, jobs_file):
|
||||
"""Authorized test client fixture."""
|
||||
from selfprivacy_api.app import app
|
||||
|
||||
|
@ -41,7 +48,7 @@ def authorized_client(tokens_file, huey_database):
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def wrong_auth_client(tokens_file, huey_database):
|
||||
def wrong_auth_client(tokens_file, huey_database, jobs_file):
|
||||
"""Wrong token test client fixture."""
|
||||
from selfprivacy_api.app import app
|
||||
|
||||
|
|
1
tests/data/jobs.json
Normal file
1
tests/data/jobs.json
Normal file
|
@ -0,0 +1 @@
|
|||
{}
|
30
tests/test_jobs.py
Normal file
30
tests/test_jobs.py
Normal file
|
@ -0,0 +1,30 @@
|
|||
# pylint: disable=redefined-outer-name
|
||||
# pylint: disable=unused-argument
|
||||
import json
|
||||
import pytest
|
||||
|
||||
from selfprivacy_api.utils import WriteUserData, ReadUserData
|
||||
from selfprivacy_api.jobs import Jobs, JobStatus
|
||||
|
||||
def test_jobs(jobs_file, shared_datadir):
|
||||
jobs = Jobs()
|
||||
assert jobs.get_jobs() == []
|
||||
|
||||
test_job = jobs.add(
|
||||
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]
|
Loading…
Reference in a new issue