diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f4abcd..287c744 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ENHANCEMENTS: * Adds support for new split monitoring permissions create_jobs, update_jobs and delete_jobs +* Remove DDI (DHCP & IPAM) code ## 0.20.0 (June 19th, 2024) diff --git a/doc/api/ipam.rst b/doc/api/ipam.rst deleted file mode 100644 index b7b72e8..0000000 --- a/doc/api/ipam.rst +++ /dev/null @@ -1,11 +0,0 @@ -ns1.ipam -========= - -A collection of classes that produce IPAM Objects. - -.. automodule:: ns1.ipam - :members: - :undoc-members: - :show-inheritance: - - diff --git a/doc/api/modules.rst b/doc/api/modules.rst index b86bf08..e5d4a98 100644 --- a/doc/api/modules.rst +++ b/doc/api/modules.rst @@ -8,5 +8,4 @@ ns1 package config zones records - ipam rest diff --git a/doc/api/rest.rst b/doc/api/rest.rst index c77f233..ed0debb 100644 --- a/doc/api/rest.rst +++ b/doc/api/rest.rst @@ -33,11 +33,6 @@ A thin layer over the NS1 REST API :undoc-members: :show-inheritance: -.. automodule:: ns1.rest.ipam - :members: - :undoc-members: - :show-inheritance: - .. automodule:: ns1.rest.team :members: :undoc-members: diff --git a/ns1/__init__.py b/ns1/__init__.py index 8ddaeb7..daab292 100644 --- a/ns1/__init__.py +++ b/ns1/__init__.py @@ -63,66 +63,6 @@ def records(self): return ns1.rest.records.Records(self.config) - def addresses(self): - """ - Return a new raw REST interface to address resources - - :rtype: :py:class:`ns1.rest.ipam.Adresses` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Addresses(self.config) - - def networks(self): - """ - Return a new raw REST interface to network resources - - :rtype: :py:class:`ns1.rest.ipam.Networks` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Networks(self.config) - - def scope_groups(self): - """ - Return a new raw REST interface to scope_group resources - - :rtype: :py:class:`ns1.rest.ipam.Scopegroups` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Scopegroups(self.config) - - def reservations(self): - """ - Return a new raw REST interface to reservation resources - - :rtype: :py:class:`ns1.rest.ipam.Reservations` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Reservations(self.config) - - def scopes(self): - """ - Return a new raw REST interface to scope resources - - :rtype: :py:class:`ns1.rest.ipam.Scopes` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Scopes(self.config) - - def optiondefs(self): - """ - Return a new raw REST interface to optiondefs resources - - :rtype: :py:class:`ns1.rest.ipam.Optiondefs` - """ - import ns1.rest.ipam - - return ns1.rest.ipam.Optiondefs(self.config) - def stats(self): """ Return a new raw REST interface to stats resources @@ -263,36 +203,6 @@ def views(self): return ns1.rest.views.Views(self.config) - def client_classes(self): - """ - Return a new raw REST interface to Client Classes resources - - :rtype: :py:class:`ns1.rest.client_classes.ClientClasses` - """ - import ns1.rest.client_classes - - return ns1.rest.client_classes.ClientClasses(self.config) - - def dhcp_option_spaces(self): - """ - Return a new raw REST interface to DHCP Option Spaces resources - - :rtype: :py:class:`ns1.rest.dhcp_option_spaces.DHCOptionSpaces` - """ - import ns1.rest.dhcp_option_spaces - - return ns1.rest.dhcp_option_spaces.DHCPOptionSpaces(self.config) - - def pools(self): - """ - Return a new raw REST interface to Pools resources - - :rtype: :py:class:`ns1.rest.pools.Pools` - """ - import ns1.rest.pools - - return ns1.rest.pools.Pools(self.config) - def datasets(self): """ Return a new raw REST interface to Datasets resources @@ -440,274 +350,3 @@ def createMonitor(self, callback=None, errback=None, **kwargs): monitor = ns1.monitoring.Monitor(self.config) return monitor.create(callback=callback, errback=errback, **kwargs) - - def loadNetworkbyID(self, id, callback=None, errback=None): - """ - Load an existing Network by ID into a high level Network object - - :param int id: id of an existing Network - """ - import ns1.ipam - - network = ns1.ipam.Network(self.config, id=id) - - return network.load(callback=callback, errback=errback) - - def loadNetworkbyName(self, name, callback=None, errback=None): - """ - Load an existing Network by name into a high level Network object - - :param str name: Name of an existing Network - """ - import ns1.ipam - - network = ns1.ipam.Network(self.config, name=name) - - return network.load(callback=callback, errback=errback) - - def createNetwork( - self, name, scope_group_id=None, callback=None, errback=None, **kwargs - ): - """ - Create a new Network - For the list of keywords available, see :attr:`ns1.rest.ipam.Networks.INT_FIELDS` and :attr:`ns1.rest.ipam.Networks.PASSTHRU_FIELDS` - - :param str name: Name of the Network to be created - :param int scope_group: (Optional) id of an existing scope group to associate with - """ - import ns1.ipam - - if scope_group_id is not None: - scope_group = ns1.ipam.Scopegroup( - self.config, id=scope_group_id - ).load() - kwargs["scope_group"] = scope_group - network = ns1.ipam.Network(self.config, name=name) - - return network.create(callback=callback, errback=errback, **kwargs) - - def loadAddressbyID(self, id, callback=None, errback=None): - """ - Load an existing address by ID into a high level Address object - - :param int id: id of an existing Address - """ - import ns1.ipam - - address = ns1.ipam.Address(self.config, id=id) - - return address.load(callback=callback, errback=errback) - - def loadAddressbyPrefix( - self, prefix, status, network_id, callback=None, errback=None - ): - """ - Load an existing address by prefix, status and network into a high level Address object - - :param str prefix: CIDR prefix of an existing Address - :param str status: The status of address assignment (planned or assigned) - :param int network_id: network_id associated with the address - """ - import ns1.ipam - - network = ns1.ipam.Network(self.config, id=network_id).load() - address = ns1.ipam.Address( - self.config, prefix=prefix, status=status, network=network - ) - - return address.load(callback=callback, errback=errback) - - def createAddress( - self, prefix, status, network_id, callback=None, errback=None, **kwargs - ): - """ - Create a new Address - For the list of keywords available, see :attr:`ns1.rest.ipam.Addresses.INT_FIELDS` and :attr:`ns1.rest.ipam.Addresses.PASSTHRU_FIELDS` - - :param str prefix: CIDR prefix of the address to be created - :param str status: The status of address assignment (planned or assigned) - :param int network_id: network_id associated with the address - """ - import ns1.ipam - - network = ns1.ipam.Network(self.config, id=network_id).load() - address = ns1.ipam.Address( - self.config, prefix=prefix, status=status, network=network - ) - - return address.create(callback=callback, errback=errback, **kwargs) - - def loadScopeGroup(self, id, callback=None, errback=None): - """ - Load an existing Scope Group into a high level Scope Group object - - :param int id: id of an existing ScopeGroup - """ - import ns1.ipam - - scope_group = ns1.ipam.Scopegroup(self.config, id=id) - - return scope_group.load(callback=callback, errback=errback) - - def createScopeGroup( - self, - name, - service_def_id, - dhcp4, - dhcp6, - callback=None, - errback=None, - **kwargs - ): - """ - Create a new Scope Group - For the list of keywords available, see :attr:`ns1.rest.ipam.ScopeGroups.INT_FIELDS` and :attr:`ns1.rest.ipam.ScopeGroups.PASSTHRU_FIELDS` - - :param str name: Name of the Scope Group to be created - :param int service_group_id: id of the service group the Scope group is associated with - :param ns1.ipam.DHCPIOptions dhcp4: DHCPOptions object that contains the options for dhcpv4 - :param ns1.ipam.DHCPIOptions dhcp6: DHCPOptions object that contains the options for dhcpv6 - """ - import ns1.ipam - - scope_group = ns1.ipam.Scopegroup( - self.config, name=name, service_def_id=service_def_id - ) - - return scope_group.create( - dhcp4=dhcp4, - dhcp6=dhcp6, - callback=callback, - errback=errback, - **kwargs - ) - - def createReservation( - self, - scopegroup_id, - address_id, - mac, - dhcp_options=None, - callback=None, - errback=None, - **kwargs - ): - """ - Create a new Reservation - For the list of keywords available, see :attr:`ns1.rest.ipam.Reservation.INT_FIELDS` and :attr:`ns1.rest.ipam.Reservation.PASSTHRU_FIELDS` - - :param int scopegroup_id: id of the scope group - :param int address_id: id of the address the reservation is associated with - :param str mac: mac address of the reservation - :param list options: dhcp options of the reservation - """ - import ns1.ipam - - reservation = ns1.ipam.Reservation( - self.config, - scopegroup_id, - address_id, - options=dhcp_options, - mac=mac, - ) - - return reservation.create(callback=callback, errback=errback, **kwargs) - - def loadReservation( - self, - scopegroup_id, - address_id, - reservation_id=None, - callback=None, - errback=None, - ): - import ns1.ipam - - reservation = ns1.ipam.Reservation( - self.config, scopegroup_id, address_id, reservation_id - ) - - return reservation.load(callback=callback, errback=errback) - - def createScope( - self, - scopegroup_id, - address_id, - dhcp_options=None, - callback=None, - errback=None, - **kwargs - ): - """ - Create a new Scope - For the list of keywords available, see :attr:`ns1.rest.ipam.Scope.INT_FIELDS` and :attr:`ns1.rest.ipam.Scope.PASSTHRU_FIELDS` - - :param int scopegroup_id: id of the scope group - :param int address_id: id of the address the scope is associated with - :param DHCPOptions options: DHCPOptions object that contains the settings for the scope - """ - import ns1.ipam - - scope = ns1.ipam.Scope( - self.config, scopegroup_id, address_id, dhcp_options - ) - - return scope.create(callback=callback, errback=errback, **kwargs) - - def loadScope( - self, scopegroup_id, address_id, callback=None, errback=None - ): - import ns1.ipam - - scope = ns1.ipam.Scope(self.config, scopegroup_id, address_id) - - return scope.load(callback=callback, errback=errback) - - def loadLeases( - self, - scope_group_id=None, - scope_id=None, - limit=None, - offset=None, - callback=None, - errback=None, - ): - import ns1.ipam - - lease = ns1.ipam.Lease(self.config) - - return lease.load( - scope_group_id, - scope_id, - limit, - offset, - callback=callback, - errback=errback, - ) - - def generateDHCPOptionsTemplate(self, address_family): - """ - Generate boilerplate dictionary to hold dhcp options - - :param str address_family: dhcpv4 or dhcpv6 - :return: dict containing valid option set for address family - """ - from ns1.ipam import DHCPOptions - - options = {} - - for option in DHCPOptions.OPTIONS[address_family]: - options[option] = "" - - return options - - def loadDHCPOptions(self, address_family, options): - """ - Create a high level DHCPOptions object - - :param str address_family: Address family of the options. Can be either dhcpv4 or dhcpv6 - :param dict options: Dictionary containing the option set to apply for this address family. Note: only those specified will be applied. Allowed options can be found in :attr:`ns1.ipam.DHCPOptions.OPTIONS` - """ - import ns1.ipam - - return ns1.ipam.DHCPOptions(address_family, options) diff --git a/ns1/client_classes.py b/ns1/client_classes.py deleted file mode 100644 index 961cd4a..0000000 --- a/ns1/client_classes.py +++ /dev/null @@ -1,77 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# -from ns1.rest.client_classes import ClientClasses - - -class ClientClassException(Exception): - pass - - -class ClientClass(object): - def __init__(self, config, client_class): - self._rest = ClientClasses(config) - self.config = config - self.client_class = client_class - self.data = None - - def __repr__(self): - return "" % self.client_class - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - if not reload and self.data: - raise ClientClassException("client class already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.retrieve( - self.client_class, callback=success, errback=errback - ) - - def delete(self, callback=None, errback=None): - return self._rest.delete( - self.client_class, callback=callback, errback=errback - ) - - def update(self, callback=None, errback=None, **kwargs): - if not self.data: - raise ClientClassException("client class not loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.client_class, callback=success, errback=errback, **kwargs - ) - - def create(self, callback=None, errback=None, **kwargs): - if self.data: - raise ClientClassException("client class already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.client_class, callback=success, errback=errback, **kwargs - ) diff --git a/ns1/dhcp_option_spaces.py b/ns1/dhcp_option_spaces.py deleted file mode 100644 index 9bd7111..0000000 --- a/ns1/dhcp_option_spaces.py +++ /dev/null @@ -1,63 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# - -from ns1.rest.dhcp_option_spaces import DHCPOptionSpaces - - -class DHCPOptionException(Exception): - pass - - -class DHCPOption(object): - def __init__(self, config, dhcp_option): - self._rest = DHCPOptionSpaces(config) - self.config = config - self.dhcp_option = dhcp_option - self.data = None - - def __repr__(self): - return "" % self.dhcp_option - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - if not reload and self.data: - raise DHCPOptionException("dhcp option already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.retrieve( - self.dhcp_option, callback=success, errback=errback - ) - - def delete(self, callback=None, errback=None): - return self._rest.delete( - self.dhcp_option, callback=callback, errback=errback - ) - - def create(self, callback=None, errback=None, **kwargs): - if self.data: - raise DHCPOptionException("dhcp option already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.dhcp_option, callback=success, errback=errback, **kwargs - ) diff --git a/ns1/ipam.py b/ns1/ipam.py deleted file mode 100644 index 674fc63..0000000 --- a/ns1/ipam.py +++ /dev/null @@ -1,1133 +0,0 @@ -# -# Copyright (c) 2019 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# -from ns1.rest.errors import ResourceException -from ns1.rest.ipam import Addresses -from ns1.rest.ipam import Leases -from ns1.rest.ipam import Networks -from ns1.rest.ipam import Optiondefs -from ns1.rest.ipam import Reservations -from ns1.rest.ipam import Scopegroups -from ns1.rest.ipam import Scopes - - -class NetworkException(Exception): - pass - - -class AddressException(Exception): - pass - - -class ScopegroupException(Exception): - pass - - -class ReservationException(Exception): - pass - - -class ScopeException(Exception): - pass - - -class DHCPOptionsException(Exception): - pass - - -class LeaseException(Exception): - pass - - -class OptiondefException(Exception): - pass - - -class Network(object): - def __init__(self, config, name=None, id=None, tags=None): - """ - Create a new high level Network object - - :param ns1.config.Config config: config object - :param str name: network name - :param int id: id of an existing Network - :param dict tags: tags of the network - """ - self._rest = Networks(config) - self.config = config - self.name = name - self.id = id - self.report = {} - self.tags = tags - self.data = None - - def __repr__(self): - return "" % self.name - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload network data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load network data from the API. - """ - - if not reload and self.data: - raise NetworkException("Network already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.name = result["name"] - self.report = self._rest.report(self.id) - - if "tags" in result: - self.tags = result["tags"] - - if callback: - return callback(self) - else: - return self - - if self.id is None: - if self.name is None: - raise NetworkException("Must at least specify an id or name") - else: - self.id = [ - network - for network in self._rest.list() - if network["name"] == self.name - ][0]["id"] - - return self._rest.retrieve(self.id, callback=success, errback=errback) - - def delete(self, callback=None, errback=None): - """ - Delete the Network and all associated addresses - """ - - return self._rest.delete(self.id, callback=callback, errback=errback) - - def update(self, callback=None, errback=None, **kwargs): - """ - Update Network configuration. Pass a list of keywords and their values to update. - For the list of keywords available for zone configuration, see :attr:`ns1.rest.ipam.Networks.INT_FIELDS` and :attr:`ns1.rest.ipam.Networks.PASSTHRU_FIELDS` - """ - - if not self.data: - raise NetworkException("Network not loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.name = result["name"] - self.report = self._rest.report(self.id) - - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.id, callback=success, errback=errback, **kwargs - ) - - def create(self, callback=None, errback=None, **kwargs): - """ - Create a new Network. Pass a list of keywords and their values to configure. - For the list of keywords available for network configuration, see :attr:`ns1.rest.ipam.Networks.INT_FIELDS` and :attr:`ns1.rest.ipam.Networks.PASSTHRU_FIELDS` - """ - - if self.data: - raise NetworkException("Network already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.name = result["name"] - self.report = self._rest.report(self.id) - - if callback: - return callback(self) - else: - return self - - return self._rest.create( - name=self.name, callback=success, errback=errback, **kwargs - ) - - def new_address( - self, prefix, status, callback=None, errback=None, **kwargs - ): - """ - Create a new address space in this Network - - :param str prefix: The CIDR prefix of the address to add - :param str status: planned, assigned - :return: The newly created Address object - """ - - if not self.data: - raise NetworkException("Network not loaded") - - return Address(self.config, prefix, status, self).create(**kwargs) - - -class Address(object): - def __init__( - self, - config, - prefix=None, - status=None, - network=None, - scope_group=None, - id=None, - tags=None, - ): - """ - Create a new high level Address object - - :param ns1.config.Config config: config object - :param str prefix: cidr prefix - :param str status: planned, assigned - :param Network network: Network Object the address will be part of - :param Scopegroup scope_group: Scopegroup Object that will be associated with the address - :param dict tags: tags of the address - """ - self._rest = Addresses(config) - self.config = config - self.prefix = prefix - self.status = status - self.network = network - # self.scope_group = scope_group - self.id = id - self.children = [] - self.report = {} - self.data = None - self.tags = tags - - def __repr__(self): - return "
" % self.prefix - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload address data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load address data from the API. - """ - - if not reload and self.data: - raise AddressException("Address already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.prefix = result["prefix"] - self.status = result["status"] - self.network = Network(self.config, id=result["network_id"]) - # self.scope_group = Scopegroup(config=self.config, id=result['scope_group_id']) NYI - self.report = self._rest.report(self.id) - children = self._rest.retrieve_children(self.id) - - if "tags" in result: - self.tags = result["tags"] - - self.children = [ - Address(self.config, id=child["id"]) - for child in children - if len(children) > 0 - ] - try: - parent = self._rest.retrieve_parent(self.id) - self.parent = Address(self.config, id=parent["id"]) - except ResourceException: - pass - - if callback: - return callback(self) - else: - return self - - if self.id is None: - if ( - self.prefix is None - or self.status is None - or self.network is None - ): - raise AddressException( - "Must at least specify an id or prefix, status, and network" - ) - else: - network_id = self.network.id - try: - self.id = [ - address - for address in self._rest.list() - if address["prefix"] == self.prefix - and address["status"] == self.status - and address["network_id"] == network_id - ][0]["id"] - except IndexError: - raise AddressException( - "Could not find address by prefix. It may not exist, or is a child address. " - "Use the topmost parent prefix or specify ID" - ) - - return self._rest.retrieve(self.id, callback=success, errback=errback) - - def delete(self, callback=None, errback=None): - """ - Delete the address and all child addresses - """ - - return self._rest.delete(self.id, callback=callback, errback=errback) - - def update(self, callback=None, errback=None, parent=True, **kwargs): - """ - Update address configuration. Pass a list of keywords and their values to - update. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Addresses.INT_FIELDS` and :attr:`ns1.rest.ipam.Addresses.PASSTHRU_FIELDS` - """ - - if not self.data: - raise AddressException("Address not loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.prefix = result["prefix"] - self.status = result["status"] - self.network = Network(self.config, id=result["network_id"]) - # self.scope_group = Scopegroup(config=self.config, id=result['scope_group_id']) - self.report = self._rest.report(self.id) - children = self._rest.retrieve_children(self.id) - self.children = [ - Address(self.config, id=child["id"]) - for child in children - if len(children) > 0 - ] - try: - parent = self._rest.retrieve_parent(self.id) - self.parent = Address(self.config, id=parent["id"]) - except ResourceException: - pass - - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.id, callback=success, errback=errback, parent=parent, **kwargs - ) - - def reserve( - self, scopegroup_id, mac, options=None, callback=None, errback=None - ): - """ - Add scope group reservation. Pass a single Address object and a MAC address as a string - """ - - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - reservation = Reservation( - self.config, scopegroup_id, self.id, options=options, mac=mac - ) - - return reservation.create(callback=callback, errback=errback) - - def create(self, callback=None, errback=None, parent=True, **kwargs): - """ - Create a new Address. Pass a list of keywords and their values to - configure. For the list of keywords available for address configuration,see :attr:`ns1.rest.ipam.Addresses.INT_FIELDS` and :attr:`ns1.rest.ipam.Addresses.PASSTHRU_FIELDS` - """ - - if self.data: - raise AddressException("Address already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.prefix = result["prefix"] - self.status = result["status"] - self.network = Network(self.config, id=result["network_id"]) - # self.scope_group = Scopegroup(config=self.config, id=result['scope_group_id']) - self.report = self._rest.report(self.id) - children = self._rest.retrieve_children(self.id) - self.children = [ - Address(self.config, id=child["id"]) - for child in children - if len(children) > 0 - ] - try: - parent = self._rest.retrieve_parent(self.id) - self.parent = Address(self.config, id=parent["id"]) - except ResourceException: - pass - - if callback: - return callback(self) - else: - return self - - # if self.scope_group is not None: - # kwargs['scope_group_id'] = self.scope_group.id - - return self._rest.create( - prefix=self.prefix, - status=self.status, - network_id=self.network.id, - callback=success, - errback=errback, - parent=parent, - **kwargs - ) - - -class Scopegroup(object): - def __init__( - self, config, name=None, service_def_id=None, id=None, tags=None - ): - """ - Create a new high level Scopegroup object - - :param ns1.config.Config config: config object - :param str name: Name of the scope group - :param int service_group_id: id of the service group the scope group is associated with - :param int id: id of the scope group - :param dict tags: tags of the scopegroup - """ - self._rest = Scopegroups(config) - self.config = config - self.id = id - self.dhcp4 = [] - self.dhcp6 = [] - self.name = name - self.dhcp_service_id = service_def_id - self.data = None - self.tags = tags - - def __repr__(self): - return "" % self.name - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload Scopegroup data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load Scopegroup data from the API. - """ - - if not reload and self.data: - raise ScopegroupException("Scope Group already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.dhcp4 = result["dhcpv4"] - self.dhcp6 = result["dhcpv6"] - self.dhcp_service_id = result.get("dhcp_service_id") - - if "tags" in result: - self.tags = result["tags"] - - if callback: - return callback(self) - else: - return self - - if self.id is None: - raise ScopegroupException( - "Must at least specify an ID to load a Scopegroup" - ) - - return self._rest.retrieve(self.id, callback=success, errback=errback) - - def delete(self, callback=None, errback=None): - """ - Delete the Scopegroup and all child addresses - """ - - return self._rest.delete(self.id, callback=callback, errback=errback) - - def update(self, callback=None, errback=None, **kwargs): - """ - Update scope group configuration. Pass a list of keywords and their values to - update. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Scopegroups.INT_FIELDS` and :attr:`ns1.rest.ipam.Scopegroups.PASSTHRU_FIELDS` - """ - - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.dhcp4 = result["dhcpv4"] - self.dhcp6 = result["dhcpv6"] - self.dhcp_service_id = result.get("dhcp_service_id") - - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.id, callback=success, errback=errback, **kwargs - ) - - def create(self, dhcp4, dhcp6, callback=None, errback=None, **kwargs): - """ - :param DHCPOptions dhcp4: DHCPOptions object that contains the settings for dhcp4 - :param DHCPOptions dhcp6: DHCPOptions object that contains the settings for dhcp6 - - Create a new Scope Group. Pass a list of keywords and their values to - configure. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Scopegroups.INT_FIELDS` and :attr:`ns1.rest.ipam.Scopegroups.PASSTHRU_FIELDS`. - For the list of settings see :attr:`ns1.ipan.Scopegroup.SETTINGS`. Note that if `enabled` is True, then `valid_lifetime_secs` must be - set to a value greater than 0. - """ - - if self.data: - raise ScopegroupException("Scope Group already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.dhcp4 = result["dhcpv4"] - self.dhcp6 = result["dhcpv6"] - self.dhcp_service_id = result.get("dhcp_service_id") - - if callback: - return callback(self) - else: - return self - - if self.name is None: - raise ScopegroupException( - "Must at least specify an name to create a Scopegroup" - ) - - return self._rest.create( - dhcpv4=dhcp4.option_list, - dhcpv6=dhcp6.option_list, - name=self.name, - dhcp_service_id=self.dhcp_service_id, - callback=success, - errback=errback, - **kwargs - ) - - def reserve( - self, address_id, mac, options=None, callback=None, errback=None - ): - """ - :param int address_id: id of the Address to reserve - :param DHCPOptions options: DHCPOptions object that contains the settings for the address - :param mac str: MAC address of the reservation - - Add scope group reservation. Pass a single Address ID and a MAC address as a string - """ - - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - reservation = Reservation( - self.config, self.id, address_id, options=options, mac=mac - ) - - return reservation.create(callback=callback, errback=errback) - - def create_scope(self, address_id, callback=None, errback=None): - """ - Add scope group scope. Pass a single Address ID - """ - - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - scope = Scope(self.config, self.id, address_id) - - return scope.create(callback=callback, errback=errback) - - @property - def reservations(self, callback=None, errback=None): - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - reservations_config = Reservations(self.config) - - return reservations_config.list( - self.id, callback=callback, errback=errback - ) - - @property - def scopes(self, callback=None, errback=None): - if not self.data: - raise ScopegroupException("Scope Group not loaded") - - scopes_config = Scopes(self.config) - - return scopes_config.list(self.id, callback=callback, errback=errback) - - -class Reservation(object): - def __init__( - self, - config, - scopegroup_id, - address_id, - reservation_id=None, - options=None, - mac=None, - tags=None, - ): - """ - Create a new high level Reservation object - - :param ns1.config.Config config: config object - :param int scopegroup_id: id of the scope group - :param int address_id: id of the address the reservation is associated with - :param int reservation_id: id of the reservation - :param list options: dhcp options of the reservation - :param str mac: mac address of the reservation - :param dict tags: tags of the reservation - """ - self._rest = Reservations(config) - self.config = config - self.id = reservation_id - self.scopegroup_id = scopegroup_id - self.address_id = address_id - self.mac = mac - self.data = None - self.tags = tags - - if options is None: - options = DHCPOptions("dhcpv4", {}) - self.options = options.option_list["options"] - - def __repr__(self): - return "" % ( - self.scopegroup_id, - self.address_id, - self.mac, - ) - - def __getitem__(self, item): - if item == "scopegroup_id": - return self.scopegroup_id - - if item == "address_id": - return self.address_id - - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload Reservation data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load Reservation data from the API. - """ - - if not reload and self.data: - raise ReservationException("Reservation already loaded") - - def success(result, *args): - self.data = result - self.address_id = result["address_id"] - self.mac = result["mac"] - self.options = result["options"] - - if "tags" in result: - self.tags = result["tags"] - - if callback: - return callback(self) - else: - return self - - if self.id is None: - raise ReservationException("Must specify a reservation_id") - - return self._rest.retrieve(self.id, callback=success, errback=errback) - - def delete(self, callback=None, errback=None): - """ - Delete the Reservation - """ - - return self._rest.delete(self.id, callback=callback, errback=errback) - - def create(self, callback=None, errback=None, **kwargs): - """ - Create a new Reservation. Pass a list of keywords and their values to - configure. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Reservations.INT_FIELDS` and :attr:`ns1.rest.ipam.Reservations.PASSTHRU_FIELDS` - """ - - if self.data: - raise ReservationException("Reservation already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.address_id = result["address_id"] - self.mac = result["mac"] - self.options = result["options"] - - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.scopegroup_id, - self.address_id, - options=self.options, - mac=self.mac, - callback=success, - errback=errback, - **kwargs - ) - - def update( - self, options, callback=None, errback=None, parent=True, **kwargs - ): - """ - Update reservation configuration. Pass a list of keywords and their values to - update. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Reservations.INT_FIELDS` and :attr:`ns1.rest.ipam.Reservations.PASSTHRU_FIELDS` - """ - - if not self.data: - raise ReservationException("Reservation not loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.address_id = result["address_id"] - self.mac = result["mac"] - self.options = result["options"] - - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.id, - options, - callback=success, - errback=errback, - parent=parent, - **kwargs - ) - - -class Optiondef(object): - def __init__(self, config, space, key): - """ - Create a new high level Optiondef object - - :param ns1.config.Config config: config object - :param str space: dhcpv4 or dhcpv6 - :param str key: option key - """ - self._rest = Optiondefs(config) - self.config = config - self.space = space - self.key = key - self.data = None - - def __repr__(self): - return "" % (self.space, self.key) - - def __getitem__(self, item): - if item == "space": - return self.space - - if item == "key": - return self.key - - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload OptionDef data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load Optiondef data from the API. - """ - - if not reload and self.data: - raise ReservationException("Optiondef already loaded") - - def success(result, *args): - self.data = result - self.space = result["space"] - self.key = result["key"] - self.code = result["code"] - self.friendly_name = result["friendly_name"] - - if callback: - return callback(self) - else: - return self - - return self._rest.retrieve( - self.space, self.key, callback=success, errback=errback - ) - - def delete(self, callback=None, errback=None): - """ - Delete the Optiondef - """ - - return self._rest.delete( - self.space, self.key, callback=callback, errback=errback - ) - - def create(self, callback=None, errback=None, **kwargs): - """ - Create a new Optiondef. Pass a list of keywords and their values to - configure. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Optiondef.INT_FIELDS` and :attr:`ns1.rest.ipam.Optiondef.PASSTHRU_FIELDS` - """ - - if self.data: - raise OptiondefException("Optiondef already loaded") - - def success(result, *args): - self.data = result - self.space = result["space"] - self.key = result["key"] - self.code = result["code"] - self.friendly_name = result["friendly_name"] - - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.space, self.key, callback=success, errback=errback, **kwargs - ) - - -class Scope(object): - def __init__( - self, - config, - scopegroup_id, - address_id, - scope_id=None, - options=None, - tags=None, - ): - """ - Create a new high level Scope object - - :param ns1.config.Config config: config object - :param int scopegroup_id: id of the scope group - :param int address_id: id of the address the scope is associated with - :param int scope_id: id of the scope - :param DHCPOptions options: DHCPOptions object that contains the settings for the scope - :param dict tags: tags of the scope - """ - self._rest = Scopes(config) - self.config = config - self.scopegroup_id = scopegroup_id - self.address_id = address_id - self.id = scope_id - self.tags = tags - - if options is None: - options = DHCPOptions("dhcpv4", {}) - self.options = options.option_list["options"] - - self.data = None - - def __repr__(self): - return "" % ( - self.scopegroup_id, - self.address_id, - ) - - def __getitem__(self, item): - if item == "scopegroup_id": - return self.scopegroup_id - - if item == "address_id": - return self.address_id - - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - """ - Reload Scope data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - """ - Load Reservation data from the API. - """ - - if not reload and self.data: - raise ScopeException("Scope already loaded") - - def success(result, *args): - self.data = result - self.address_id = result["address_id"] - self.options = result["options"] - - if "tags" in result: - self.tags = result["tags"] - - if callback: - return callback(self) - else: - return self - - if self.id is None: - raise ScopeException("Must specify a scope_id") - - return self._rest.retrieve(self.id, callback=success, errback=errback) - - def delete(self, callback=None, errback=None): - """ - Delete the Scope - """ - - return self._rest.delete(self.id, callback=callback, errback=errback) - - def create(self, callback=None, errback=None, **kwargs): - """ - Create a new Scope. Pass a list of keywords and their values to - configure. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Scope.INT_FIELDS` and :attr:`ns1.rest.ipam.Reservations.PASSTHRU_FIELDS` - """ - - if self.data: - raise ScopeException("Scope already loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.address_id = result["address_id"] - self.options = result["options"] - - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.scopegroup_id, - self.address_id, - self.options, - callback=success, - errback=errback, - **kwargs - ) - - def update( - self, address_id, options, callback=None, errback=None, **kwargs - ): - """ - Update Scope configuration. Pass a list of keywords and their values to - update. For the list of keywords available for address configuration, see :attr:`ns1.rest.ipam.Scopes.INT_FIELDS` and :attr:`ns1.rest.ipam.Scopes.PASSTHRU_FIELDS` - """ - - if not self.data: - raise ScopeException("Scope not loaded") - - def success(result, *args): - self.data = result - self.id = result["id"] - self.address_id = result["address_id"] - self.options = result["options"] - - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.id, - address_id, - options, - callback=success, - errback=errback, - **kwargs - ) - - -class Lease(object): - def __init__(self, config): - """ - Create a new high level Lease object - - :param ns1.config.Config config: config object - """ - self._rest = Leases(config) - self.config = config - self.leases = None - self.data = None - - def __repr__(self): - return "" - - def reload(self, callback=None, errback=None): - """ - Reload Lease data from the API. - """ - - return self.load(reload=True, callback=callback, errback=errback) - - def load( - self, - scope_group_id=None, - scope_id=None, - limit=None, - offset=None, - callback=None, - errback=None, - reload=False, - ): - """ - Load Lease data from the API. - """ - - if not reload and self.data: - raise LeaseException("Lease already loaded") - - def success(result, *args): - self.data = result - self.leases = self.data - - if callback: - return callback(self) - else: - return self - - return self._rest.list( - scope_group_id, - scope_id, - limit, - offset, - callback=success, - errback=errback, - ) - - -class DHCPOptions: - AF = ["dhcpv4", "dhcpv6"] - OPTIONS = { - "dhcpv4": [ - "bootfile-name", - "domain-name", - "domain-name-servers", - "host-name", - "routers", - "tftp-server-name", - "time-servers", - "vendor-class-identifier", - ], - "dhcpv6": ["dns-servers"], - } - - def __init__(self, address_family, options, server_options=None): - """ - Create the DHCP options class that can be used by the IPAM API - - :param str address_family: This is either dhcpv4 or dhcpv6 - :param list options: This is a list of :class:`ns1.ipam.DHCPOptionsValue` objects representing the DHCP options - """ - self.option_list = {} - self.address_family = "" - self.options = [] - - if server_options is None: - self.server_options = {} - else: - self.server_options = server_options - - self.update(address_family, options, self.server_options) - - def __repr__(self): - return "" % self.address_family - - def update(self, address_family, options, server_options=None): - if server_options is not None: - self.server_options = server_options - - self.address_family = address_family - self.options = options - self.__dict__.update(server_options) - - self.option_list = { - "options": [ - option.generate_option(address_family) - for option in self.options - ] - } - - self.option_list.update(self.server_options) - - -class DHCPOptionValue: - def __init__(self, key, value, always_send=None): - """ - Create the DHCPOptionValue class that can be used as value with :class:`ns1.ipam.DHCPOptions` - - :param key str option name - :param value any option value - :param always_send bool indicates whether this option be sent back in lease or not - """ - self.key = key - self.value = value - self.always_send = always_send - - def generate_option(self, address_family): - """ - Generates dhcp option value with a proper format - - :param address_family str one of dhcpv4 or dhcpv6 family name - """ - option = { - "name": "%s/%s" % (address_family, self.key), - "value": self.value, - } - if self.always_send is not None: - option["always_send"] = self.always_send - - return option diff --git a/ns1/pools.py b/ns1/pools.py deleted file mode 100644 index 25e0e35..0000000 --- a/ns1/pools.py +++ /dev/null @@ -1,86 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# -from ns1.rest.pools import Pools - - -class PoolException(Exception): - pass - - -class Pool(object): - def __init__(self, config, address, pool): - self._rest = Pools(config) - self.config = config - self.address = address - self.pool = pool - self.data = None - - def __repr__(self): - return "" % self.pool - - def __getitem__(self, item): - return self.data.get(item, None) - - def reload(self, callback=None, errback=None): - return self.load(reload=True, callback=callback, errback=errback) - - def load(self, callback=None, errback=None, reload=False): - if not reload and self.data: - raise PoolException("pool already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.retrieve( - self.address, self.pool, callback=success, errback=errback - ) - - def delete(self, callback=None, errback=None): - return self._rest.delete( - self.address, self.pool, callback=callback, errback=errback - ) - - def update(self, callback=None, errback=None, **kwargs): - if not self.data: - raise PoolException("pool not loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.update( - self.address, - self.pool, - callback=success, - errback=errback, - **kwargs - ) - - def create(self, callback=None, errback=None, **kwargs): - if self.data: - raise PoolException("pool already loaded") - - def success(result, *args): - self.data = result - if callback: - return callback(self) - else: - return self - - return self._rest.create( - self.address, - self.pool, - callback=success, - errback=errback, - **kwargs - ) diff --git a/ns1/rest/apikey.py b/ns1/rest/apikey.py index 2a81e3b..2ad91e3 100644 --- a/ns1/rest/apikey.py +++ b/ns1/rest/apikey.py @@ -21,10 +21,7 @@ def create(self, name, callback=None, errback=None, **kwargs): body = {"name": name} if "permissions" not in kwargs: - if self._config["ddi"]: - body["permissions"] = permissions._default_perms_ddi - else: - body["permissions"] = permissions._default_perms + body["permissions"] = permissions._default_perms self._buildStdBody(body, kwargs) diff --git a/ns1/rest/client_classes.py b/ns1/rest/client_classes.py deleted file mode 100644 index 1bc7fda..0000000 --- a/ns1/rest/client_classes.py +++ /dev/null @@ -1,73 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. - -from . import resource - - -class ClientClasses(resource.BaseResource): - ROOT = "dhcp" - CLIENT_CLASS_ROOT = "clientclass" - - INT_FIELDS = ["clientclassID"] - PASSTHRU_FIELDS = [ - "name", - "test", - "options", - "next_server", - "server_hostname", - "boot_file_name", - ] - - def _buildBody(self, client_class, **kwargs): - body = {} - body["client_class"] = client_class - self._buildStdBody(body, kwargs) - return body - - def create(self, client_class, callback=None, errback=None, **kwargs): - body = self._buildBody(client_class, **kwargs) - - return self._make_request( - "PUT", - "%s/%s" % (self.ROOT, self.CLIENT_CLASS_ROOT), - body=body, - callback=callback, - errback=errback, - ) - - def update(self, client_class_id, callback=None, errback=None, **kwargs): - body = self._buildBody(client_class_id, **kwargs) - - return self._make_request( - "POST", - "%s/%s/%s" % (self.ROOT, self.CLIENT_CLASS_ROOT, client_class_id), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, client_class_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s/%s" % (self.ROOT, self.CLIENT_CLASS_ROOT, client_class_id), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s" % (self.ROOT, self.CLIENT_CLASS_ROOT), - callback=callback, - errback=errback, - ) - - def retrieve(self, client_class_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/%s" % (self.ROOT, self.CLIENT_CLASS_ROOT, client_class_id), - callback=callback, - errback=errback, - ) diff --git a/ns1/rest/dhcp_option_spaces.py b/ns1/rest/dhcp_option_spaces.py deleted file mode 100644 index 2570de5..0000000 --- a/ns1/rest/dhcp_option_spaces.py +++ /dev/null @@ -1,59 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# - -from . import resource - - -class DHCPOptionSpaces(resource.BaseResource): - ROOT = "dhcp" - OPTION_SPACE_ROOT = "optionspace" - - PASSTHRU_FIELDS = [ - "name", - ] - BOOL_FIELDS = ["standard"] - - def _buildBody(self, dhcp, **kwargs): - body = {} - body["dhcp"] = dhcp - self._buildStdBody(body, kwargs) - return body - - def create(self, dhcp, callback=None, errback=None, **kwargs): - body = self._buildBody(dhcp, **kwargs) - return self._make_request( - "PUT", - "%s/%s" % (self.ROOT, self.OPTION_SPACE_ROOT), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, dhcp_option_space_name, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s/%s" - % (self.ROOT, self.OPTION_SPACE_ROOT, dhcp_option_space_name), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s" % (self.ROOT, self.OPTION_SPACE_ROOT), - callback=callback, - errback=errback, - ) - - def retrieve(self, dhcp_option_space_name, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/%s" - % (self.ROOT, self.OPTION_SPACE_ROOT, dhcp_option_space_name), - callback=callback, - errback=errback, - ) diff --git a/ns1/rest/ipam.py b/ns1/rest/ipam.py deleted file mode 100644 index cd1caf4..0000000 --- a/ns1/rest/ipam.py +++ /dev/null @@ -1,516 +0,0 @@ -# -# Copyright (c) 2019 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. -# - -from . import resource - - -class Addresses(resource.BaseResource): - ROOT = "ipam/address" - INT_FIELDS = [ - "network_id", - "address_id", - "root_address_id", - " merged_address_id", - "scope_group_id", - ] - PASSTHRU_FIELDS = ["prefix", "status", "desc", "tags", "reserve"] - BOOL_FIELDS = ["parent"] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - return body - - def create(self, callback=None, errback=None, parent=True, **kwargs): - body = self._buildBody(**kwargs) - params = {"parent": parent} - return self._make_request( - "PUT", - self.ROOT, - params=params, - body=body, - callback=callback, - errback=errback, - ) - - def update( - self, address_id, callback=None, errback=None, parent=True, **kwargs - ): - body = self._buildBody(**kwargs) - params = {"parent": parent} - return self._make_request( - "POST", - "%s/%s" % (self.ROOT, address_id), - params=params, - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, address_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None): - return self._make_request( - "GET", "%s" % self.ROOT, callback=callback, errback=errback - ) - - def retrieve(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - def retrieve_children(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/children" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - def retrieve_parent(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/parent" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - def report(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/report" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - def retrieve_dhcp_option(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/options" % (self.ROOT, address_id), - callback=callback, - errback=errback, - ) - - # NYI in API - # def retrieve_next(self, address_id, callback=None, errback=None): - # return self._make_request('GET', '%s/%s/next' % (self.ROOT, address_id), - # callback=callback, - # errback=errback) - # - # def retrieve_previous(self, address_id, callback=None, errback=None): - # return self._make_request('GET', '%s/%s/previous' % (self.ROOT, address_id), - # callback=callback, - # errback=errback) - # - # def retrieve_available(self, address_id, callback=None, errback=None): - # return self._make_request('GET', '%s/%s/available' % (self.ROOT, address_id), - # callback=callback, - # errback=errback) - # - # def merge(self, callback=None, errback=None, force=True, **kwargs): - # body = self._buildBody(**kwargs) - # params = {'force': force} - # return self._make_request('POST', - # '%s/merge' % self.ROOT, - # params=params, - # body=body, - # callback=callback, - # errback=errback) - # - # def split(self, address_id, callback=None, errback=None, **kwargs): - # body = self._buildBody(**kwargs) - # return self._make_request('POST', - # '%s/%s/split' % (self.ROOT, address_id), - # body=body, - # callback=callback, - # errback=errback) - - def search(self, network_id, prefix, callback=None, errback=None): - return self._make_request( - "GET", - "%s/search/%s/%s" % (self.ROOT, network_id, prefix), - callback=callback, - errback=errback, - ) - - -class Networks(resource.BaseResource): - ROOT = "ipam/network" - INT_FIELDS = ["network_id"] - PASSTHRU_FIELDS = ["rt", "name", "desc", "tags"] - BOOL_FIELDS = [] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - return body - - def create(self, callback=None, errback=None, **kwargs): - body = self._buildBody(**kwargs) - return self._make_request( - "PUT", self.ROOT, body=body, callback=callback, errback=errback - ) - - def update(self, network_id, callback=None, errback=None, **kwargs): - body = self._buildBody(**kwargs) - return self._make_request( - "POST", - "%s/%s" % (self.ROOT, network_id), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, network_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s" % (self.ROOT, network_id), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None, expand=True): - params = {"expand": expand} - return self._make_request( - "GET", self.ROOT, params=params, callback=callback, errback=errback - ) - - def retrieve(self, network_id, callback=None, errback=None, expand=True): - params = {"expand": expand} - return self._make_request( - "GET", - "%s/%s" % (self.ROOT, network_id), - params=params, - callback=callback, - errback=errback, - ) - - def report(self, network_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/report" % (self.ROOT, network_id), - callback=callback, - errback=errback, - ) - - # def search(self, q=None, callback=None, errback=None): - # params = {} - # if q is not None: - # params['q'] = q - # return self._make_request('GET', self.SEARCH_ROOT, - # params=params, - # callback=callback, - # errback=errback) - - -class Scopegroups(resource.BaseResource): - ROOT = "dhcp/scopegroup" - INT_FIELDS = ["id", "dhcp_service_id", "valid_lifetime_secs"] - PASSTHRU_FIELDS = ["dhcpv4", "dhcpv6", "name", "tags"] - BOOL_FIELDS = ["enabled", "echo_client_id"] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - eci = body.get("echo_client_id") - if eci is not None: - del body["echo_client_id"] - dhcpv4 = body.get("dhcpv4", {}) - dhcpv4["echo_client_id"] = eci - body["dhcpv4"] = dhcpv4 - return body - - def create(self, callback=None, errback=None, **kwargs): - body = self._buildBody(**kwargs) - return self._make_request( - "PUT", self.ROOT, body=body, callback=callback, errback=errback - ) - - def update(self, scope_group_id, callback=None, errback=None, **kwargs): - body = self._buildBody(**kwargs) - return self._make_request( - "POST", - "%s/%s" % (self.ROOT, scope_group_id), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, scopegroup_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s" % (self.ROOT, scopegroup_id), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None): - return self._make_request( - "GET", self.ROOT, callback=callback, errback=errback - ) - - def retrieve(self, scope_group_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s" % (self.ROOT, scope_group_id), - callback=callback, - errback=errback, - ) - - -class Scopes(resource.BaseResource): - ROOT = "dhcp/scope" - INT_FIELDS = ["scope_group_id", "address_id", "valid_lifetime_secs"] - PASSTHRU_FIELDS = ["options", "tags"] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - return body - - @classmethod - def select_from_list(cls, result, scope_id): - for item in result: - if item.get("id") == int(scope_id): - return item - return None - - def create( - self, - scopegroup_id, - address_id, - options, - callback=None, - errback=None, - **kwargs - ): - kwargs["address_id"] = address_id - kwargs["scope_group_id"] = scopegroup_id - kwargs["options"] = options - body = self._buildBody(**kwargs) - - return self._make_request( - "PUT", - "%s" % (self.ROOT), - body=body, - callback=callback, - errback=errback, - ) - - def update( - self, - scope_id, - address_id, - options, - scopegroup_id=None, - callback=None, - errback=None, - **kwargs - ): - kwargs["address_id"] = address_id - kwargs["options"] = options - - if scopegroup_id is not None: - kwargs["scope_group_id"] = scopegroup_id - - body = self._buildBody(**kwargs) - - return self._make_request( - "POST", - "%s/%s" % (self.ROOT, scope_id), - body=body, - callback=callback, - errback=errback, - ) - - def list(self, scopegroup_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s?scopeGroupId=%d" % (self.ROOT, int(scopegroup_id)), - callback=callback, - errback=errback, - ) - - def delete(self, scope_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s" % (self.ROOT, scope_id), - callback=callback, - errback=errback, - ) - - def retrieve(self, scope_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%d" % (self.ROOT, int(scope_id)), - callback=callback, - errback=errback, - ) - - -class Leases(resource.BaseResource): - ROOT = "dhcp/lease" - INT_FIELDS = ["scope_group_id", "scope_id", "limit", "offset"] - PASSTHRU_FIELDS = [] - BOOL_FIELDS = [] - - def list( - self, - scope_group_id=None, - scope_id=None, - limit=None, - offset=None, - callback=None, - errback=None, - ): - params = {} - if scope_group_id is not None: - params["scopeGroupId"] = scope_group_id - if scope_id is not None: - params["scopeId"] = scope_id - if limit is not None: - params["limit"] = limit - if offset is not None: - params["offset"] = offset - return self._make_request( - "GET", self.ROOT, callback=callback, errback=errback, params=params - ) - - -class Optiondefs(resource.BaseResource): - ROOT = "dhcp/optiondef" - INT_FIELDS = ["code"] - PASSTHRU_FIELDS = [ - "space", - "key", - "friendly_name", - "description", - "schema", - ] - BOOL_FIELDS = ["standard"] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - return body - - def create(self, space, key, callback=None, errback=None, **kwargs): - body = self._buildBody(**kwargs) - body["space"] = space - body["key"] = key - return self._make_request( - "PUT", - "%s/%s/%s" % (self.ROOT, space, key), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, space, key, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s/%s" % (self.ROOT, space, key), - callback=callback, - errback=errback, - ) - - def list(self, callback=None, errback=None): - return self._make_request( - "GET", self.ROOT, callback=callback, errback=errback - ) - - def retrieve(self, space, key, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/%s" % (self.ROOT, space, key), - callback=callback, - errback=errback, - ) - - -class Reservations(resource.BaseResource): - ROOT = "dhcp/reservation" - INT_FIELDS = ["scope_group_id", "address_id"] - PASSTHRU_FIELDS = ["mac", "options", "tags"] - BOOL_FIELDS = ["dhcpv6"] - - def _buildBody(self, **kwargs): - body = {} - self._buildStdBody(body, kwargs) - return body - - @classmethod - def select_from_list(cls, result, address_id): - for item in result: - if item.get("address_id") == int(address_id): - return item - return None - - def create( - self, - scopegroup_id, - address_id, - options, - callback=None, - errback=None, - **kwargs - ): - kwargs["address_id"] = address_id - kwargs["scope_group_id"] = scopegroup_id - kwargs["options"] = options - body = self._buildBody(**kwargs) - - reservation = self._make_request( - "PUT", self.ROOT, body=body, callback=callback, errback=errback - ) - return reservation - - def update( - self, reservation_id, options, callback=None, errback=None, **kwargs - ): - kwargs["options"] = options - body = self._buildBody(**kwargs) - - return self._make_request( - "POST", - "%s/%s" % (self.ROOT, reservation_id), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, reservation_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s" % (self.ROOT, reservation_id), - callback=callback, - errback=errback, - ) - - def list(self, scopegroup_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s?scopeGroupId=%d" % (self.ROOT, int(scopegroup_id)), - callback=callback, - errback=errback, - ) - - def retrieve(self, reservation_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%d" % (self.ROOT, int(reservation_id)), - callback=callback, - errback=errback, - ) diff --git a/ns1/rest/permissions.py b/ns1/rest/permissions.py index 6526f4a..5c19616 100644 --- a/ns1/rest/permissions.py +++ b/ns1/rest/permissions.py @@ -37,36 +37,3 @@ "zones_allow_by_default": False, }, } - -_default_perms_ddi = { - "data": { - "push_to_datafeeds": False, - "manage_datasources": False, - "manage_datafeeds": False, - }, - "account": { - "manage_users": False, - "manage_teams": False, - "manage_account_settings": False, - "manage_apikeys": False, - "view_activity_log": False, - }, - "monitoring": { - "manage_jobs": False, - "create_jobs": False, - "update_jobs": False, - "delete_jobs": False, - "manage_lists": False, - "view_jobs": False, - }, - "security": {"manage_global_2fa": False, "manage_active_directory": False}, - "dns": { - "zones_allow": [], - "manage_zones": False, - "zones_deny": [], - "view_zones": False, - "zones_allow_by_default": False, - }, - "dhcp": {"manage_dhcp": False, "view_dhcp": False}, - "ipam": {"manage_ipam": False, "view_ipam": False}, -} diff --git a/ns1/rest/pools.py b/ns1/rest/pools.py deleted file mode 100644 index f51c5ee..0000000 --- a/ns1/rest/pools.py +++ /dev/null @@ -1,100 +0,0 @@ -# -# Copyright (c) 2014 NSONE, Inc. -# -# License under The MIT License (MIT). See LICENSE in project root. - -from . import resource - - -class Pools(resource.BaseResource): - ROOT = "ipam" - ADDRESS_ROOT = "address" - POOL_ROOT = "pool" - - INT_FIELDS = ["addressID", "poolID"] - PASSTHRU_FIELDS = [ - "range", - "options", - "client_class", - "required_client_class", - "tags", - "blocked_tags", - ] - - def _buildBody(self, address=None, pool=None, **kwargs): - body = {} - body["address"] = address - body["pool"] = pool - self._buildStdBody(body, kwargs) - return body - - def create(self, address_id, callback=None, errback=None, **kwargs): - body = self._buildBody(address_id, **kwargs) - - return self._make_request( - "PUT", - "%s/%s/%s/%s" - % (self.ROOT, self.ADDRESS_ROOT, address_id, self.POOL_ROOT), - body=body, - callback=callback, - errback=errback, - ) - - def update( - self, address_id, pool_id, callback=None, errback=None, **kwargs - ): - body = self._buildBody(address_id, pool_id, **kwargs) - - return self._make_request( - "POST", - "%s/%s/%s/%s/%s" - % ( - self.ROOT, - self.ADDRESS_ROOT, - address_id, - self.POOL_ROOT, - pool_id, - ), - body=body, - callback=callback, - errback=errback, - ) - - def delete(self, address_id, pool_id, callback=None, errback=None): - return self._make_request( - "DELETE", - "%s/%s/%s/%s/%s" - % ( - self.ROOT, - self.ADDRESS_ROOT, - address_id, - self.POOL_ROOT, - pool_id, - ), - callback=callback, - errback=errback, - ) - - def list(self, address_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/%s/%s" - % (self.ROOT, self.ADDRESS_ROOT, address_id, self.POOL_ROOT), - callback=callback, - errback=errback, - ) - - def retrieve(self, address_id, pool_id, callback=None, errback=None): - return self._make_request( - "GET", - "%s/%s/%s/%s/%s" - % ( - self.ROOT, - self.ADDRESS_ROOT, - address_id, - self.POOL_ROOT, - pool_id, - ), - callback=callback, - errback=errback, - ) diff --git a/ns1/rest/team.py b/ns1/rest/team.py index 066180a..a338934 100644 --- a/ns1/rest/team.py +++ b/ns1/rest/team.py @@ -15,10 +15,7 @@ def create(self, name, callback=None, errback=None, **kwargs): body = {"name": name} if "permissions" not in kwargs: - if self._config["ddi"]: - body["permissions"] = permissions._default_perms_ddi - else: - body["permissions"] = permissions._default_perms + body["permissions"] = permissions._default_perms self._buildStdBody(body, kwargs) diff --git a/ns1/rest/tsig.py b/ns1/rest/tsig.py index e330465..d68869a 100644 --- a/ns1/rest/tsig.py +++ b/ns1/rest/tsig.py @@ -29,10 +29,7 @@ def create( ): body = {"algorithm": algorithm, "secret": secret} if "permissions" not in kwargs: - if self._config["ddi"]: - body["permissions"] = permissions._default_perms_ddi - else: - body["permissions"] = permissions._default_perms + body["permissions"] = permissions._default_perms self._buildStdBody(body, kwargs) diff --git a/ns1/rest/user.py b/ns1/rest/user.py index f1b1e1d..b6a707c 100644 --- a/ns1/rest/user.py +++ b/ns1/rest/user.py @@ -26,10 +26,7 @@ def create( body = {"name": name, "username": username, "email": email} if "permissions" not in kwargs: - if self._config["ddi"]: - body["permissions"] = permissions._default_perms_ddi - else: - body["permissions"] = permissions._default_perms + body["permissions"] = permissions._default_perms self._buildStdBody(body, kwargs) diff --git a/tests/unit/test_address.py b/tests/unit/test_address.py deleted file mode 100644 index 5169385..0000000 --- a/tests/unit/test_address.py +++ /dev/null @@ -1,105 +0,0 @@ -import pytest - -import ns1.rest.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def address_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_address_list(address_config): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.list() - z._make_request.assert_called_once_with( - "GET", "ipam/address", callback=None, errback=None - ) - - -@pytest.mark.parametrize("address_id, url", [("1", "ipam/address/1")]) -def test_rest_address_retrieve(address_config, address_id, url): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.retrieve(address_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize("address_id, url", [("1", "ipam/address/1/report")]) -def test_rest_address_report(address_config, address_id, url): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.report(address_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "prefix, network_id, address_status, url", - [("192.168.0.0/24", 1, "planned", "ipam/address")], -) -def test_rest_address_create( - address_config, prefix, network_id, address_status, url -): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.create(prefix=prefix, network_id=network_id, status=address_status) - z._make_request.assert_called_once_with( - "PUT", - url, - callback=None, - errback=None, - params={"parent": True}, - body={ - "network_id": network_id, - "prefix": prefix, - "status": address_status, - }, - ) - - -@pytest.mark.parametrize( - "address_id, prefix, url", [("1", "192.168.0.0/24", "ipam/address/1")] -) -def test_rest_address_update(address_config, address_id, prefix, url): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.update(1, prefix=prefix) - z._make_request.assert_called_once_with( - "POST", - url, - callback=None, - errback=None, - body={"prefix": prefix}, - params={"parent": True}, - ) - - -@pytest.mark.parametrize("address_id, url", [("1", "ipam/address/1")]) -def test_rest_address_delete(address_config, address_id, url): - z = ns1.rest.ipam.Addresses(address_config) - z._make_request = mock.MagicMock() - z.delete(address_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_client_classes.py b/tests/unit/test_client_classes.py deleted file mode 100644 index 01faff0..0000000 --- a/tests/unit/test_client_classes.py +++ /dev/null @@ -1,92 +0,0 @@ -import pytest - -import ns1.rest.client_classes - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def client_class_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_client_class_list(client_class_config): - z = ns1.rest.client_classes.ClientClasses(client_class_config) - z._make_request = mock.MagicMock() - z.list() - z._make_request.assert_called_once_with( - "GET", - "dhcp/clientclass", - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize("client_class_id, url", [("1", "dhcp/clientclass/1")]) -def test_rest_client_class_retrieve(client_class_config, client_class_id, url): - z = ns1.rest.client_classes.ClientClasses(client_class_config) - z._make_request = mock.MagicMock() - z.retrieve(client_class_id=client_class_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize("client_class_id, url", [("1", "dhcp/clientclass")]) -def test_rest_client_class_create(client_class_config, client_class_id, url): - z = ns1.rest.client_classes.ClientClasses(client_class_config) - z._make_request = mock.MagicMock() - z.create(client_class=client_class_id) - z._make_request.assert_called_once_with( - "PUT", - url, - body={ - "client_class": client_class_id, - }, - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize( - "client_class_id, url", - [("1", "dhcp/clientclass/1")], -) -def test_rest_client_class_update(client_class_config, client_class_id, url): - z = ns1.rest.client_classes.ClientClasses(client_class_config) - z._make_request = mock.MagicMock() - z.update(client_class_id=client_class_id) - z._make_request.assert_called_once_with( - "POST", - url, - callback=None, - errback=None, - body={ - "client_class": client_class_id, - }, - ) - - -@pytest.mark.parametrize("client_class_id, url", [("1", "dhcp/clientclass/1")]) -def test_rest_client_class_delete(client_class_config, client_class_id, url): - z = ns1.rest.client_classes.ClientClasses(client_class_config) - z._make_request = mock.MagicMock() - z.delete(client_class_id=client_class_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_dhcp_option_spaces.py b/tests/unit/test_dhcp_option_spaces.py deleted file mode 100644 index fc3f041..0000000 --- a/tests/unit/test_dhcp_option_spaces.py +++ /dev/null @@ -1,87 +0,0 @@ -import pytest - -import ns1.rest.dhcp_option_spaces - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def dhcp_option_space_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_dhcp_option_space_list(dhcp_option_space_config): - z = ns1.rest.dhcp_option_spaces.DHCPOptionSpaces(dhcp_option_space_config) - z._make_request = mock.MagicMock() - z.list() - z._make_request.assert_called_once_with( - "GET", - "dhcp/optionspace", - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize( - "dhcp_option_space_name, url", - [("test_dhcp", "dhcp/optionspace/test_dhcp")], -) -def test_rest_dhcp_option_space_retrieve( - dhcp_option_space_config, dhcp_option_space_name, url -): - z = ns1.rest.dhcp_option_spaces.DHCPOptionSpaces(dhcp_option_space_config) - z._make_request = mock.MagicMock() - z.retrieve(dhcp_option_space_name=dhcp_option_space_name) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "dhcp_option_space_name, url", [("test_dhcp", "dhcp/optionspace")] -) -def test_rest_dhcp_option_space_create( - dhcp_option_space_config, dhcp_option_space_name, url -): - z = ns1.rest.dhcp_option_spaces.DHCPOptionSpaces(dhcp_option_space_config) - z._make_request = mock.MagicMock() - z.create(dhcp=dhcp_option_space_name) - z._make_request.assert_called_once_with( - "PUT", - url, - body={ - "dhcp": dhcp_option_space_name, - }, - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize( - "dhcp_option_space_name, url", - [("test_dhcp", "dhcp/optionspace/test_dhcp")], -) -def test_rest_dhcp_option_space_delete( - dhcp_option_space_config, dhcp_option_space_name, url -): - z = ns1.rest.dhcp_option_spaces.DHCPOptionSpaces(dhcp_option_space_config) - z._make_request = mock.MagicMock() - z.delete(dhcp_option_space_name=dhcp_option_space_name) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_init.py b/tests/unit/test_init.py index 7d58542..ffa20d5 100644 --- a/tests/unit/test_init.py +++ b/tests/unit/test_init.py @@ -4,14 +4,6 @@ from ns1.rest.account import Plan from ns1.rest.apikey import APIKey from ns1.rest.data import Feed, Source -from ns1.rest.ipam import ( - Addresses, - Networks, - Reservations, - Scopegroups, - Scopes, - Optiondefs, -) from ns1.rest.monitoring import JobTypes, Monitors, NotifyLists, Regions from ns1.rest.records import Records from ns1.rest.stats import Stats @@ -43,12 +35,6 @@ def ns1_config(config): [ ("zones", Zones), ("records", Records), - ("addresses", Addresses), - ("networks", Networks), - ("scope_groups", Scopegroups), - ("reservations", Reservations), - ("scopes", Scopes), - ("optiondefs", Optiondefs), ("stats", Stats), ("datasource", Source), ("datafeed", Feed), diff --git a/tests/unit/test_ipam.py b/tests/unit/test_ipam.py deleted file mode 100644 index 4cecac6..0000000 --- a/tests/unit/test_ipam.py +++ /dev/null @@ -1,82 +0,0 @@ -import pytest - -import ns1.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def reservation_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_reservation_load(reservation_config): - z = ns1.ipam.Reservation( - reservation_config, "scopegroup_id", "address_id", "reservation_id" - ) - z._rest = mock.MagicMock() - z.load() - - args, kwargs = z._rest.retrieve.call_args_list[0] - assert args == (z.id,) - assert len(kwargs) == 2 - assert kwargs["callback"] is not None - assert kwargs["errback"] is None - - -def test_reservation_create(reservation_config): - z = ns1.ipam.Reservation( - reservation_config, "scopegroup_id", "address_id", "reservation_id" - ) - z._rest = mock.MagicMock() - z.create() - - args, kwargs = z._rest.create.call_args_list[0] - assert args == (z.scopegroup_id, z.address_id) - assert len(kwargs) == 4 - assert kwargs["options"] == z.options - assert kwargs["mac"] == z.mac - assert kwargs["callback"] is not None - assert kwargs["errback"] is None - - -def test_reservation_delete(reservation_config): - z = ns1.ipam.Reservation( - reservation_config, "scopegroup_id", "address_id", "reservation_id" - ) - z._rest = mock.MagicMock() - z.delete() - - assert z.id == "reservation_id" - z._rest.delete.assert_called_once_with(z.id, callback=None, errback=None) - - -def test_reservation_update(reservation_config): - z = ns1.ipam.Reservation( - reservation_config, "scopegroup_id", "address_id", "reservation_id" - ) - z._rest = mock.MagicMock() - z.data = "my_data" - z.update("my_options") - - args, kwargs = z._rest.update.call_args_list[0] - assert args == (z.id, "my_options") - assert len(kwargs) == 3 - assert kwargs["callback"] is not None - assert kwargs["errback"] is None - assert kwargs["parent"] is True diff --git a/tests/unit/test_network.py b/tests/unit/test_network.py deleted file mode 100644 index 8ef3b36..0000000 --- a/tests/unit/test_network.py +++ /dev/null @@ -1,93 +0,0 @@ -import pytest - -import ns1.rest.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def network_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_network_list(network_config): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.list() - z._make_request.assert_called_once_with( - "GET", - "ipam/network", - callback=None, - errback=None, - params={"expand": True}, - ) - - -@pytest.mark.parametrize("network_id, url", [("1", "ipam/network/1")]) -def test_rest_network_retrieve(network_config, network_id, url): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.retrieve(network_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None, params={"expand": True} - ) - - -@pytest.mark.parametrize("network_id, url", [("1", "ipam/network/1/report")]) -def test_rest_network_report(network_config, network_id, url): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.report(network_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "network_name, url", [("test_network", "ipam/network")] -) -def test_rest_network_create(network_config, network_name, url): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.create(name=network_name) - z._make_request.assert_called_once_with( - "PUT", url, callback=None, errback=None, body={"name": network_name} - ) - - -@pytest.mark.parametrize( - "network_id, network_desc, url", - [("1", "awesome network", "ipam/network/1")], -) -def test_rest_network_update(network_config, network_id, network_desc, url): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.update(1, desc=network_desc) - z._make_request.assert_called_once_with( - "POST", url, callback=None, errback=None, body={"desc": network_desc} - ) - - -@pytest.mark.parametrize("network_id, url", [("1", "ipam/network/1")]) -def test_rest_network_delete(network_config, network_id, url): - z = ns1.rest.ipam.Networks(network_config) - z._make_request = mock.MagicMock() - z.delete(network_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_pools.py b/tests/unit/test_pools.py deleted file mode 100644 index c47f450..0000000 --- a/tests/unit/test_pools.py +++ /dev/null @@ -1,99 +0,0 @@ -import pytest - -import ns1.rest.pools - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def pool_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -@pytest.mark.parametrize("address_id, url", [("1", "ipam/address/1/pool")]) -def test_rest_pool_list(pool_config, address_id, url): - z = ns1.rest.pools.Pools(pool_config) - z._make_request = mock.MagicMock() - z.list(address_id=address_id) - z._make_request.assert_called_once_with( - "GET", - url, - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize( - "address_id, pool_id, url", [("1", "2", "ipam/address/1/pool/2")] -) -def test_rest_pools_retrieve(pool_config, address_id, pool_id, url): - z = ns1.rest.pools.Pools(pool_config) - z._make_request = mock.MagicMock() - z.retrieve(address_id=address_id, pool_id=pool_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize("address_id, url", [("1", "ipam/address/1/pool")]) -def test_rest_pool_create(pool_config, address_id, url): - z = ns1.rest.pools.Pools(pool_config) - z._make_request = mock.MagicMock() - z.create(address_id=address_id) - z._make_request.assert_called_once_with( - "PUT", - url, - body={ - "address": address_id, - "pool": None, - }, - callback=None, - errback=None, - ) - - -@pytest.mark.parametrize( - "address_id, pool_id, url", - [("1", "2", "ipam/address/1/pool/2")], -) -def test_rest_pool_update(pool_config, address_id, pool_id, url): - z = ns1.rest.pools.Pools(pool_config) - z._make_request = mock.MagicMock() - z.update(address_id=address_id, pool_id=pool_id) - z._make_request.assert_called_once_with( - "POST", - url, - callback=None, - errback=None, - body={ - "address": address_id, - "pool": pool_id, - }, - ) - - -@pytest.mark.parametrize( - "address_id, pool_id, url", [("1", "2", "ipam/address/1/pool/2")] -) -def test_rest_pool_delete(pool_config, address_id, pool_id, url): - z = ns1.rest.pools.Pools(pool_config) - z._make_request = mock.MagicMock() - z.delete(address_id=address_id, pool_id=pool_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_reservation.py b/tests/unit/test_reservation.py deleted file mode 100644 index 24e8089..0000000 --- a/tests/unit/test_reservation.py +++ /dev/null @@ -1,95 +0,0 @@ -import pytest - -import ns1.rest.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -def Any(): - class Any: - def __eq__(self, other): - return True - - return Any() - - -@pytest.fixture -def reservation_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_reservation_list(reservation_config): - z = ns1.rest.ipam.Reservations(reservation_config) - z._make_request = mock.MagicMock() - z.list(1) - z._make_request.assert_called_once_with( - "GET", "dhcp/reservation?scopeGroupId=1", callback=None, errback=None - ) - - -@pytest.mark.parametrize("reservation_id, url", [(1, "dhcp/reservation/1")]) -def test_rest_reservation_retrieve(reservation_config, reservation_id, url): - z = ns1.rest.ipam.Reservations(reservation_config) - z._make_request = mock.MagicMock() - z.retrieve(reservation_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "sgroup_id, address_id, mac, options, url", - [ - ( - 1, - 2, - "12:34:56:78:90:ab", - [{"name": "dhcpv4/bootfile-name", "value": "boot.iso"}], - "dhcp/reservation", - ) - ], -) -def test_rest_reservation_create( - reservation_config, sgroup_id, address_id, mac, options, url -): - z = ns1.rest.ipam.Reservations(reservation_config) - z._make_request = mock.MagicMock() - - z.create(sgroup_id, address_id, options=options, mac=mac) - z._make_request.assert_called_once_with( - "PUT", - url, - callback=Any(), - errback=None, - body={ - "address_id": address_id, - "scope_group_id": sgroup_id, - "mac": mac, - "options": options, - }, - ) - - -@pytest.mark.parametrize("reservation_id, url", [(1, "dhcp/reservation/1")]) -def test_rest_reservation_delete(reservation_config, reservation_id, url): - z = ns1.rest.ipam.Reservations(reservation_config) - z._make_request = mock.MagicMock() - z.delete(reservation_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_scope.py b/tests/unit/test_scope.py deleted file mode 100644 index 4eea683..0000000 --- a/tests/unit/test_scope.py +++ /dev/null @@ -1,91 +0,0 @@ -import pytest - -import ns1.rest.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -def Any(): - class Any: - def __eq__(self, other): - return True - - return Any() - - -@pytest.fixture -def scope_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_scope_list(scope_config): - z = ns1.rest.ipam.Scopes(scope_config) - z._make_request = mock.MagicMock() - z.list(1) - z._make_request.assert_called_once_with( - "GET", "dhcp/scope?scopeGroupId=1", callback=None, errback=None - ) - - -@pytest.mark.parametrize("scope_id, url", [(1, "dhcp/scope/1")]) -def test_rest_scope_retrieve(scope_config, scope_id, url): - z = ns1.rest.ipam.Scopes(scope_config) - z._make_request = mock.MagicMock() - z.retrieve(scope_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "sgroup_id, address_id, options, url", - [ - ( - 1, - 2, - [{"name": "dhcpv4/bootfile-name", "value": "boot.iso"}], - "dhcp/scope", - ) - ], -) -def test_rest_scope_create(scope_config, sgroup_id, address_id, options, url): - z = ns1.rest.ipam.Scopes(scope_config) - z._make_request = mock.MagicMock() - - z.create(sgroup_id, address_id, options) - z._make_request.assert_called_once_with( - "PUT", - url, - callback=Any(), - errback=None, - body={ - "address_id": address_id, - "scope_group_id": sgroup_id, - "options": options, - }, - ) - - -@pytest.mark.parametrize("scope_id, url", [(1, "dhcp/scope/1")]) -def test_rest_scope_delete(scope_config, scope_id, url): - z = ns1.rest.ipam.Scopes(scope_config) - z._make_request = mock.MagicMock() - z.delete(scope_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - ) diff --git a/tests/unit/test_scope_group.py b/tests/unit/test_scope_group.py deleted file mode 100644 index fcdd814..0000000 --- a/tests/unit/test_scope_group.py +++ /dev/null @@ -1,92 +0,0 @@ -import pytest - -import ns1.rest.ipam - -try: # Python 3.3 + - import unittest.mock as mock -except ImportError: - import mock - - -@pytest.fixture -def scope_group_config(config): - config.loadFromDict( - { - "endpoint": "api.nsone.net", - "default_key": "test1", - "keys": { - "test1": { - "key": "key-1", - "desc": "test key number 1", - } - }, - } - ) - return config - - -def test_rest_scope_group_list(scope_group_config): - z = ns1.rest.ipam.Scopegroups(scope_group_config) - z._make_request = mock.MagicMock() - z.list() - z._make_request.assert_called_once_with( - "GET", "dhcp/scopegroup", callback=None, errback=None - ) - - -@pytest.mark.parametrize("scope_group_id, url", [("1", "dhcp/scopegroup/1")]) -def test_rest_scope_group_retrieve(scope_group_config, scope_group_id, url): - z = ns1.rest.ipam.Scopegroups(scope_group_config) - z._make_request = mock.MagicMock() - z.retrieve(scope_group_id) - z._make_request.assert_called_once_with( - "GET", url, callback=None, errback=None - ) - - -@pytest.mark.parametrize( - "scope_group_name, url, echo_client_id", - [("test_scope_group", "dhcp/scopegroup", False)], -) -def test_rest_scope_group_create( - scope_group_config, scope_group_name, url, echo_client_id -): - z = ns1.rest.ipam.Scopegroups(scope_group_config) - z._make_request = mock.MagicMock() - z.create(name=scope_group_name, echo_client_id=echo_client_id) - z._make_request.assert_called_once_with( - "PUT", - url, - callback=None, - errback=None, - body={"name": scope_group_name, "dhcpv4": {"echo_client_id": False}}, - ) - - -@pytest.mark.parametrize( - "scope_group_id, scope_group_name, url", - [("1", "awesome scope_group", "dhcp/scopegroup/1")], -) -def test_rest_scope_group_update( - scope_group_config, scope_group_id, scope_group_name, url -): - z = ns1.rest.ipam.Scopegroups(scope_group_config) - z._make_request = mock.MagicMock() - z.update(scope_group_id, name=scope_group_name) - z._make_request.assert_called_once_with( - "POST", - url, - callback=None, - errback=None, - body={"name": scope_group_name}, - ) - - -@pytest.mark.parametrize("scope_group_id, url", [("1", "dhcp/scopegroup/1")]) -def test_rest_scope_group_delete(scope_group_config, scope_group_id, url): - z = ns1.rest.ipam.Scopegroups(scope_group_config) - z._make_request = mock.MagicMock() - z.delete(scope_group_id) - z._make_request.assert_called_once_with( - "DELETE", url, callback=None, errback=None - )