Merge pull request #1244 from jooola/api_client_pytest
Migrate api_clients tests to pytest
This commit is contained in:
commit
8bbd0d608c
|
@ -4,6 +4,9 @@ echo "::group::Install Python apps"
|
|||
pip3 install nose mock
|
||||
|
||||
for app in $(ls python_apps); do
|
||||
if [[ -f "python_apps/$app/requirements-dev.txt" ]]; then
|
||||
pip3 install -r "python_apps/$app/requirements-dev.txt"
|
||||
fi
|
||||
pip3 install -e python_apps/$app
|
||||
done
|
||||
echo "::endgroup::"
|
||||
|
|
|
@ -4,21 +4,19 @@ failed='f'
|
|||
# Starting at repo root
|
||||
|
||||
echo "::group::Airtime Analyzer"
|
||||
cd python_apps/airtime_analyzer
|
||||
pushd python_apps/airtime_analyzer
|
||||
if ! nosetests . -x; then
|
||||
failed='t'
|
||||
fi
|
||||
popd
|
||||
echo "::endgroup::"
|
||||
|
||||
echo "::group::API Client"
|
||||
cd ../api_clients
|
||||
if ! nosetests . -x; then
|
||||
if ! make -C python_apps/api_clients test; then
|
||||
failed='t'
|
||||
fi
|
||||
echo "::endgroup::"
|
||||
|
||||
# Reset to repo root
|
||||
cd ../..
|
||||
if [[ "$failed" = "t" ]]; then
|
||||
echo "Python tests failed"
|
||||
exit 1
|
||||
|
|
|
@ -0,0 +1,16 @@
|
|||
.PHONY: lint test
|
||||
|
||||
SHELL := bash
|
||||
CPU_CORES := $(shell nproc)
|
||||
|
||||
MODULE_APP := api_clients
|
||||
MODULE_TESTS := tests
|
||||
|
||||
lint:
|
||||
pylint ${MODULE_APP}
|
||||
pylint ${MODULE_TESTS}
|
||||
|
||||
test:
|
||||
pytest -n ${CPU_CORES} --color=yes -v --cov=${MODULE_APP} ${MODULE_TESTS}
|
||||
|
||||
all: lint test
|
|
@ -0,0 +1,5 @@
|
|||
mock
|
||||
pylint
|
||||
pytest
|
||||
pytest-cov
|
||||
pytest-xdist
|
|
@ -1,33 +1,32 @@
|
|||
import unittest
|
||||
|
||||
import pytest
|
||||
from api_clients.utils import ApcUrl, IncompleteUrl, UrlBadParam
|
||||
|
||||
|
||||
class TestApcUrl(unittest.TestCase):
|
||||
def test_init(self):
|
||||
url = "/testing"
|
||||
u = ApcUrl(url)
|
||||
self.assertEqual(u.base_url, url)
|
||||
@pytest.mark.parametrize(
|
||||
"url, params, expected",
|
||||
[
|
||||
("one/two/three", {}, "one/two/three"),
|
||||
("/testing/{key}", {"key": "aaa"}, "/testing/aaa"),
|
||||
(
|
||||
"/more/{key_a}/{key_b}/testing",
|
||||
{"key_a": "aaa", "key_b": "bbb"},
|
||||
"/more/aaa/bbb/testing",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_apc_url(url: str, params: dict, expected: str):
|
||||
found = ApcUrl(url)
|
||||
assert found.base_url == url
|
||||
assert found.params(**params).url() == expected
|
||||
|
||||
def test_params_1(self):
|
||||
u = ApcUrl("/testing/{key}")
|
||||
self.assertEqual(u.params(key="val").url(), "/testing/val")
|
||||
|
||||
def test_params_2(self):
|
||||
u = ApcUrl("/testing/{key}/{api}/more_testing")
|
||||
full_url = u.params(key="AAA", api="BBB").url()
|
||||
self.assertEqual(full_url, "/testing/AAA/BBB/more_testing")
|
||||
def test_apc_url_bad_param():
|
||||
url = ApcUrl("/testing/{key}")
|
||||
with pytest.raises(UrlBadParam):
|
||||
url.params(bad_key="testing")
|
||||
|
||||
def test_params_ex(self):
|
||||
u = ApcUrl("/testing/{key}")
|
||||
with self.assertRaises(UrlBadParam):
|
||||
u.params(bad_key="testing")
|
||||
|
||||
def test_url(self):
|
||||
u = "one/two/three"
|
||||
self.assertEqual(ApcUrl(u).url(), u)
|
||||
|
||||
def test_url_ex(self):
|
||||
u = ApcUrl("/{one}/{two}/three").params(two="testing")
|
||||
with self.assertRaises(IncompleteUrl):
|
||||
u.url()
|
||||
def test_apc_url_incomplete():
|
||||
url = ApcUrl("/{one}/{two}/three").params(two="testing")
|
||||
with pytest.raises(IncompleteUrl):
|
||||
url.url()
|
||||
|
|
|
@ -1,46 +1,33 @@
|
|||
import json
|
||||
import unittest
|
||||
|
||||
from api_clients.utils import ApcUrl, ApiRequest
|
||||
from mock import MagicMock, patch
|
||||
|
||||
|
||||
class ResponseInfo:
|
||||
@property
|
||||
def headers(self):
|
||||
return {"content-type": "application/json"}
|
||||
|
||||
def json(self):
|
||||
return {"ok", "ok"}
|
||||
def test_api_request_init():
|
||||
u = ApiRequest("request_name", ApcUrl("/test/ing"))
|
||||
assert u.name == "request_name"
|
||||
|
||||
|
||||
class TestApiRequest(unittest.TestCase):
|
||||
def test_init(self):
|
||||
u = ApiRequest("request_name", ApcUrl("/test/ing"))
|
||||
self.assertEqual(u.name, "request_name")
|
||||
def test_api_request_call_json():
|
||||
return_value = {"ok": "ok"}
|
||||
|
||||
def test_call_json(self):
|
||||
ret = {"ok": "ok"}
|
||||
read = MagicMock()
|
||||
read.headers = {"content-type": "application/json"}
|
||||
read.json = MagicMock(return_value=ret)
|
||||
u = "http://localhost/testing"
|
||||
with patch("requests.get") as mock_method:
|
||||
mock_method.return_value = read
|
||||
request = ApiRequest("mm", ApcUrl(u))()
|
||||
self.assertEqual(request, ret)
|
||||
read = MagicMock()
|
||||
read.headers = {"content-type": "application/json"}
|
||||
read.json = MagicMock(return_value=return_value)
|
||||
|
||||
def test_call_html(self):
|
||||
ret = "<html><head></head><body></body></html>"
|
||||
read = MagicMock()
|
||||
read.headers = {"content-type": "application/html"}
|
||||
read.text = MagicMock(return_value=ret)
|
||||
u = "http://localhost/testing"
|
||||
with patch("requests.get") as mock_method:
|
||||
mock_method.return_value = read
|
||||
request = ApiRequest("mm", ApcUrl(u))()
|
||||
self.assertEqual(request.text(), ret)
|
||||
with patch("requests.get") as mock_method:
|
||||
mock_method.return_value = read
|
||||
request = ApiRequest("mm", ApcUrl("http://localhost/testing"))()
|
||||
assert request == return_value
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
def test_api_request_call_html():
|
||||
return_value = "<html><head></head><body></body></html>"
|
||||
|
||||
read = MagicMock()
|
||||
read.headers = {"content-type": "application/html"}
|
||||
read.text = MagicMock(return_value=return_value)
|
||||
|
||||
with patch("requests.get") as mock_method:
|
||||
mock_method.return_value = read
|
||||
request = ApiRequest("mm", ApcUrl("http://localhost/testing"))()
|
||||
assert request.text() == return_value
|
||||
|
|
|
@ -1,39 +1,33 @@
|
|||
import json
|
||||
import unittest
|
||||
|
||||
import pytest
|
||||
from api_clients.utils import RequestProvider
|
||||
from api_clients.version1 import api_config
|
||||
from configobj import ConfigObj
|
||||
from mock import MagicMock, patch
|
||||
|
||||
|
||||
class TestRequestProvider(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.cfg = api_config
|
||||
self.cfg["general"] = {}
|
||||
self.cfg["general"]["base_dir"] = "/test"
|
||||
self.cfg["general"]["base_port"] = 80
|
||||
self.cfg["general"]["base_url"] = "localhost"
|
||||
self.cfg["general"]["api_key"] = "TEST_KEY"
|
||||
self.cfg["api_base"] = "api"
|
||||
|
||||
def test_test(self):
|
||||
self.assertTrue("general" in self.cfg)
|
||||
|
||||
def test_init(self):
|
||||
rp = RequestProvider(self.cfg, {})
|
||||
self.assertEqual(len(rp.available_requests()), 0)
|
||||
|
||||
def test_contains(self):
|
||||
methods = {
|
||||
"upload_recorded": "/1/",
|
||||
"update_media_url": "/2/",
|
||||
"list_all_db_files": "/3/",
|
||||
}
|
||||
rp = RequestProvider(self.cfg, methods)
|
||||
for meth in methods:
|
||||
self.assertTrue(meth in rp.requests)
|
||||
@pytest.fixture()
|
||||
def config():
|
||||
return {
|
||||
**api_config,
|
||||
"general": {
|
||||
"base_dir": "/test",
|
||||
"base_port": 80,
|
||||
"base_url": "localhost",
|
||||
"api_key": "TEST_KEY",
|
||||
},
|
||||
"api_base": "api",
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
def test_request_provider_init(config):
|
||||
request_provider = RequestProvider(config, {})
|
||||
assert len(request_provider.available_requests()) == 0
|
||||
|
||||
|
||||
def test_request_provider_contains(config):
|
||||
endpoints = {
|
||||
"upload_recorded": "/1/",
|
||||
"update_media_url": "/2/",
|
||||
"list_all_db_files": "/3/",
|
||||
}
|
||||
request_provider = RequestProvider(config, endpoints)
|
||||
for endpoint in endpoints:
|
||||
assert endpoint in request_provider.requests
|
||||
|
|
|
@ -1,102 +1,57 @@
|
|||
import configparser
|
||||
import datetime
|
||||
import unittest
|
||||
from configparser import ConfigParser
|
||||
|
||||
import pytest
|
||||
from api_clients import utils
|
||||
|
||||
|
||||
def get_force_ssl(value, useConfigParser):
|
||||
config = {}
|
||||
if useConfigParser:
|
||||
config = configparser.ConfigParser()
|
||||
config["general"] = {
|
||||
"base_port": 80,
|
||||
"force_ssl": value,
|
||||
}
|
||||
return utils.get_protocol(config)
|
||||
def test_time_in_seconds():
|
||||
time = datetime.time(hour=0, minute=3, second=34, microsecond=649600)
|
||||
assert abs(utils.time_in_seconds(time) - 214.65) < 0.009
|
||||
|
||||
|
||||
class TestTime(unittest.TestCase):
|
||||
def test_time_in_seconds(self):
|
||||
time = datetime.time(hour=0, minute=3, second=34, microsecond=649600)
|
||||
self.assertTrue(abs(utils.time_in_seconds(time) - 214.65) < 0.009)
|
||||
|
||||
def test_time_in_milliseconds(self):
|
||||
time = datetime.time(hour=0, minute=0, second=0, microsecond=500000)
|
||||
self.assertEqual(utils.time_in_milliseconds(time), 500)
|
||||
def test_time_in_milliseconds():
|
||||
time = datetime.time(hour=0, minute=0, second=0, microsecond=500000)
|
||||
assert utils.time_in_milliseconds(time) == 500
|
||||
|
||||
|
||||
class TestGetProtocol(unittest.TestCase):
|
||||
def test_dict_config_empty_http(self):
|
||||
config = {"general": {}}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "http")
|
||||
@pytest.mark.parametrize(
|
||||
"payload, expected",
|
||||
[({}, "http"), ({"base_port": 80}, "http"), ({"base_port": 443}, "https")],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"use_config",
|
||||
[False, True],
|
||||
)
|
||||
def test_get_protocol(payload, use_config, expected):
|
||||
config = ConfigParser() if use_config else {}
|
||||
config["general"] = {**payload}
|
||||
|
||||
def test_dict_config_http(self):
|
||||
config = {
|
||||
"general": {
|
||||
"base_port": 80,
|
||||
},
|
||||
}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "http")
|
||||
|
||||
def test_dict_config_https(self):
|
||||
config = {
|
||||
"general": {
|
||||
"base_port": 443,
|
||||
},
|
||||
}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "https")
|
||||
|
||||
def test_dict_config_force_https(self):
|
||||
postive_values = ["yes", "Yes", "True", "true", True]
|
||||
negative_values = ["no", "No", "False", "false", False]
|
||||
for value in postive_values:
|
||||
self.assertEqual(get_force_ssl(value, False), "https")
|
||||
for value in negative_values:
|
||||
self.assertEqual(get_force_ssl(value, False), "http")
|
||||
|
||||
def test_configparser_config_empty_http(self):
|
||||
config = configparser.ConfigParser()
|
||||
config["general"] = {}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "http")
|
||||
|
||||
def test_configparser_config_http(self):
|
||||
config = configparser.ConfigParser()
|
||||
config["general"] = {
|
||||
"base_port": 80,
|
||||
}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "http")
|
||||
|
||||
def test_configparser_config_https(self):
|
||||
config = configparser.ConfigParser()
|
||||
config["general"] = {
|
||||
"base_port": 443,
|
||||
}
|
||||
protocol = utils.get_protocol(config)
|
||||
self.assertEqual(protocol, "https")
|
||||
|
||||
def test_configparser_config_force_https(self):
|
||||
postive_values = ["yes", "Yes", "True", "true", True]
|
||||
negative_values = ["no", "No", "False", "false", False]
|
||||
for value in postive_values:
|
||||
self.assertEqual(get_force_ssl(value, True), "https")
|
||||
for value in negative_values:
|
||||
self.assertEqual(get_force_ssl(value, True), "http")
|
||||
|
||||
def test_fromisoformat(self):
|
||||
time = {
|
||||
"00:00:00.500000": datetime.time(microsecond=500000),
|
||||
"00:04:30.092540": datetime.time(minute=4, second=30, microsecond=92540),
|
||||
}
|
||||
for time_string, expected in time.items():
|
||||
result = utils.fromisoformat(time_string)
|
||||
self.assertEqual(result, expected)
|
||||
assert utils.get_protocol(config) == expected
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@pytest.mark.parametrize("payload", [{}, {"base_port": 80}])
|
||||
@pytest.mark.parametrize("use_config", [False, True])
|
||||
@pytest.mark.parametrize(
|
||||
"values, expected",
|
||||
[
|
||||
(["yes", "Yes", "True", "true", True], "https"),
|
||||
(["no", "No", "False", "false", False], "http"),
|
||||
],
|
||||
)
|
||||
def test_get_protocol_force_https(payload, use_config, values, expected):
|
||||
for value in values:
|
||||
config = ConfigParser() if use_config else {}
|
||||
config["general"] = {**payload, "force_ssl": value}
|
||||
assert utils.get_protocol(config) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload, expected",
|
||||
[
|
||||
("00:00:00.500000", datetime.time(microsecond=500000)),
|
||||
("00:04:30.092540", datetime.time(minute=4, second=30, microsecond=92540)),
|
||||
],
|
||||
)
|
||||
def test_fromisoformat(payload, expected):
|
||||
assert utils.fromisoformat(payload) == expected
|
||||
|
|
Loading…
Reference in New Issue