Sonny Bakker
e93d864ae1
Some checks failed
ci/woodpecker/push/tests Pipeline failed
Allows using different ways to retrieve hosts IP address
359 lines
12 KiB
Python
359 lines
12 KiB
Python
import json
|
|
|
|
from pathlib import Path
|
|
from unittest import TestCase
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from click.testing import CliRunner
|
|
from requests import HTTPError
|
|
|
|
from transip_client.cli import run
|
|
from transip_client.main import API_URL
|
|
|
|
|
|
class RunTestCase(TestCase):
|
|
def setUp(self):
|
|
patcher = patch("transip_client.main.requests.get")
|
|
self.mocked_get = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
patcher = patch("transip_client.main.requests.put")
|
|
self.mocked_put = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
patcher = patch("transip_client.main.requests.Session.send")
|
|
self.mocked_session = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
patcher = patch("transip_client.main.import_string")
|
|
self.mocked_import_string = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
self.runner = CliRunner()
|
|
|
|
self.private_key_path = (
|
|
Path(__file__).resolve().parent / "files" / "test-private-key.pem"
|
|
)
|
|
|
|
def test_simple(self):
|
|
mocked_adapter = MagicMock(get_ip=lambda: "111.420")
|
|
self.mocked_import_string.return_value = MagicMock(return_value=mocked_adapter)
|
|
|
|
self.mocked_get.side_effect = [
|
|
MagicMock(
|
|
json=lambda: {
|
|
"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": "token"}
|
|
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path), "foobar.com"],
|
|
catch_exceptions=False,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 0)
|
|
|
|
self.mocked_get.assert_called_with(
|
|
f"{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"{API_URL}/domains/foobar.com/dns",
|
|
data=expected_json,
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
|
|
def test_error_response(self):
|
|
mocked_adapter = MagicMock(get_ip=lambda: "111.420")
|
|
self.mocked_import_string.return_value = MagicMock(return_value=mocked_adapter)
|
|
|
|
self.mocked_session.return_value.json.return_value = {"token": "token"}
|
|
self.mocked_get.side_effect = [HTTPError]
|
|
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path), "foobar.com"],
|
|
catch_exceptions=False,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 0)
|
|
|
|
self.mocked_get.assert_called_with(
|
|
f"{API_URL}/domains/foobar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
|
|
self.mocked_put.assert_not_called()
|
|
|
|
def test_update_error_response(self):
|
|
mocked_adapter = MagicMock(get_ip=lambda: "111.420")
|
|
self.mocked_import_string.return_value = MagicMock(return_value=mocked_adapter)
|
|
|
|
self.mocked_session.return_value.json.return_value = {"token": "token"}
|
|
|
|
self.mocked_get.side_effect = [
|
|
MagicMock(
|
|
json=lambda: {
|
|
"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",
|
|
},
|
|
],
|
|
}
|
|
),
|
|
MagicMock(
|
|
json=lambda: {
|
|
"dnsEntries": [
|
|
{
|
|
"name": "@",
|
|
"expire": 60,
|
|
"type": "A",
|
|
"content": "111.421",
|
|
}
|
|
],
|
|
"_links": [
|
|
{
|
|
"rel": "self",
|
|
"link": "https://api.transip.nl/v6/domains/boofar.com/dns",
|
|
},
|
|
{
|
|
"rel": "domain",
|
|
"link": "https://api.transip.nl/v6/domains/boofar.com",
|
|
},
|
|
],
|
|
}
|
|
),
|
|
]
|
|
|
|
def raise_exception() -> None:
|
|
raise HTTPError
|
|
|
|
self.mocked_put.side_effect = (
|
|
MagicMock(raise_for_status=raise_exception),
|
|
MagicMock(status=200),
|
|
)
|
|
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path), "foobar.com", "boofar.com"],
|
|
catch_exceptions=False,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 0)
|
|
|
|
self.mocked_get.assert_has_calls(
|
|
(
|
|
call(
|
|
f"{API_URL}/domains/foobar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
),
|
|
call(
|
|
f"{API_URL}/domains/boofar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
),
|
|
)
|
|
)
|
|
|
|
self.assertCountEqual(
|
|
self.mocked_put.mock_calls,
|
|
(
|
|
call(
|
|
f"{API_URL}/domains/foobar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
data=json.dumps(
|
|
{
|
|
"dnsEntries": [
|
|
{
|
|
"name": "@",
|
|
"expire": 60,
|
|
"type": "A",
|
|
"content": "111.420",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
),
|
|
call(
|
|
f"{API_URL}/domains/boofar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
data=json.dumps(
|
|
{
|
|
"dnsEntries": [
|
|
{
|
|
"name": "@",
|
|
"expire": 60,
|
|
"type": "A",
|
|
"content": "111.420",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_matching_ip(self):
|
|
mocked_adapter = MagicMock(get_ip=lambda: "111.420")
|
|
self.mocked_import_string.return_value = MagicMock(return_value=mocked_adapter)
|
|
|
|
self.mocked_get.side_effect = [
|
|
MagicMock(
|
|
json=lambda: {
|
|
"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",
|
|
},
|
|
],
|
|
}
|
|
),
|
|
]
|
|
|
|
self.mocked_session.return_value.json.return_value = {"token": "token"}
|
|
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path), "foobar.com"],
|
|
catch_exceptions=False,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 0)
|
|
|
|
self.mocked_get.assert_called_with(
|
|
f"{API_URL}/domains/foobar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
|
|
self.mocked_put.assert_not_called()
|
|
|
|
def test_readonly(self):
|
|
mocked_adapter = MagicMock(get_ip=lambda: "111.420")
|
|
self.mocked_import_string.return_value = MagicMock(return_value=mocked_adapter)
|
|
|
|
self.mocked_get.side_effect = [
|
|
MagicMock(
|
|
json=lambda: {
|
|
"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": "token"}
|
|
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path), "foobar.com", "--read-only"],
|
|
catch_exceptions=False,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 0)
|
|
|
|
self.mocked_get.assert_called_with(
|
|
f"{API_URL}/domains/foobar.com/dns",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
|
|
self.mocked_put.assert_not_called()
|
|
|
|
def test_no_domains(self):
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str(self.private_key_path)],
|
|
catch_exceptions=True,
|
|
)
|
|
|
|
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_unknown_private_key_path(self):
|
|
result = self.runner.invoke(
|
|
run,
|
|
["my-user", str("/tmp/foo"), "foobar.com"],
|
|
catch_exceptions=True,
|
|
)
|
|
|
|
self.assertEqual(result.exit_code, 1)
|
|
self.assertEqual(str(result.exception), "Unknown private key path: /tmp/foo")
|
|
|
|
self.mocked_get.assert_not_called()
|
|
self.mocked_put.assert_not_called()
|