This commit is contained in:
sonny 2021-07-18 15:43:09 +02:00
parent 0cc7aef5eb
commit b06d4192a3
8 changed files with 426 additions and 106 deletions

View file

@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEpgIBAAKCAQEAz7ojmkmZ2QA2L0QW/K2BLoxLs6eGWw/LBV5E+beTPz9rejla
Qj/Dq8T7qnBvM8TtLH3Bwwu6H15oqnVis0OhAZTuAFwQuUrzw9g83S+KHxBGATVb
eWQxwkf9ma8GIX+xFnby3XGvVXZjKZgUMZAeFK7c/eXxnRJ8QnN/aoXibiFP0bgS
jjGpsoS6FdjRkau7BbdOJg0fRcTIjkdd11r5pg1oRGAApwU6g+HtjQWYLiNoIxba
8Vv0SSdK9V/hgk/aqLXGmCCjvLqvEK0KxK9y035Sn6C4AnohdiSO5SRYZgipM9F0
RJzc20qCcTL6CpIS4SupA6121eno/bxAQY4q8wIDAQABAoIBAQC3KM9fqWoIFtGw
F3+VSH9RRc8yF5K2FFTU5Ow4q48gA5GG8a8OHx8vA79L51uF8CuYQUJp8psoMZxk
QKDIo+cBeAnrM0Jjvxz1IGN6PAKzpSu0wRFpFdlyDvwjWFo1i1vgDP3UF/ubhYmm
ETws/4AmiJC/JtNFxhjeluxQRsECjLDf52iqO/LbyxQUAElIkMC24ni2SHnLZfQ3
KIBS3MdVULGrAO6dF805vF/CbPajzgsDZzuzcOFjYetHm+qvUV8Tp34M00Mzf5Ar
7joBLCHgeFWfwPIY0l7LnRhs8Z79vAi/slMi/zrQnFi/5aiLE86y195LXeXh59If
BS7eq+YhAoGBAPmGmVR282oMhDFkzR6N2vkKbRCg8p6QMYK3L6/vYYVbf+S4ePH+
GJo41iG76YIx77UPR5wE6+KRStmc6SkvXPy8mJzNVRzMX7xIgtLbNfx37HSqIkKy
CN3rppQ8VTxsaEM8LwKvAQ/V1xxvtmhct08oEQCPqeQHHdrOEvFwl7HbAoGBANUd
56NRDcKmF0mnoCgJpTZxVImpbOcrislQHKJvVjz8JegBcv+JBoX4p/g62FXh0CIJ
fr7KyKsWpH77zniNFu9xgwEs3RaK+z/+GsHsH4IWBkgj4ABweGOiCw+oeN0WdA8w
4okF2VYZbzXbpH3ULrwAvKnElGFcBboSY9yLwHLJAoGBAIGsKz6z2mfAPWqV4esA
+Uz22BsOKUex06kEnemmU13EYUBxhZjs3cg3xUAesYkRfmrvl91CyXsi2m0gmCLp
FD/bmsvSAWtH4nCsliAR/4pGoEE4sTlL4EPD1PuwJvORutVGD4Arhje+f12tyHOP
y0t9nOhXwIhaEm/FLB8AzjSFAoGBAK3NSKZ5KLawi1dnHAbq7tC6lg36nTTd3r6U
1fVmxTbRD/zoiad6UVaa1ilrnBhWI05O3g2tBP/6ZEanBthrf+Pgd81SkC+dQpAK
pDm4Xm3Rlmfo0fqpvpTKhyjK5V6wvA/Tdzv2CCvebELJEJoJm9943mO5TKUlzgnU
i5pGYrl5AoGBAN+b0liV/tDP7NT2pwSTqoW+8itEhBu0QxiHr/QM2941C4+tV4g9
wDC2CkQftSm1zx3unnDvU/WXSeJN4GWI7RmQNdsj84mPZVMfFqhvS8F7czbBMHV8
OHJdQYlyGPHP9WeP1H0K/zuFEcU6sY/Prl0fRIqhTroI6xEaEd+1bZx7
-----END RSA PRIVATE KEY-----

View file

