From 3e3ebe586385a83b10c8f1d0b9ba9b67c8b56d2f Mon Sep 17 00:00:00 2001 From: Unit 193 Date: Wed, 20 Nov 2024 01:17:40 -0500 Subject: New upstream version 2.0.0. --- tests/client.py | 1025 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1025 insertions(+) create mode 100644 tests/client.py (limited to 'tests/client.py') diff --git a/tests/client.py b/tests/client.py new file mode 100644 index 0000000..9362be2 --- /dev/null +++ b/tests/client.py @@ -0,0 +1,1025 @@ +import json +import tempfile +import unittest +from pathlib import Path +from urllib.parse import urljoin + +import responses +from responses import matchers +from responses.registries import OrderedRegistry + +from pkb_client.client import ( + PKBClient, + PKBClientException, + API_ENDPOINT, + DNSRestoreMode, +) +from pkb_client.client import SSLCertBundle +from pkb_client.client.dns import DNSRecord, DNSRecordType +from pkb_client.client.forwarding import URLForwarding, URLForwardingType + + +class TestClientAuth(unittest.TestCase): + @responses.activate + def test_valid_auth(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "ping"), + json={"status": "SUCCESS", "yourIp": "127.0.0.1"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + ip_address = pkb_client.ping() + self.assertEqual("127.0.0.1", ip_address) + + @responses.activate + def test_invalid_auth(self): + pkb_client = PKBClient("key" + "s", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "ping"), + json={"status": "ERROR", "message": "Invalid credentials"}, + status=401, + ) + with self.assertRaises(PKBClientException): + pkb_client.ping() + + @responses.activate + def test_ping(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "ping"), + json={"status": "SUCCESS", "yourIp": "127.0.0.1"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + ip_address = pkb_client.ping() + self.assertEqual("127.0.0.1", ip_address) + + @responses.activate(registry=OrderedRegistry) + def test_create_dns_record(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "123456"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": 3600, + "prio": None, + } + ) + ], + ) + assert "123456" == pkb_client.create_dns_record( + "example.com", DNSRecordType.A, "127.0.0.1", "sub.example.com", 3600 + ) + + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "234561"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "sub.example.com", + "type": "MX", + "content": "127.0.0.1", + "ttl": 3600, + "prio": 2, + } + ) + ], + ) + assert "234561" == pkb_client.create_dns_record( + "example.com", DNSRecordType.MX, "127.0.0.1", "sub.example.com", 3600, 2 + ) + + def test_create_dns_record_invalid_prio_record_type(self): + pkb_client = PKBClient("key", "secret") + with self.assertRaises(ValueError): + pkb_client.create_dns_record( + "example.com", DNSRecordType.A, "127.0.0.1", "sub.example.com", 3600, 2 + ) + + @responses.activate(registry=OrderedRegistry) + def test_update_dns_record(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/edit/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "type": "A", + "content": "127.0.0.1", + "name": "sub.example.com", + "ttl": 3600, + "prio": None, + } + ) + ], + ) + + success = pkb_client.update_dns_record( + "example.com", + "123456", + DNSRecordType.A, + "127.0.0.1", + "sub.example.com", + 3600, + ) + self.assertTrue(success) + + responses.post( + url=urljoin(API_ENDPOINT, "dns/edit/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "type": "MX", + "content": "127.0.0.1", + "name": "sub.example.com", + "ttl": 3600, + "prio": 2, + } + ) + ], + ) + + success = pkb_client.update_dns_record( + "example.com", + "123456", + DNSRecordType.MX, + "127.0.0.1", + "sub.example.com", + 3600, + 2, + ) + self.assertTrue(success) + + def test_update_dns_records_invalid_prio_record_type(self): + pkb_client = PKBClient("key", "secret") + with self.assertRaises(ValueError): + pkb_client.update_dns_record( + "example.com", + "123456", + DNSRecordType.A, + "127.0.0.1", + "sub.example.com", + 3600, + 2, + ) + + @responses.activate + def test_update_all_dns_records(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/editByNameType/example.com/A/sub"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "type": "A", + "content": "127.0.0.1", + "ttl": 1234, + "prio": None, + } + ) + ], + ) + + success = pkb_client.update_all_dns_records( + "example.com", DNSRecordType.A, "sub", "127.0.0.1", 1234 + ) + + self.assertTrue(success) + + def test_update_all_dns_records_all_invalid_prio_record_type(self): + pkb_client = PKBClient("key", "secret") + + with self.assertRaises(ValueError): + pkb_client.update_all_dns_records( + "example.com", DNSRecordType.A, "sub", "127.0.0.1", 1234, 2 + ) + + @responses.activate + def test_delete_dns_record(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/delete/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + success = pkb_client.delete_dns_record("example.com", "123456") + + self.assertTrue(success) + + @responses.activate + def test_delete_all_dns_records(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/deleteByNameType/example.com/A/sub"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + success = pkb_client.delete_all_dns_records( + "example.com", DNSRecordType.A, "sub" + ) + + self.assertTrue(success) + + @responses.activate + def test_get_dns_records(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + records = pkb_client.get_dns_records("example.com") + + expected_records = [ + DNSRecord( + "123456", "example.com", DNSRecordType.A, "127.0.0.1", 600, None, "" + ), + DNSRecord( + "1234567", + "sub.example.com", + DNSRecordType.A, + "127.0.0.2", + 600, + None, + "", + ), + ] + self.assertEqual(expected_records, records) + + @responses.activate + def test_get_all_dns_records(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieveByNameType/example.com/A/sub"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + } + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + records = pkb_client.get_all_dns_records("example.com", DNSRecordType.A, "sub") + + expected_records = [ + DNSRecord( + "1234567", + "sub.example.com", + DNSRecordType.A, + "127.0.0.2", + 600, + None, + "", + ) + ] + self.assertEqual(expected_records, records) + + @responses.activate + def test_update_dns_servers(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "domain/updateNs/example.com"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "ns": ["ns1.example.com", "ns2.example.com"], + } + ) + ], + ) + + success = pkb_client.update_dns_servers( + "example.com", ["ns1.example.com", "ns2.example.com"] + ) + + self.assertTrue(success) + + @responses.activate + def test_get_url_forwards(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "domain/getUrlForwarding/example.com"), + json={ + "status": "SUCCESS", + "forwards": [ + { + "id": "123456", + "subdomain": "", + "location": "https://example.com", + "type": "temporary", + "includePath": "no", + "wildcard": "yes", + }, + { + "id": "234567", + "subdomain": "sub1", + "location": "https://sub1.example.com", + "type": "permanent", + "includePath": "no", + "wildcard": "yes", + }, + ], + }, + ) + + forwards = pkb_client.get_url_forwards("example.com") + + expected_forwards = [ + URLForwarding( + "123456", + "", + "https://example.com", + URLForwardingType.temporary, + False, + True, + ), + URLForwarding( + "234567", + "sub1", + "https://sub1.example.com", + URLForwardingType.permanent, + False, + True, + ), + ] + + self.assertEqual(expected_forwards, forwards) + + @responses.activate + def test_create_url_forward(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "domain/addUrlForward/example.com"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "subdomain": "sub.example.com", + "location": "https://www.example.com", + "type": "permanent", + "includePath": False, + "wildcard": False, + } + ) + ], + ) + + success = pkb_client.create_url_forward( + "example.com", + "sub.example.com", + "https://www.example.com", + URLForwardingType.permanent, + False, + False, + ) + + self.assertTrue(success) + + @responses.activate + def test_delete_url_forward(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "domain/deleteUrlForward/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + success = pkb_client.delete_url_forward("example.com", "123456") + + self.assertTrue(success) + + @responses.activate + def test_get_domain_pricing(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "pricing/get"), + json={ + "status": "SUCCESS", + "pricing": { + "com": { + "registration": "42.42", + "renewal": "4.2", + "transfer": "42.2", + "coupons": [], + }, + "test": { + "registration": "4.42", + "renewal": "44.2", + "transfer": "4.2", + "coupons": [], + }, + }, + }, + ) + + pricing = pkb_client.get_domain_pricing() + + expected_pricing = { + "com": { + "registration": "42.42", + "renewal": "4.2", + "transfer": "42.2", + "coupons": [], + }, + "test": { + "registration": "4.42", + "renewal": "44.2", + "transfer": "4.2", + "coupons": [], + }, + } + + self.assertEqual(expected_pricing, pricing) + + @responses.activate + def test_get_ssl_bundle(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "ssl/retrieve/example.com"), + json={ + "status": "SUCCESS", + "certificatechain": "----BEGIN CERTIFICATE-----\nabc1-----END CERTIFICATE-----\n\n----BEGIN CERTIFICATE-----\nabc2-----END CERTIFICATE-----\n\n----BEGIN CERTIFICATE-----\nabc3-----END CERTIFICATE-----\n", + "privatekey": "-----BEGIN PRIVATE KEY-----\nabc4-----END PRIVATE KEY-----\n", + "publickey": "-----BEGIN PUBLIC KEY-----\nabc5-----END PUBLIC KEY-----\n", + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + ssl_cert_bundle = pkb_client.get_ssl_bundle("example.com") + + expected_ssl_cert_bundle = SSLCertBundle( + certificate_chain="----BEGIN CERTIFICATE-----\nabc1-----END CERTIFICATE-----\n\n----BEGIN CERTIFICATE-----\nabc2-----END CERTIFICATE-----\n\n----BEGIN CERTIFICATE-----\nabc3-----END CERTIFICATE-----\n", + private_key="-----BEGIN PRIVATE KEY-----\nabc4-----END PRIVATE KEY-----\n", + public_key="-----BEGIN PUBLIC KEY-----\nabc5-----END PUBLIC KEY-----\n", + ) + + self.assertEqual(expected_ssl_cert_bundle, ssl_cert_bundle) + + @responses.activate + def test_export_dns_records(self): + pkb_client = PKBClient("key", "secret") + + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": "1200", + "prio": None, + "notes": "This is a comment", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + with tempfile.NamedTemporaryFile() as f: + pkb_client.export_dns_records("example.com", f.name) + + with open(f.name, "r") as f: + exported_dns_file = json.load(f) + + expected_exported_dns_file = { + "123456": { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": 600, + "prio": None, + "notes": "", + }, + "1234567": { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 1200, + "prio": None, + "notes": "This is a comment", + }, + } + + self.assertEqual(expected_exported_dns_file, exported_dns_file) + + @responses.activate(registry=OrderedRegistry, assert_all_requests_are_fired=True) + def test_import_dns_records_clear(self): + pkb_client = PKBClient("key", "secret") + + # first all records should be retrieved + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + # then all records should be deleted + responses.post( + url=urljoin(API_ENDPOINT, "dns/delete/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + responses.post( + url=urljoin(API_ENDPOINT, "dns/delete/example.com/1234567"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + # then all records should be imported / created + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "123456"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "", + "type": "A", + "content": "127.0.0.3", + "ttl": 600, + "prio": None, + } + ) + ], + ) + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "1234567"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "sub", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + } + ) + ], + ) + + with tempfile.TemporaryDirectory() as temp_dir: + filename = Path(temp_dir, "records.json") + with open(filename, "w") as f: + json.dump( + { + "123456": { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.3", + "ttl": 600, + "prio": None, + }, + "1234567": { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + }, + }, + f, + ) + + pkb_client.import_dns_records( + "example.com", str(filename), DNSRestoreMode.clear + ) + + @responses.activate(registry=OrderedRegistry, assert_all_requests_are_fired=True) + def test_import_dns_records_replace(self): + pkb_client = PKBClient("key", "secret") + + # first all records should be retrieved + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + # same record should be updated + responses.post( + url=urljoin(API_ENDPOINT, "dns/edit/example.com/1234567"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "type": "A", + "content": "127.0.0.3", + "name": "sub", + "ttl": 600, + "prio": None, + } + ) + ], + ) + + with tempfile.TemporaryDirectory() as temp_dir: + filename = Path(temp_dir, "records.json") + with open(filename, "w") as f: + json.dump( + { + "123451": { + "id": "123451", + "name": "test.example.com", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + }, + "1234562": { + "id": "1234562", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.3", + "ttl": 600, + "prio": None, + }, + }, + f, + ) + + pkb_client.import_dns_records( + "example.com", str(filename), DNSRestoreMode.replace + ) + + @responses.activate(registry=OrderedRegistry, assert_all_requests_are_fired=True) + def test_import_dns_records_keep(self): + pkb_client = PKBClient("key", "secret") + + # first all records should be retrieved + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + + # only new records should be created + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "1234562"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "test", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + } + ) + ], + ) + + with tempfile.TemporaryDirectory() as temp_dir: + filename = Path(temp_dir, "records.json") + with open(filename, "w") as f: + json.dump( + { + "123451": { + "id": "123451", + "name": "test.example.com", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + }, + "1234562": { + "id": "1234562", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.3", + "ttl": 600, + "prio": None, + }, + }, + f, + ) + + pkb_client.import_dns_records( + "example.com", str(filename), DNSRestoreMode.keep + ) + + @responses.activate(registry=OrderedRegistry, assert_all_requests_are_fired=True) + def test_import_bind_dns_records(self): + pkb_client = PKBClient("key", "secret") + + # first all records should be retrieved + responses.post( + url=urljoin(API_ENDPOINT, "dns/retrieve/example.com"), + json={ + "status": "SUCCESS", + "records": [ + { + "id": "123456", + "name": "example.com", + "type": "A", + "content": "127.0.0.1", + "ttl": "600", + "prio": None, + "notes": "", + }, + { + "id": "1234567", + "name": "sub.example.com", + "type": "A", + "content": "127.0.0.2", + "ttl": 600, + "prio": None, + "notes": "", + }, + ], + }, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + # then all records should be deleted + responses.post( + url=urljoin(API_ENDPOINT, "dns/delete/example.com/123456"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + responses.post( + url=urljoin(API_ENDPOINT, "dns/delete/example.com/1234567"), + json={"status": "SUCCESS"}, + match=[ + matchers.json_params_matcher( + {"apikey": "key", "secretapikey": "secret"} + ) + ], + ) + # then all records should be imported / created + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "123456"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "", + "type": "A", + "content": "127.0.0.3", + "ttl": 600, + "prio": None, + } + ) + ], + ) + responses.post( + url=urljoin(API_ENDPOINT, "dns/create/example.com"), + json={"status": "SUCCESS", "id": "1234567"}, + match=[ + matchers.json_params_matcher( + { + "apikey": "key", + "secretapikey": "secret", + "name": "sub", + "type": "A", + "content": "127.0.0.4", + "ttl": 600, + "prio": None, + } + ) + ], + ) + + with tempfile.TemporaryDirectory() as temp_dir: + filename = Path(temp_dir, "records.bind") + with open(filename, "w") as f: + f.write( + ( + "$ORIGIN example.com.\n" + "$TTL 1234\n" + "@ IN SOA dns.example.com. dns2.example.com. (100 300 100 6000 600)\n" + "example.com. IN 600 A 127.0.0.3\n" + "sub.example.com. 600 IN A 127.0.0.4" + ) + ) + + pkb_client.import_bind_dns_records(filename, DNSRestoreMode.clear) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3