def test_cidr_cast(self):
import ipaddress as ip
cur = self.conn.cursor()
psycopg2.extras.register_ipaddress(cur)
cur.execute("select null::cidr")
self.assert_(cur.fetchone()[0] is None)
cur.execute("select '127.0.0.0/24'::cidr")
obj = cur.fetchone()[0]
self.assert_(isinstance(obj, ip.IPv4Network), repr(obj))
self.assertEquals(obj, ip.ip_network('127.0.0.0/24'))
cur.execute("select '::ffff:102:300/128'::cidr")
obj = cur.fetchone()[0]
self.assert_(isinstance(obj, ip.IPv6Network), repr(obj))
self.assertEquals(obj, ip.ip_network('::ffff:102:300/128'))
python类IPv4Network()的实例源码
def removeSAPNAT(self, SAPSwitch):
SAPip = SAPSwitch.ip
SAPNet = str(ipaddress.IPv4Network(unicode(SAPip), strict=False))
# due to a bug with python-iptables, removing and finding rules does not succeed when the mininet CLI is running
# so we use the iptables tool
rule0_ = "iptables -t nat -D POSTROUTING ! -o {0} -s {1} -j MASQUERADE".format(SAPSwitch.deployed_name, SAPNet)
p = Popen(shlex.split(rule0_))
p.communicate()
rule1_ = "iptables -D FORWARD -o {0} -j ACCEPT".format(SAPSwitch.deployed_name)
p = Popen(shlex.split(rule1_))
p.communicate()
rule2_ = "iptables -D FORWARD -i {0} -j ACCEPT".format(SAPSwitch.deployed_name)
p = Popen(shlex.split(rule2_))
p.communicate()
info("remove SAP NAT rules for: {0} - {1}\n".format(SAPSwitch.name, SAPNet))
def __init__(self, address: str, mask: int=32):
"""
:param address: The IP address
:param mask: this is the mask it should take for the number of hosts that you are planning to access
the mask can be a valid host mask or a valid net_mask or a num_digit slash.
host mask starts with 0 and contain either 0's and 255's
net masks start with 255 and contain either 0's and 255's
"""
try:
self.ip_address = Resolve(address)
self.__mask = mask
addr = self.ip_address + str(mask)
self.mask_addr = ipaddress.IPv4Network(addr, strict=False) if self.ip_address.version == "IPv4" else \
ipaddress.IPv6Network(addr, strict=False)
except UnResolvedException as unRes:
print(unRes.args[1])
sys.exit()
def _validate_ip_name(self, tree):
if any(isinstance(name, IPAddress) and not isinstance(
name.value, (ipaddress.IPv4Network, ipaddress.IPv6Network)
) for name in tree):
raise TypeError(
"IPAddress name constraints must be an IPv4Network or"
" IPv6Network object"
)
def _validate_ip_name(self, tree):
if any(isinstance(name, IPAddress) and not isinstance(
name.value, (ipaddress.IPv4Network, ipaddress.IPv6Network)
) for name in tree):
raise TypeError(
"IPAddress name constraints must be an IPv4Network or"
" IPv6Network object"
)
def register_ipaddress(conn_or_curs=None):
"""
Register conversion support between `ipaddress` objects and `network types`__.
:param conn_or_curs: the scope where to register the type casters.
If `!None` register them globally.
After the function is called, PostgreSQL :sql:`inet` values will be
converted into `~ipaddress.IPv4Interface` or `~ipaddress.IPv6Interface`
objects, :sql:`cidr` values into into `~ipaddress.IPv4Network` or
`~ipaddress.IPv6Network`.
.. __: https://www.postgresql.org/docs/current/static/datatype-net-types.html
"""
global ipaddress
import ipaddress
global _casters
if _casters is None:
_casters = _make_casters()
for c in _casters:
register_type(c, conn_or_curs)
for t in [ipaddress.IPv4Interface, ipaddress.IPv6Interface,
ipaddress.IPv4Network, ipaddress.IPv6Network]:
register_adapter(t, adapt_ipaddress)
def test_cidr_array_cast(self):
import ipaddress as ip
cur = self.conn.cursor()
psycopg2.extras.register_ipaddress(cur)
cur.execute("select '{NULL,127.0.0.1,::ffff:102:300/128}'::cidr[]")
l = cur.fetchone()[0]
self.assertTrue(l[0] is None)
self.assertEqual(l[1], ip.ip_network('127.0.0.1'))
self.assertEqual(l[2], ip.ip_network('::ffff:102:300/128'))
self.assertTrue(isinstance(l[1], ip.IPv4Network), l)
self.assertTrue(isinstance(l[2], ip.IPv6Network), l)
def register_ipaddress(conn_or_curs=None):
"""
Register conversion support between `ipaddress` objects and `network types`__.
:param conn_or_curs: the scope where to register the type casters.
If `!None` register them globally.
After the function is called, PostgreSQL :sql:`inet` values will be
converted into `~ipaddress.IPv4Interface` or `~ipaddress.IPv6Interface`
objects, :sql:`cidr` values into into `~ipaddress.IPv4Network` or
`~ipaddress.IPv6Network`.
.. __: https://www.postgresql.org/docs/current/static/datatype-net-types.html
"""
global ipaddress
import ipaddress
global _casters
if _casters is None:
_casters = _make_casters()
for c in _casters:
register_type(c, conn_or_curs)
for t in [ipaddress.IPv4Interface, ipaddress.IPv6Interface,
ipaddress.IPv4Network, ipaddress.IPv6Network]:
register_adapter(t, adapt_ipaddress)
def setup(self):
self.invalid_types = (True, False, 1)
self.address01 = IPv4Network(u'10.0.0.0/8')
self.address02 = IPv4Network(u'192.168.0.0/16')
self.address03 = IPv4Network(u'172.16.0.0/12')
def test_addresses_getitem(self):
group = NetworkGroup()
assert_equals(group.addresses[0], IPv4Network(u'0.0.0.0/0'))
def test_add(self):
group = NetworkGroup()
address04 = IPv4Network(u'10.0.0.0/8')
assert_true(group.add(self.address01))
assert_false(group.add(self.address01))
assert_true(group.add(self.address02))
assert_false(group.add(self.address02))
assert_false(group.add(self.address01))
assert_false(group.add(address04))
def network_address(value):
""" Convert a string to a network IP """
value = NetTest.convert.string.cidr(value)
return ipaddress.IPv4Network(value).network_address.exploded
def broadcast_address(value):
""" Convert a string to a broadcast IP """
value = NetTest.convert.string.cidr(value)
return ipaddress.IPv4Network(value).broadcast_address.exploded
def address_builder(value):
if isinstance(value, (basestring)):
try:
value = IPv4Interface(unicode(value)).network
except AddressValueError:
message = 'Unsupported string initialization format \'{}\''
message = message.format(value)
raise ValueError(message)
elif not isinstance(value, IPv4Network):
raise_type_exception(value, (IPv4Network, ), 'build with')
return value
def _validate_ip_name(self, tree):
if any(isinstance(name, IPAddress) and not isinstance(
name.value, (ipaddress.IPv4Network, ipaddress.IPv6Network)
) for name in tree):
raise TypeError(
"IPAddress name constraints must be an IPv4Network or"
" IPv6Network object"
)
def __init__(self, prefix=None, strict=False):
"""Init new Prefix instance."""
if isinstance(prefix, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
self._prefix = prefix
else:
self._prefix = ipaddress.ip_network(prefix, strict=strict)
def create_ips_for_subnet(sender, instance, created, **kwargs):
if not created:
return
subnet = ipaddress.IPv4Network(instance.network + "/" + instance.netmask)
for addr in subnet.hosts():
address = Ip4Address(address=str(addr), subnet=instance)
address.save()
def num_addresses(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return subnet.num_addresses
def broadcast_address(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return subnet.broadcast_address
def first_address(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return next(subnet.hosts())
