1
0
mirror of https://github.com/samba-team/samba.git synced 2025-01-22 22:04:08 +03:00

python/samba/tests: PY3 port samba.tests.dns

Misc hanges needed to get make test TEST=samba.tests.dns &
samb.tests.dns_fowarder to run and pass under PY3

* socket.send needs bytes not string
* rec.dwTimeStamp expects int not float (in PY3 / operator
  will give float results, for int use '//' instead)
* re.match using bytes needs a bytes search term

Signed-off-by: Noel Power <noel.power@suse.com>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
Noel Power 2018-11-04 17:32:19 +00:00 committed by Noel Power
parent 0d73a2b0a2
commit a25bd836ee
4 changed files with 13 additions and 12 deletions

View File

@ -988,7 +988,7 @@ class TestInvalidQueries(DNSTest):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
s.connect((self.server_ip, 53))
s.send("", 0)
s.send(b"", 0)
finally:
if s is not None:
s.close()
@ -1253,13 +1253,13 @@ class TestZones(DNSTest):
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
rec.dwTimeStamp -= interval / 2
rec.dwTimeStamp -= interval // 2
self.ldap_modify_dnsrecs(name, mod_ts)
update_during_norefresh = self.dns_update_record(name, txt)
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
rec.dwTimeStamp -= interval + interval / 2
rec.dwTimeStamp -= interval + interval // 2
self.ldap_modify_dnsrecs(name, mod_ts)
update_during_refresh = self.dns_update_record(name, txt)
self.assertEqual(update_during_norefresh.dwTimeStamp,

View File

@ -189,7 +189,7 @@ class TestDnsForwarding(DNSTest):
time.sleep(0.05)
s.connect((host, port))
try:
s.send('timeout 0', 0)
s.send(b'timeout 0', 0)
except socket.error as e:
if e.errno in (errno.ECONNREFUSED, errno.EHOSTUNREACH):
continue
@ -206,7 +206,7 @@ class TestDnsForwarding(DNSTest):
def test_comatose_forwarder(self):
s = self.start_toy_server(dns_servers[0], 53, 'forwarder1')
s.send("timeout 1000000", 0)
s.send(b"timeout 1000000", 0)
# make DNS query
name = "an-address-that-will-not-resolve"
@ -332,7 +332,7 @@ class TestDnsForwarding(DNSTest):
def test_single_forwarder_waiting_forever(self):
s = self.start_toy_server(dns_servers[0], 53, 'forwarder1')
s.send('timeout 10000', 0)
s.send(b'timeout 10000', 0)
ad = contact_real_server(server_ip, 53)
name = "dsfsfds.dsfsdfs"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
@ -361,7 +361,7 @@ class TestDnsForwarding(DNSTest):
return
s1 = self.start_toy_server(dns_servers[0], 53, 'forwarder1')
s2 = self.start_toy_server(dns_servers[1], 53, 'forwarder2')
s1.send('timeout 1000', 0)
s1.send(b'timeout 1000', 0)
ad = contact_real_server(server_ip, 53)
name = "dsfsfds.dsfsdfs"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
@ -419,8 +419,8 @@ class TestDnsForwarding(DNSTest):
return
s1 = self.start_toy_server(dns_servers[0], 53, 'forwarder1')
s2 = self.start_toy_server(dns_servers[1], 53, 'forwarder2')
s1.send('timeout 1.5', 0)
s2.send('timeout 1.5', 0)
s1.send(b'timeout 1.5', 0)
s2.send(b'timeout 1.5', 0)
ad = contact_real_server(server_ip, 53)
name = "dsfsfds.dsfsdfs"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
@ -505,7 +505,7 @@ class TestDnsForwarding(DNSTest):
return
s1 = self.start_toy_server(dns_servers[0], 53, 'forwarder1')
s2 = self.start_toy_server(dns_servers[1], 53, 'forwarder2')
s1.send('timeout 10000', 0)
s1.send(b'timeout 10000', 0)
name = 'resolve.cname.%s' % self.get_dns_domain()
self.make_cname_update(name, "dsfsfds.dsfsdfs")

View File

@ -76,7 +76,7 @@ class DnsHandler(SocketServer.BaseRequestHandler):
debug("%s: %s wrote:" % (SERVER_ID, self.client_address[0]))
global timeout
m = re.match('^timeout\s+([\d.]+)$', data.strip())
m = re.match(b'^timeout\s+([\d.]+)$', data.strip())
if m:
timeout = float(m.group(1))
debug("timing out at %s" % timeout)

View File

@ -50,6 +50,7 @@ from samba.netcmd.dns import cmd_dns
from samba import gensec
from samba.kcc import kcc_utils
from samba.compat import get_string
from samba.compat import text_type
import ldb
samba.ensure_third_party_module("dns", "dnspython")
@ -677,7 +678,7 @@ def rodc_dns_update(d, t, op):
else:
name.dns_register = False
dns_names.names = [ name ]
site_name = sub_vars['SITE'].decode('utf-8')
site_name = text_type(sub_vars['SITE'])
global error_count