Skip to content
Snippets Groups Projects

Resolve "2nd cronjob to scan for updated Scholix links for all existing literature publications (once per month, over the weekend?)"

Files
53
@@ -10,10 +10,11 @@ from requests.exceptions import ConnectTimeout
from requests.exceptions import ReadTimeout
from requests_mock import Mocker
from toolbox.data_enricher.fair_meter.fuji_scorer import FUJI_API
from toolbox.data_enricher.fair_meter.fuji_scorer import FujiScorer
from toolbox.data_enricher.fair_meter.fuji_scorer import RETRY_WAITING_TIME
from toolbox.type_definitions import DataMetadata
from toolbox.fair_meter.fuji_scorer.fuji_conf import EXTRA_TIMEOUT
from toolbox.fair_meter.fuji_scorer.fuji_conf import FUJI_API
from toolbox.fair_meter.fuji_scorer.fuji_conf import INITIAL_TIMEOUT
from toolbox.fair_meter.fuji_scorer.fuji_conf import RETRY_WAITING_TIME
from toolbox.fair_meter.fuji_scorer.fuji_scorer import get_fuji_score
class TestFujiScorer(unittest.TestCase):
@@ -25,7 +26,6 @@ class TestFujiScorer(unittest.TestCase):
with open(self.path, encoding='utf-8') as file:
self.response = js.load(file)
self.scorer = FujiScorer()
self.data_pid = "10.5281/zenodo.259703"
self.data_pid_type = "doi"
@@ -34,9 +34,10 @@ class TestFujiScorer(unittest.TestCase):
with Mocker() as mock:
mock.post(FUJI_API, json=self.response)
data = self.scorer.get_fuji_score(self.data_pid, self.data_pid_type)
data = get_fuji_score(self.data_pid, self.data_pid_type)
# asserts method gives correct answers
assert data is not None
self.assertEqual(data["pid"], self.data_pid)
self.assertEqual(data["pid_type"], self.data_pid_type)
self.assertEqual(
@@ -66,10 +67,13 @@ class TestFujiScorer(unittest.TestCase):
mock.post(FUJI_API, answers)
# calls get_fuji_score function
answer = self.scorer.get_fuji_score(self.data_pid, self.data_pid_type)
# asserts Function was called 5 times, with the right sleep time on the last call and gets an answer
answer = get_fuji_score(self.data_pid, self.data_pid_type)
# function was called 5 times, with the right sleep time on the last call and gets an answer
self.assertEqual(mock.call_count, 5)
self.assertEqual(mock.request_history[4].timeout, 2200)
self.assertEqual(
mock.request_history[4].timeout,
INITIAL_TIMEOUT + 4 * EXTRA_TIMEOUT,
)
sleep.assert_called_with(4 * RETRY_WAITING_TIME)
self.assertEqual(answer["pid"], self.data_pid)
@@ -79,42 +83,14 @@ class TestFujiScorer(unittest.TestCase):
# stops normal mock object and instead uses a mock object that gives back Exceptions
with Mocker() as mock:
mock.post(FUJI_API, exc=ReadTimeout)
answer = self.scorer.get_fuji_score(self.data_pid, self.data_pid_type)
answer = get_fuji_score(self.data_pid, self.data_pid_type)
# asserts Function was called 5 times, with the right timeout on the last call and gets an answer
# function was called 10 times, with the right timeout on the last call and gets no answer
self.assertEqual(mock.call_count, 11)
self.assertEqual(mock.request_history[10].timeout, 4000)
self.assertEqual(
mock.request_history[10].timeout,
INITIAL_TIMEOUT + 10 * EXTRA_TIMEOUT,
)
sleep.assert_called_with(10 * RETRY_WAITING_TIME)
self.assertIsNone(answer)
def test_add_data(self):
"""tests add_fuji_scores function"""
# generate entries for data_pubs in add_fuji_scores
entry1 = DataMetadata()
entry1.DataPID = "10.5281/zenodo.259703"
entry1.DataPIDType = "doi"
entry2 = DataMetadata()
entry2.DataPID = "test"
entry2.DataPIDType = "doi"
# fill data_pubs
data_pubs = [entry1, entry2]
with Mocker() as mock:
mock.post(FUJI_API, json=self.response)
answer = self.scorer.add_fuji_scores(data_pubs)
# asserts mock was called for each data entry
self.assertEqual(mock.call_count, len(data_pubs))
data_pubs[0].FAIRScores = self.scorer.get_fuji_score(
data_pubs[0].DataPID, data_pubs[0].DataPIDType
)
data_pubs[1].FAIRScores = self.scorer.get_fuji_score(
data_pubs[1].DataPID, data_pubs[1].DataPIDType
)
# asserts add_fuji_scores gives back the right answer
self.assertEqual(answer, data_pubs)
Loading