from unittest.mock import ANY import pytest import requests_mock from ophyd_devices.utils.http_signal import HttpRestError, HttpRestSignal @pytest.fixture(autouse=True) def mock_server(): with requests_mock.Mocker() as m: mock_data = "data" def get_cb(request, context): nonlocal mock_data return mock_data def put_cb(request, context): nonlocal mock_data mock_data = request.text def put_req_valid(request): try: val = int(request.text) except: return False return -50 < val < 50 def put_can_fail_cb(request, context): context.reason = "" if put_req_valid(request) else "out of range" context.status_code = 202 if put_req_valid(request) else 422 m.get("http://test.psi.ch/get_data", text=get_cb) m.put("http://test.psi.ch/put_data", text=put_cb) m.get("http://test.psi.ch/bad_get_endpoint", status_code=404, reason="test not found") m.put("http://test.psi.ch/put_can_fail", text=put_can_fail_cb) yield requests_mock def test_signal_get(): sig = HttpRestSignal(name="get", get_uri="http://test.psi.ch/get_data") assert sig.read() == {"get": {"timestamp": ANY, "value": "data"}} def test_signal_put(): sig = HttpRestSignal( name="put_get", get_uri="http://test.psi.ch/get_data", put_uri="http://test.psi.ch/put_data" ) assert sig.read() == {"put_get": {"timestamp": ANY, "value": "data"}} sig.put("test_value") assert sig.read() == {"put_get": {"timestamp": ANY, "value": "test_value"}} def test_bad_signal_get(): sig = HttpRestSignal(name="get", get_uri="http://test.psi.ch/bad_get_endpoint") with pytest.raises(HttpRestError) as e: sig.read() assert e.match("test not found") def test_bad_signal_put(): sig = HttpRestSignal(name="get", get_uri="http://test.psi.ch/put_can_fail") sig.put("20") with pytest.raises(HttpRestError) as e: sig.put("50") assert e.match("Could not PUT 50") assert e.match("Code: 422. Reason: out of range.")