From 2e045dad547cdfbcbb71aa55772e5f09ad76673a Mon Sep 17 00:00:00 2001 From: Ferdinando Formica Date: Mon, 29 Jan 2024 16:54:51 +0000 Subject: [PATCH] Add new redirect APIs --- ns1/__init__.py | 20 +++ ns1/redirect.py | 254 ++++++++++++++++++++++++++++++++++++ ns1/rest/redirect.py | 224 +++++++++++++++++++++++++++++++ tests/unit/test_redirect.py | 203 ++++++++++++++++++++++++++++ 4 files changed, 701 insertions(+) create mode 100644 ns1/redirect.py create mode 100644 ns1/rest/redirect.py create mode 100644 tests/unit/test_redirect.py diff --git a/ns1/__init__.py b/ns1/__init__.py index 384003b..31991e3 100644 --- a/ns1/__init__.py +++ b/ns1/__init__.py @@ -293,6 +293,26 @@ def pools(self): return ns1.rest.pools.Pools(self.config) + def redirects(self): + """ + Return a new raw REST interface to Redirect resources + + :rtype: :py:class:`ns1.rest.redirect.Redirects` + """ + import ns1.rest.redirect + + return ns1.rest.redirect.Redirects(self.config) + + def redirect_certificates(self): + """ + Return a new raw REST interface to RedirectCertificate resources + + :rtype: :py:class:`ns1.rest.redirect.RedirectCertificates` + """ + import ns1.rest.redirect + + return ns1.rest.redirect.RedirectCertificates(self.config) + # HIGH LEVEL INTERFACE def loadZone(self, zone, callback=None, errback=None): """ diff --git a/ns1/redirect.py b/ns1/redirect.py new file mode 100644 index 0000000..8ec2b9e --- /dev/null +++ b/ns1/redirect.py @@ -0,0 +1,254 @@ +from ns1.rest.redirect import Redirects +from ns1.rest.redirect import RedirectCertificates + + +class RedirectException(Exception): + pass + + +class Redirect(object): + """ + High level object representing a redirect. + """ + + def __init__(self, config): + """ + Create a new high level Redirect object + + :param ns1.config.Config config: config object + """ + self._rest = Redirects(config) + self.config = config + self.data = None + + def __repr__(self): + return "" % (self.__getitem__("domain"), self.__getitem__("path"), self.__getitem__("target")) + + def __getitem__(self, item): + if not self.data: + raise RedirectException("redirect not loaded") + return self.data.get(item, None) + + def reload(self, callback=None, errback=None): + """ + Reload redirect data from the API. + """ + return self.load(reload=True, callback=callback, errback=errback) + + def load(self, id=None, callback=None, errback=None, reload=False): + """ + Load redirect data from the API. + :param str id: redirect id to load + """ + if not reload and self.data: + raise RedirectException("redirect already loaded") + if id == None and self.data: + id = self.__getitem__("id") + if id == None: + raise RedirectException("no redirect id: did you mean to create?") + + def success(result, *args): + self.data = result + if callback: + return callback(self) + else: + return self + + return self._rest.retrieve(id, callback=success, errback=errback) + + def loadFromDict(self, cfg): + """ + Load redirect data from a dictionary. + :param dict cfg: dictionary containing *at least* either an id or domain/path/target + """ + if "id" in cfg or ("domain" in cfg and "path" in cfg and "target" in cfg): + self.data = cfg + return self + else: + raise RedirectException("insufficient configuration") + + def delete(self, callback=None, errback=None): + """ + Delete the redirect. + """ + id = self.__getitem__("id") + return self._rest.delete(id, callback=callback, errback=errback) + + def update(self, callback=None, errback=None, **kwargs): + """ + Update redirect configuration. Pass a list of keywords and their values to + update. For the list of keywords available for zone configuration, see + :attr:`ns1.rest.redirect.Redirects.INT_FIELDS` and + :attr:`ns1.rest.redirect.Redirects.PASSTHRU_FIELDS` + """ + if not self.data: + raise RedirectException("redirect not loaded") + + def success(result, *args): + self.data = result + if callback: + return callback(self) + else: + return self + + return self._rest.update( + self.data, callback=success, errback=errback, **kwargs + ) + + def create(self, domain, path, target, callback=None, errback=None, **kwargs): + """ + Create a new redirect. Pass a list of keywords and their values to + configure. For the list of keywords available for zone configuration, + see :attr:`ns1.rest.redirect.Redirects.INT_FIELDS` and + :attr:`ns1.rest.redirect.Redirects.PASSTHRU_FIELDS` + :param str domain: the domain to redirect from + :param str path: the path on the domain to redirect from + :param str target: the url to redirect to + """ + if self.data: + raise RedirectException("redirect already loaded") + + def success(result, *args): + self.data = result + if callback: + return callback(self) + else: + return self + + return self._rest.create(domain, path, target, callback=success, errback=errback, **kwargs) + + def retrieveCertificate(self, callback=None, errback=None): + """ + Retrieve the certificate associated to this redirect. + :return: the RedirectCertificate object + """ + return RedirectCertificate(self.config).load(self.__getitem__("certificate_id")) + + +def listRedirects(config, callback=None, errback=None): + """ + Lists all redirects currently configured. + :return: a list of Redirect objects + """ + def success(result, *args): + ret = [] + cfgs = result.get("results", None) + for cfg in cfgs: + ret.append(Redirect(config).loadFromDict(cfg)) + if callback: + return callback(ret) + else: + return ret + + return Redirects(config).list(callback=success, errback=errback) + + +class RedirectCertificate(object): + """ + High level object representing a redirect certificate. + """ + + def __init__(self, config): + """ + Create a new high level RedirectCertificate object + + :param ns1.config.Config config: config object + """ + self._rest = RedirectCertificates(config) + self.config = config + self.data = None + + def __repr__(self): + return "" % self.__getitem__("domain") + + def __getitem__(self, item): + if not self.data: + raise RedirectException("redirect certificate not loaded") + return self.data.get(item, None) + + def reload(self, callback=None, errback=None): + """ + Reload redirect certificate data from the API. + """ + return self.load(reload=True, callback=callback, errback=errback) + + def load(self, id=None, callback=None, errback=None, reload=False): + """ + Load redirect certificate data from the API. + :param str id: redirect certificate id to load + """ + if not reload and self.data: + raise RedirectException("redirect certificate already loaded") + if id == None and self.data: + id = self.__getitem__("id") + if id == None: + raise RedirectException("no redirect certificate id: did you mean to create?") + + def success(result, *args): + self.data = result + if callback: + return callback(self) + else: + return self + + return self._rest.retrieve(id, callback=success, errback=errback) + + def loadFromDict(self, cfg): + """ + Load redirect data from a dictionary. + :param dict cfg: dictionary containing *at least* either an id or a domain + """ + if "id" in cfg or "domain" in cfg: + self.data = cfg + return self + else: + raise RedirectException("insufficient configuration") + + def delete(self, callback=None, errback=None): + """ + Requests to revoke the redirect certificate. + """ + id = self.__getitem__("id") + return self._rest.delete(id, callback=callback, errback=errback) + + def update(self, callback=None, errback=None): + """ + Requests to renew the redirect certificate. + """ + id = self.__getitem__("id") + return self._rest.update(id, callback=callback, errback=errback) + + def create(self, domain, callback=None, errback=None): + """ + Request a new redirect certificate. + :param str domain: the domain to issue the certificate for + """ + if self.data: + raise RedirectException("redirect certificate already loaded") + + def success(result, *args): + self.data = result + if callback: + return callback(self) + else: + return self + + return self._rest.create(domain, callback=success, errback=errback) + + +def listRedirectCertificates(config, callback=None, errback=None): + """ + Lists all redirects certificates currently configured. + :return: a list of RedirectCertificate objects + """ + def success(result, *args): + ret = [] + cfgs = result.get("results", None) + for cfg in cfgs: + ret.append(RedirectCertificate(config).loadFromDict(cfg)) + if callback: + return callback(ret) + else: + return ret + + return RedirectCertificates(config).list(callback=success, errback=errback) diff --git a/ns1/rest/redirect.py b/ns1/rest/redirect.py new file mode 100644 index 0000000..3aa70bb --- /dev/null +++ b/ns1/rest/redirect.py @@ -0,0 +1,224 @@ +from . import resource + + +class Redirects(resource.BaseResource): + ROOT = "redirect" + SEARCH_ROOT = "redirect" + + PASSTHRU_FIELDS = [ + "id", + "certificate_id", + "domain", + "path", + "target", + "tags", + "forwarding_mode", + "forwarding_type", + ] + BOOL_FIELDS = ["ssl_enabled", "force_redirect", "query_forwarding"] + + def _buildBody(self, domain, path, target, **kwargs): + body = { + "domain": domain, + "path": path, + "target": target, + } + self._buildStdBody(body, kwargs) + return body + + def import_file(self, cfg, cfgFile, callback=None, errback=None, **kwargs): + files = [("cfgfile", (cfgFile, open(cfgFile, "rb"), "text/plain"))] + return self._make_request( + "PUT", + "%s/importexport" % self.ROOT, + files=files, + callback=callback, + errback=errback, + ) + + def create(self, domain, path, target, callback=None, errback=None, **kwargs): + body = self._buildBody(domain, path, target, **kwargs) + return self._make_request( + "PUT", + "%s" % self.ROOT, + body=body, + callback=callback, + errback=errback, + ) + + def update(self, cfg, callback=None, errback=None, **kwargs): + self._buildStdBody(cfg, kwargs) + return self._make_request( + "POST", + "%s/%s" % (self.ROOT, cfg["id"]), + body=cfg, + callback=callback, + errback=errback, + ) + + def delete(self, cfgId, callback=None, errback=None): + return self._make_request( + "DELETE", + "%s/%s" % (self.ROOT, cfgId), + callback=callback, + errback=errback, + ) + + def list(self, callback=None, errback=None): + return self._make_request( + "GET", + "%s" % self.ROOT, + callback=callback, + errback=errback, + pagination_handler=redirect_list_pagination, + ) + + def retrieve(self, cfgId, callback=None, errback=None): + return self._make_request( + "GET", + "%s/%s" % (self.ROOT, cfgId), + callback=callback, + errback=errback, + ) + + def searchSource( + self, + query, + max=None, + callback=None, + errback=None, + ): + request = "{}?source={}".format(self.SEARCH_ROOT, query) + if max is not None: + request += "&limit=" + str(max) + return self._make_request( + "GET", + request, + params={}, + callback=callback, + errback=errback, + ) + + def searchTarget( + self, + query, + max=None, + callback=None, + errback=None, + ): + request = "{}?target={}".format(self.SEARCH_ROOT, query) + if max is not None: + request += "&limit=" + str(max) + return self._make_request( + "GET", + request, + params={}, + callback=callback, + errback=errback, + ) + + def searchTag( + self, + query, + max=None, + callback=None, + errback=None, + ): + request = "{}?tag={}".format(self.SEARCH_ROOT, query) + if max is not None: + request += "&limit=" + str(max) + return self._make_request( + "GET", + request, + params={}, + callback=callback, + errback=errback, + ) + + +class RedirectCertificates(resource.BaseResource): + ROOT = "redirect/certificates" + SEARCH_ROOT = "redirect/certificates" + + PASSTHRU_FIELDS = [ + "id", + "domain", + "certificate", + "errors", + ] + BOOL_FIELDS = ["processing"] + + def _buildBody(self, domain, **kwargs): + body = { + "domain": domain, + } + self._buildStdBody(body, kwargs) + return body + + def create(self, domain, callback=None, errback=None, **kwargs): + body = self._buildBody(domain, **kwargs) + return self._make_request( + "PUT", + "%s" % self.ROOT, + body=body, + callback=callback, + errback=errback, + ) + + def update(self, certId, callback=None, errback=None, **kwargs): + return self._make_request( + "POST", + "%s/%s" % (self.ROOT, certId), + callback=callback, + errback=errback, + ) + + def delete(self, certId, callback=None, errback=None): + return self._make_request( + "DELETE", + "%s/%s" % (self.ROOT, certId), + callback=callback, + errback=errback, + ) + + def list(self, callback=None, errback=None): + return self._make_request( + "GET", + "%s" % self.ROOT, + callback=callback, + errback=errback, + pagination_handler=redirect_list_pagination, + ) + + def retrieve(self, certId, callback=None, errback=None): + return self._make_request( + "GET", + "%s/%s" % (self.ROOT, certId), + callback=callback, + errback=errback, + ) + + def search( + self, + query, + max=None, + callback=None, + errback=None, + ): + request = "{}?domain={}".format(self.SEARCH_ROOT, query) + if max is not None: + request += "&limit=" + str(max) + return self._make_request( + "GET", + request, + params={}, + callback=callback, + errback=errback, + ) + + +# successive pages extend the list and the count +def redirect_list_pagination(curr_json, next_json): + curr_json["count"] += next_json["count"] + curr_json["results"].extend(next_json["results"]) + return curr_json diff --git a/tests/unit/test_redirect.py b/tests/unit/test_redirect.py new file mode 100644 index 0000000..6b734c5 --- /dev/null +++ b/tests/unit/test_redirect.py @@ -0,0 +1,203 @@ +from ns1 import NS1 +from ns1.redirect import Redirect, RedirectCertificate, listRedirects +import ns1.rest.redirect +import pytest + +try: # Python 3.3 + + import unittest.mock as mock +except ImportError: + import mock + + +@pytest.fixture +def redirect_config(config): + config.loadFromDict( + { + "endpoint": "api.nsone.net", + "default_key": "test1", + "keys": { + "test1": { + "key": "key-1", + "desc": "test key number 1", + "writeLock": True, + } + }, + } + ) + + return config + + +def test_rest_redirect_list(redirect_config): + z = NS1(config=redirect_config).redirects() + z._make_request = mock.MagicMock() + z.list() + z._make_request.assert_called_once_with( + "GET", + "redirect", + callback=None, + errback=None, + pagination_handler=ns1.rest.redirect.redirect_list_pagination, + ) + + +@pytest.mark.parametrize("cfgId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_redirect_retrieve(redirect_config, cfgId, url): + z = NS1(config=redirect_config).redirects() + z._make_request = mock.MagicMock() + z.retrieve(cfgId) + z._make_request.assert_called_once_with( + "GET", + url, + callback=None, + errback=None, + ) + + +def test_rest_redirect_create(redirect_config): + z = NS1(config=redirect_config).redirects() + z._make_request = mock.MagicMock() + z.create(domain="www.test.com", path="/", target="http://localhost") + z._make_request.assert_called_once_with( + "PUT", + "redirect", + body = { + "domain": "www.test.com", + "path": "/", + "target": "http://localhost", + }, + callback=None, + errback=None, + ) + + +@pytest.mark.parametrize("cfgId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_redirect_update(redirect_config, cfgId, url): + z = NS1(config=redirect_config).redirects() + z._make_request = mock.MagicMock() + cfg = { + "id": cfgId, + "domain": "www.test.com", + "path": "/", + "target": "https://www.google.com", + } + z.update(cfg, domain="www.test.com", path="/", target="http://localhost") + z._make_request.assert_called_once_with( + "POST", + url, + body = { + "id": cfgId, + "domain": "www.test.com", + "path": "/", + "target": "http://localhost", + }, + callback=None, + errback=None, + ) + + +@pytest.mark.parametrize("cfgId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_redirect_delete(redirect_config, cfgId, url): + z = NS1(config=redirect_config).redirects() + z._make_request = mock.MagicMock() + z.delete(cfgId) + z._make_request.assert_called_once_with( + "DELETE", + url, + callback=None, + errback=None, + ) + + +def test_rest_redirect_buildbody(redirect_config): + z = ns1.rest.redirect.Redirects(redirect_config) + kwargs = { + "domain": "www.test.com", + "path": "/", + "target": "http://localhost", + } + body = { + "domain": "www.test.com", + "path": "/", + "target": "http://localhost", + } + assert z._buildBody(**kwargs) == body + + +def test_rest_certificate_list(redirect_config): + z = NS1(config=redirect_config).redirect_certificates() + z._make_request = mock.MagicMock() + z.list() + z._make_request.assert_called_once_with( + "GET", + "redirect/certificates", + callback=None, + errback=None, + pagination_handler=ns1.rest.redirect.redirect_list_pagination, + ) + + +@pytest.mark.parametrize("cfgId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/certificates/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_certificate_retrieve(redirect_config, cfgId, url): + z = NS1(config=redirect_config).redirect_certificates() + z._make_request = mock.MagicMock() + z.retrieve(cfgId) + z._make_request.assert_called_once_with( + "GET", + url, + callback=None, + errback=None, + ) + + +def test_rest_certificate_create(redirect_config): + z = NS1(config=redirect_config).redirect_certificates() + z._make_request = mock.MagicMock() + z.create(domain="www.test.com") + z._make_request.assert_called_once_with( + "PUT", + "redirect/certificates", + body = { + "domain": "www.test.com", + }, + callback=None, + errback=None, + ) + + +@pytest.mark.parametrize("certId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/certificates/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_certificate_update(redirect_config, certId, url): + z = NS1(config=redirect_config).redirect_certificates() + z._make_request = mock.MagicMock() + z.update(certId) + z._make_request.assert_called_once_with( + "POST", + url, + callback=None, + errback=None, + ) + + +@pytest.mark.parametrize("certId, url", [("96529d62-fb0c-4150-b5ad-6e5b8b2736f6", "redirect/certificates/96529d62-fb0c-4150-b5ad-6e5b8b2736f6")]) +def test_rest_certificate_delete(redirect_config, certId, url): + z = NS1(config=redirect_config).redirect_certificates() + z._make_request = mock.MagicMock() + z.delete(certId) + z._make_request.assert_called_once_with( + "DELETE", + url, + callback=None, + errback=None, + ) + + +def test_rest_certificate_buildbody(redirect_config): + z = ns1.rest.redirect.Redirects(redirect_config) + kwargs = { + "domain": "www.test.com", + } + body = { + "domain": "www.test.com", + } + assert z._buildBody(**kwargs) == body +