Repository : http://git.fedorahosted.org/cgit/copr.git On branch : master commit b37a5f84b95c9aa69ac87fd4ba2eade4e480c1af Author: Valentin Gologuzov <vgologuz(a)redhat.com&gt; Date: Fri Nov 28 13:47:01 2014 +0100 [backend] unittest for frontendclient backend/backend/frontend.py | 28 +++----- backend/tests/test_frontend.py | 136 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 17 deletions(-) diff --git a/backend/backend/frontend.py b/backend/backend/frontend.py index c34857a..51551f4 100644 --- a/backend/backend/frontend.py +++ b/backend/backend/frontend.py @@ -1,5 +1,5 @@ import json -import requests +from requests import post, RequestException import time @@ -27,13 +27,12 @@ class FrontendClient(object): self.msg = None try: - response = requests.post(url, data=json.dumps(data), auth=auth, - headers=headers) - if response.status_code != 200: + response = post(url, data=json.dumps(data), auth=auth, headers=headers) + if response.status_code >= 400: self.msg = "Failed to submit to frontend: {0}: {1}".format( response.status_code, response.text) - raise requests.RequestException(self.msg) - except requests.RequestException as e: + raise RequestException(self.msg) + except RequestException as e: self.msg = "Post request failed: {0}".format(e) raise return response @@ -42,18 +41,13 @@ class FrontendClient(object): """ Make a request max_repeats-time to the frontend """ - repeats = 0 - while repeats <= max_repeats: + for i in range(max_repeats): try: - response = self._post_to_frontend(data, url_path) - break - except requests.RequestException: - - if repeats == max_repeats: - raise - repeats += 1 + return self._post_to_frontend(data, url_path) + except RequestException: time.sleep(5) - return response + else: + raise RequestException("Failed to post to frontend for {} times".format(max_repeats)) def update(self, data): """ @@ -70,5 +64,5 @@ class FrontendClient(object): data = {"build_id": build_id, "chroot": chroot_name} response = self._post_to_frontend_repeatedly(data, "starting_build") if "can_start" not in response.json(): - raise requests.RequestException("Bad respond from the frontend") + raise RequestException("Bad respond from the frontend") return response.json()["can_start"] diff --git a/backend/tests/test_frontend.py b/backend/tests/test_frontend.py new file mode 100644 index 0000000..c18d1f6 --- /dev/null +++ b/backend/tests/test_frontend.py @@ -0,0 +1,136 @@ +# coding: utf-8 +import copy + +from collections import defaultdict +import json +import multiprocessing +from pprint import pprint +from bunch import Bunch +from requests import RequestException +from backend.exceptions import BuilderError, BuilderTimeOutError, MockRemoteError, CoprSignError + +import tempfile +import shutil +import os + +import six +from backend.frontend import FrontendClient + +if six.PY3: + from unittest import mock + from unittest.mock import patch, MagicMock +else: + import mock + from mock import patch, MagicMock + +import pytest + + +(a)pytest.yield_fixture +def post_req(): + with mock.patch("backend.frontend.post") as obj: + yield obj + + +(a)pytest.yield_fixture +def mc_time(): + with mock.patch("backend.frontend.time") as obj: + yield obj + +class TestFrontendClient(object): + + def setup_method(self, method): + self.opts = Bunch( + frontend_url="http://example.com/";, + frontend_auth="12345678", + ) + self.events = multiprocessing.Queue() + self.fc = FrontendClient(self.opts, self.events) + + self.data = { + "foo": "bar", + "bar": [1, 3, 5], + } + self.url_path = "sub_path" + + self.build_id = 12345 + self.chroot_name = "fedora-20-x86_64" + + @pytest.fixture + def mask_post_to_fe(self): + self.ptf = MagicMock() + self.fc._post_to_frontend = self.ptf + + def test_post_to_frontend(self, post_req): + post_req.return_value.status_code = 200 + self.fc._post_to_frontend(self.data, self.url_path) + + assert post_req.called + + def test_post_to_frontend_not_200(self, post_req): + post_req.return_value.status_code = 501 + with pytest.raises(RequestException): + self.fc._post_to_frontend(self.data, self.url_path) + + assert post_req.called + + def test_post_to_frontend_post_error(self, post_req): + post_req.side_effect = RequestException() + with pytest.raises(RequestException): + self.fc._post_to_frontend(self.data, self.url_path) + + assert post_req.called + + def test_post_to_frontend_repeated_first_try_ok(self, mask_post_to_fe, mc_time): + response = "ok\n" + self.ptf.return_value = response + + assert self.fc._post_to_frontend_repeatedly(self.data, self.url_path) == response + assert not mc_time.sleep.called + + def test_post_to_frontend_repeated_second_try_ok(self, mask_post_to_fe, mc_time): + response = "ok\n" + self.ptf.side_effect = [ + RequestException(), + response, + ] + + assert self.fc._post_to_frontend_repeatedly(self.data, self.url_path) == response + assert mc_time.sleep.called + + def test_post_to_frontend_repeated_all_attempts_failed(self, mask_post_to_fe, mc_time): + self.ptf.side_effect = RequestException() + + with pytest.raises(RequestException): + self.fc._post_to_frontend_repeatedly(self.data, self.url_path) + + assert mc_time.sleep.called + + def test_update(self): + ptfr = MagicMock() + self.fc._post_to_frontend_repeatedly = ptfr + self.fc.update(self.data) + assert ptfr.call_args == mock.call(self.data, "update") + + def test_starting_build(self): + ptfr = MagicMock() + self.fc._post_to_frontend_repeatedly = ptfr + for val in [True, False]: + ptfr.return_value.json.return_value = {"can_start": val} + + assert self.fc.starting_build(self.build_id, self.chroot_name) == val + + def test_starting_build_err(self): + ptfr = MagicMock() + self.fc._post_to_frontend_repeatedly = ptfr + + with pytest.raises(RequestException): + self.fc.starting_build(self.build_id, self.chroot_name) + + def test_starting_build_err_2(self): + ptfr = MagicMock() + self.fc._post_to_frontend_repeatedly = ptfr + ptfr.return_value.json.return_value = {} + + with pytest.raises(RequestException): + self.fc.starting_build(self.build_id, self.chroot_name)