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 update from transip_client.main import API_URL class UpdateTestCase(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( update, ["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( update, ["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( update, ["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( update, ["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( update, ["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( update, ["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( update, ["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()