+
+
+ def test_join_records_can_update(self):
+ dc_creds = Credentials()
+ dc_creds.guess(self.join_ctx.lp)
+ dc_creds.set_machine_account(self.join_ctx.lp)
+
+ self.tkey_trans(creds=dc_creds)
+
+ p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
+ q = self.make_name_question(self.join_ctx.dnsdomain,
+ dns.DNS_QTYPE_SOA,
+ dns.DNS_QCLASS_IN)
+ questions = []
+ questions.append(q)
+ self.finish_name_packet(p, questions)
+
+ updates = []
+ # Delete the old expected IPs
+ IPs = samba.interface_ips(self.lp)
+ for IP in IPs[1:]:
+ if ":" in IP:
+ r = dns.res_rec()
+ r.name = self.join_ctx.dnshostname
+ r.rr_type = dns.DNS_QTYPE_AAAA
+ r.rr_class = dns.DNS_QCLASS_NONE
+ r.ttl = 0
+ r.length = 0xffff
+ rdata = IP
+ else:
+ r = dns.res_rec()
+ r.name = self.join_ctx.dnshostname
+ r.rr_type = dns.DNS_QTYPE_A
+ r.rr_class = dns.DNS_QCLASS_NONE
+ r.ttl = 0
+ r.length = 0xffff
+ rdata = IP
+
+ r.rdata = rdata
+ updates.append(r)
+
+ p.nscount = len(updates)
+ p.nsrecs = updates
+
+ mac = self.sign_packet(p, self.key_name)
+ (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.verify_packet(response, response_p, mac)
+
+ p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
+ questions = []
+
+ name = self.join_ctx.dnshostname
+ q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
+ questions.append(q)
+
+ self.finish_name_packet(p, questions)
+ (response, response_packet) = self.dns_transaction_tcp(p, host=self.server_ip)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
+ self.assertEquals(response.ancount, 1)