diff --git a/ipam/client/abstractipam.py b/ipam/client/abstractipam.py index 13251e7..44f03e3 100644 --- a/ipam/client/abstractipam.py +++ b/ipam/client/abstractipam.py @@ -70,6 +70,10 @@ def get_ip_list_by_desc(self, description): def get_ip_by_desc(self, description): raise NotImplementedError() + @abstractmethod + def get_ip_by_desc_and_subnet(self, description, subnet_description): + raise NotImplementedError() + @abstractmethod def get_ip_list_by_mac(self, mac): raise NotImplementedError() diff --git a/ipam/client/backends/phpipam.py b/ipam/client/backends/phpipam.py index c77a148..c4d667e 100644 --- a/ipam/client/backends/phpipam.py +++ b/ipam/client/backends/phpipam.py @@ -680,6 +680,33 @@ def get_ip_by_desc(self, description): else: return iplist[0] + def get_ip_by_desc_and_subnet(self, description, subnet): + self.cur.execute("SELECT id \ + FROM subnets \ + WHERE subnet = '%i'" % int(ip_address((subnet)))) + row = self.cur.fetchone() + if row is not None: + subnet_id = int(row[0]) + else: + raise ValueError( + "Unable to get subnet id from database " + "for this subnet: {}".format(subnet) + ) + + self.cur.execute("SELECT ip_addr \ + FROM ipaddresses \ + WHERE description LIKE '%s' AND subnetId = '%i'" + % (description, subnet_id)) + + row = self.cur.fetchone() + if row is not None: + return ip_address(int(row[0])) + + raise ValueError( + "Unable to get {} in the subnet {} from database".format( + description, subnet) + ) + def get_ip_list_by_mac(self, mac): self.cur.execute("SELECT ip_addr,description,%s,state,mac \ FROM ipaddresses \ diff --git a/ipam/client/tests/test_phpipam.py b/ipam/client/tests/test_phpipam.py index f4f319f..a74be94 100644 --- a/ipam/client/tests/test_phpipam.py +++ b/ipam/client/tests/test_phpipam.py @@ -587,6 +587,26 @@ def test_get_ip_by_desc(testphpipam): assert testip['state'] == testphpipam.used_ip_state +def test_get_ip_by_desc_and_subnet(testphpipam): + with pytest.raises( + ValueError, + match='Unable to get subnet id from database for this subnet: ' + '10.0.0.0' + ): + testphpipam.get_ip_by_desc_and_subnet('unknown ip', '10.0.0.0') + + testip = testphpipam.get_ip_by_desc_and_subnet( + 'test ip #2', '10.1.0.0' + ) + assert testip == ip_address('10.1.0.2') + + with pytest.raises( + ValueError, + match='Unable to get unknown ip in the subnet 10.1.0.0' + ): + testphpipam.get_ip_by_desc_and_subnet('unknown ip', '10.1.0.0') + + def test_get_ip_interface_by_desc(testphpipam): assert testphpipam.get_ip_by_desc('unknown ip') is None