Source code for ewoksmx.tests.tasks.test_icat_task

import logging
from contextlib import contextmanager
from typing import Optional
from unittest import mock

import requests

from ewoksmx.tasks.base_tasks.icat_task import IcatCallbackTask




def test_icat_callback_task(caplog):
    payload = {"dummy": None}
    task = _Dummy({"callback_url": None, "payload": payload})

    with _mock_requests_get([{"jobId": 999}]):
        with _mock_requests_put() as mock_put:
            with caplog.at_level(logging.INFO):
                task.execute()

            assert (
                f"Skip notification to ICAT (no callback URL): {payload}" in caplog.text
            )

            mock_put.assert_not_called()

            assert task.get_output_values() == {"icat_jobs": []}





def test_icat_callback_task_with_url(caplog):
    payload = {"dummy": None}
    task = _Dummy(inputs={"callback_url": "http://[MASKED]", "payload": payload})

    with _mock_requests_get([{"jobId": 999}]):
        with _mock_requests_put() as mock_put:
            with caplog.at_level(logging.INFO):
                task.execute()

            mock_put.assert_called_once_with("http://[MASKED]", json=payload)

            assert f"Send notification to ICAT: {payload}" in caplog.text

            assert task.get_output_values() == {"icat_jobs": [{"jobId": 999}]}



@contextmanager
def _mock_requests_put():
    with mock.patch.object(requests, "put", autospec=True) as mock_put:
        yield mock_put


@contextmanager
def _mock_requests_get(json_response):
    """Context manager to mock requests.put with a given JSON response and status code."""
    with mock.patch.object(requests, "get", autospec=True) as mock_get:
        # Define the mock response
        mock_response = mock.Mock()
        mock_response.json.return_value = json_response
        mock_response.status_code = 200
        mock_get.return_value = mock_response

        yield


class _Dummy(
    IcatCallbackTask,
    input_names=["callback_url", "payload"],
    output_names=["icat_jobs"],
):

    def run(self):
        self.notify_icat(self.inputs.payload)
        self.outputs.icat_jobs = self.get_icat_jobs()

    @property
    def _icat_callback_url(self) -> Optional[str]:
        return self.inputs.callback_url