@ -0,0 +1,505 @@
import json
import os
from unittest import TestCase
from unittest.mock import call, patch
from pathlib import Path
from click.testing import CliRunner
from requests import HTTPError
from transip_client.cli import DEFAULT_API_URL, run
class RunTestCase(TestCase):
def setUp(self):
patcher = patch("transip_client.main.subprocess.check_output")
self.mocked_dns = patcher.start()
patcher = patch("transip_client.main.requests.get")
self.mocked_get = patcher.start()
patcher = patch("transip_client.main.requests.put")
self.mocked_put = patcher.start()
patcher = patch("transip_client.main.requests.Session.send")
self.mocked_session = patcher.start()
self.runner = CliRunner()
def test_simple(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(run, ["foobar.com"], env={"TOKEN": "token"})
self.assertEqual(
logger.output, ["INFO:transip_client.main:Updated domain foobar.com"]
)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
expected_json = json.dumps(
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
]
}
)
self.mocked_put.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
data=expected_json,
headers={"Authorization": "Bearer token"},
)
def test_error_response(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.raise_for_status.side_effect = HTTPError
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(run, ["foobar.com"], env={"TOKEN": "token"})
error_log = logger.output[0]
self.assertIn("Failed retrieving information for foobar.com", error_log)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
self.mocked_put.assert_not_called()
def test_matching_ip(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
result = self.runner.invoke(run, ["foobar.com"], env={"TOKEN": "token"})
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
self.mocked_put.assert_not_called()
def test_readonly(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
result = self.runner.invoke(
run, ["foobar.com", "--read-only"], env={"TOKEN": "token"}
)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
self.mocked_put.assert_not_called()
def test_different_api_url(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(
run,
["foobar.com", "--api-url", "https://other-provider.com"],
env={"TOKEN": "token"},
)
self.assertEqual(
logger.output, ["INFO:transip_client.main:Updated domain foobar.com"]
)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"https://other-provider.com/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
expected_json = json.dumps(
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
]
}
)
self.mocked_put.assert_called_with(
f"https://other-provider.com/domains/foobar.com/dns",
data=expected_json,
headers={"Authorization": "Bearer token"},
)
def test_env_var(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(
run,
["foobar.com"],
env={
"TOKEN": "token",
"API_URL": "https://new-api.com",
},
)
self.assertEqual(
logger.output, ["INFO:transip_client.main:Updated domain foobar.com"]
)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
"https://new-api.com/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
)
expected_json = json.dumps(
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
]
}
)
self.mocked_put.assert_called_with(
"https://new-api.com/domains/foobar.com/dns",
data=expected_json,
headers={"Authorization": "Bearer token"},
)
def test_multi_arg_env_var(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.side_effect = [
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
},
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foofoo.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foofoo.com",
},
],
},
]
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(
run, [], env={"TOKEN": "token", "DOMAINS": "foobar.com foofoo.com"}
)
self.assertIsNone(result.exception)
self.assertEqual(
logger.output,
[
"INFO:transip_client.main:Updated domain foobar.com",
"INFO:transip_client.main:Updated domain foofoo.com",
],
)
self.assertEqual(result.exit_code, 0)
expected_calls = [
call(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer token"},
),
call().raise_for_status(),
call().json(),
call(
f"{DEFAULT_API_URL}/domains/foofoo.com/dns",
headers={"Authorization": "Bearer token"},
),
call().raise_for_status(),
call().json(),
]
# use any_order because of the asynchronous requests
self.mocked_get.assert_has_calls(expected_calls, any_order=True)
expected_json = json.dumps(
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
]
}
)
expected_calls = [
call(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
data=expected_json,
headers={"Authorization": "Bearer token"},
),
call().raise_for_status(),
call(
f"{DEFAULT_API_URL}/domains/foofoo.com/dns",
data=expected_json,
headers={"Authorization": "Bearer token"},
),
call().raise_for_status(),
]
self.mocked_put.assert_has_calls(expected_calls, any_order=True)
def test_no_domains(self):
result = self.runner.invoke(run, [], env={"TOKEN": "token"})
self.assertEqual(result.exit_code, 1)
self.assertEqual(str(result.exception), "No domain(s) specified")
self.mocked_get.assert_not_called()
self.mocked_put.assert_not_called()
def test_no_token_or_login_with_private_key_path(self):
result = self.runner.invoke(run, ["foobar.com"])
self.assertEqual(result.exit_code, 1)
self.assertEqual(
str(result.exception),
"Either a token or a login name with a path to a private key need"
" to be specified"
)
self.mocked_get.assert_not_called()
self.mocked_put.assert_not_called()
def test_login_without_private_key_path(self):
result = self.runner.invoke(run, ["foobar.com"], env={"LOGIN": "foo"})
self.assertEqual(result.exit_code, 1)
self.assertEqual(
str(result.exception),
"Both a login name and the path to a private key need to be specified"
)
self.mocked_get.assert_not_called()
self.mocked_put.assert_not_called()
def test_login_with_private_key_path(self):
self.mocked_dns.return_value = b"111.420\n"
self.mocked_get.return_value.json.return_value = {
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.421",
}
],
"_links": [
{
"rel": "self",
"link": "https://api.transip.nl/v6/domains/foobar.com/dns",
},
{
"rel": "domain",
"link": "https://api.transip.nl/v6/domains/foobar.com",
},
],
}
self.mocked_session.return_value.json.return_value = {"token": "FOOBAR"}
private_key_path = (
Path(os.path.dirname(__file__)) / "files" / "test-private-key.pem"
)
with self.assertLogs("transip_client.main", level="INFO") as logger:
result = self.runner.invoke(
run,
["foobar.com"],
env={"LOGIN": "foo", "PRIVATE_KEY_PATH": str(private_key_path)}
)
self.assertEqual(
logger.output, ["INFO:transip_client.main:Updated domain foobar.com"]
)
self.assertEqual(result.exit_code, 0)
self.mocked_get.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
headers={"Authorization": "Bearer FOOBAR"},
)
expected_json = json.dumps(
{
"dnsEntries": [
{
"name": "@",
"expire": 60,
"type": "A",
"content": "111.420",
}
]
}
)
self.mocked_put.assert_called_with(
f"{DEFAULT_API_URL}/domains/foobar.com/dns",
data=expected_json,
headers={"Authorization": "Bearer FOOBAR"},
)