405 lines
12 KiB
Python
405 lines
12 KiB
Python
import json
|
|
|
|
from unittest import TestCase, skipIf
|
|
from unittest.mock import call, patch
|
|
|
|
from click.testing import CliRunner
|
|
from pkg_resources import get_distribution
|
|
from requests import HTTPError
|
|
|
|
from ip_listener.cli import DEFAULT_API_URL, run
|
|
|
|
|
|
class RunTestCase(TestCase):
|
|
def setUp(self):
|
|
patcher = patch("ip_listener.main.subprocess.check_output")
|
|
self.mocked_dns = patcher.start()
|
|
|
|
patcher = patch("ip_listener.main.requests.get")
|
|
self.mocked_get = patcher.start()
|
|
|
|
patcher = patch("ip_listener.main.requests.put")
|
|
self.mocked_put = 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("ip_listener.main", level="INFO") as logger:
|
|
result = self.runner.invoke(run, ["foobar.com", "TOKEN"])
|
|
|
|
self.assertEqual(
|
|
logger.output, ["INFO:ip_listener.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("ip_listener.main", level="INFO") as logger:
|
|
result = self.runner.invoke(run, ["foobar.com", "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", "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", "TOKEN", "--read-only"])
|
|
|
|
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("ip_listener.main", level="INFO") as logger:
|
|
result = self.runner.invoke(
|
|
run, ["foobar.com", "TOKEN", "--api-url", "https://other-provider.com"]
|
|
)
|
|
|
|
self.assertEqual(
|
|
logger.output, ["INFO:ip_listener.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("ip_listener.main", level="INFO") as logger:
|
|
result = self.runner.invoke(
|
|
run, ["foobar.com", "TOKEN"], env={"API_URL": "https://new-api.com"}
|
|
)
|
|
|
|
self.assertEqual(
|
|
logger.output, ["INFO:ip_listener.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"},
|
|
)
|
|
|
|
# see https://click.palletsprojects.com/en/master/changelog/#version-8-0
|
|
@skipIf(
|
|
get_distribution("click").version < "8",
|
|
"Multiple arguments through env variables not supported",
|
|
)
|
|
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("ip_listener.main", level="INFO") as logger:
|
|
result = self.runner.invoke(
|
|
run, ["TOKEN"], env={"DOMAINS": "foobar.com foofoo.com"}
|
|
)
|
|
|
|
self.assertIsNone(result.exception)
|
|
self.assertEqual(
|
|
logger.output,
|
|
[
|
|
"INFO:ip_listener.main:Updated domain foobar.com",
|
|
"INFO:ip_listener.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(),
|
|
]
|
|
|
|
self.mocked_get.assert_has_calls(expected_calls)
|
|
|
|
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)
|
|
|
|
def test_no_domains(self):
|
|
result = self.runner.invoke(run, ["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()
|