JezK

Name: test
Start: 2022-11-25
End: 2022-11-25

Edit File: dns_aging.py

# Unix SMB/CIFS implementation.
# Copyright (C) Kai Blin  <kai@samba.org> 2011
# Copyright (C) Catalyst.NET 2021
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import sys
from samba import dsdb
from samba import dsdb_dns
from samba.ndr import ndr_unpack, ndr_pack
from samba.samdb import SamDB
from samba.auth import system_session
import ldb
from samba import credentials
from samba.dcerpc import dns, dnsp, dnsserver
from samba.dnsserver import TXTRecord, ARecord
from samba.dnsserver import recbuf_from_string, ipv6_normalise
from samba.tests.subunitrun import SubunitOptions, TestProgram
from samba import werror, WERRORError
from samba.tests.dns_base import DNSTest
import samba.getopt as options
import optparse
import time
from samba.colour import c_RED, c_GREEN, c_DARK_YELLOW

parser = optparse.OptionParser(
    "dns_aging.py <server name> <server ip> [options]")
sambaopts = options.SambaOptions(parser)
parser.add_option_group(sambaopts)

# use command line creds if available
credopts = options.CredentialsOptions(parser)
parser.add_option_group(credopts)
subunitopts = SubunitOptions(parser)
parser.add_option_group(subunitopts)

opts, args = parser.parse_args()
if len(args) < 2:
    parser.print_usage()
    sys.exit(1)

LP = sambaopts.get_loadparm()
CREDS = credopts.get_credentials(LP)
SERVER_NAME = args[0]
SERVER_IP = args[1]
CREDS.set_krb_forwardable(credentials.NO_KRB_FORWARDABLE)

DOMAIN = CREDS.get_realm().lower()

# Unix time start, in DNS timestamp (24 * 365.25 * 369)
# These are ballpark extremes for the timestamp.
DNS_TIMESTAMP_1970 = 3234654
DNS_TIMESTAMP_2101 = 4383000
DNS_TIMESTAMP_1981 = 3333333  # a middling timestamp

IPv4_ADDR = "127.0.0.33"
IPv6_ADDR = "::1"
IPv4_ADDR_2 = "127.0.0.66"
IPv6_ADDR_2 = "1::1"

def get_samdb():
    return SamDB(url=f"ldap://{SERVER_IP}",
                 lp=LP,
                 session_info=system_session(),
                 credentials=CREDS)

def get_file_samdb():
    # For Samba only direct file access, needed for the tombstoning functions.
    # (For Windows, we instruct it to tombstone over RPC).
    return SamDB(url=LP.samdb_url(),
                 lp=LP,
                 session_info=system_session(),
                 credentials=CREDS)

def get_rpc():
    return dnsserver.dnsserver(f"ncacn_ip_tcp:{SERVER_IP}[sign]", LP, CREDS)

def create_zone(name, rpc=None, aging=True):
    if rpc is None:
        rpc = get_rpc()
    z = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
    z.pszZoneName = name
    z.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
    z.fAging = int(bool(aging))
    z.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT
    z.fDsIntegrated = 1
    z.fLoadExisting = 1
    z.fAllowUpdate = dnsp.DNS_ZONE_UPDATE_UNSECURE
    rpc.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
                         0,
                         SERVER_IP,
                         None,
                         0,
                         'ZoneCreate',
                         dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
                         z)

def delete_zone(name, rpc=None):
    if rpc is None:
        rpc = get_rpc()
    rpc.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
                         0,
                         SERVER_IP,
                         name,
                         0,
                         'DeleteZoneFromDs',
                         dnsserver.DNSSRV_TYPEID_NULL,
                         None)

def txt_s_list(txt):
    """Construct a txt record string list, which is a fiddly matter."""
    if isinstance(txt, str):
        txt = [txt]
    s_list = dnsp.string_list()
    s_list.count = len(txt)
    s_list.str = txt
    return s_list

def make_txt_record(txt):
    r = dns.txt_record()
    r.txt = txt_s_list(txt)
    return r

def copy_rec(rec):
    copy = dnsserver.DNS_RPC_RECORD()
    copy.wType = rec.wType
    copy.dwFlags = rec.dwFlags
    copy.dwSerial = rec.dwSerial
    copy.dwTtlSeconds = rec.dwTtlSeconds
    copy.data = rec.data
    copy.dwTimeStamp = rec.dwTimeStamp
    return copy

def guess_wtype(data):
    if isinstance(data, list):
        data = make_txt_record(data)
        return (data, dnsp.DNS_TYPE_TXT)
    if ":" in data:
        return (data, dnsp.DNS_TYPE_AAAA)
    return (data, dnsp.DNS_TYPE_A)

class TestDNSAging(DNSTest):
    """Probe DNS aging and scavenging, using LDAP and RPC to set and test
    the timestamps behind DNS's back."""
    server = SERVER_NAME
    server_ip = SERVER_IP
    creds = CREDS

def setUp(self):
        super().setUp()
        self.rpc_conn = get_rpc()
        self.samdb = get_samdb()

# We always have a zone of our own named after the test function.
        self.zone = self.id().rsplit('.', 1)[1]
        self.addCleanup(delete_zone, self.zone, self.rpc_conn)
        try:
            create_zone(self.zone, self.rpc_conn)
        except WERRORError as e:
            if e.args[0] != werror.WERR_DNS_ERROR_ZONE_ALREADY_EXISTS:
                raise
            print(f"zone {self.zone} already exists")

# Though we set this in create_zone(), that doesn't work on
        # Windows, so we repeat again here.
        self.set_zone_int_params(AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)

self.zone_dn = (f"DC={self.zone},CN=MicrosoftDNS,DC=DomainDNSZones,"
                        f"{self.samdb.get_default_basedn()}")

def set_zone_int_params(self, zone=None, **kwargs):
        """Keyword arguments set parameters on the zone. e.g.:

self.set_zone_int_params(Aging=1,
                                     RefreshInterval=222)

See [MS-DNSP] 3.1.1.2.1 "DNS Zone Integer Properties" for names.
        """
        if zone is None:
            zone = self.zone
        for key, val in kwargs.items():
            name_param = dnsserver.DNS_RPC_NAME_AND_PARAM()
            name_param.dwParam = val
            name_param.pszNodeName = key
            try:
                self.rpc_conn.DnssrvOperation2(
                    dnsserver.DNS_CLIENT_VERSION_LONGHORN,
                    0,
                    SERVER_IP,
                    zone,
                    0,
                    'ResetDwordProperty',
                    dnsserver.DNSSRV_TYPEID_NAME_AND_PARAM,
                    name_param)
            except WERRORError as e:
                self.fail(str(e))

def rpc_replace(self, name, old=None, new=None):
        """Replace a DNS_RPC_RECORD or DNS_RPC_RECORD_BUF"""
        # wrap our recs, if necessary
        if isinstance(new, dnsserver.DNS_RPC_RECORD):
            rec = new
            new = dnsserver.DNS_RPC_RECORD_BUF()
            new.rec = rec

if isinstance(old, dnsserver.DNS_RPC_RECORD):
            rec = old
            old = dnsserver.DNS_RPC_RECORD_BUF()
            old.rec = rec

try:
            self.rpc_conn.DnssrvUpdateRecord2(
                dnsserver.DNS_CLIENT_VERSION_LONGHORN,
                0,
                SERVER_IP,
                self.zone,
                name,
                new,
                old)
        except WERRORError as e:
            self.fail(f"could not replace record ({e})")

def get_unique_txt_record(self, name, txt):
        """Get the TXT record on Name with value txt, asserting that there is
        only one."""
        if isinstance(txt, str):
            txt = [txt]
        recs = self.ldap_get_records(name)

match = None
        for r in recs:
            if r.wType != dnsp.DNS_TYPE_TXT:
                continue
            txt2 = [x for x in r.data.str]
            if txt2 == txt:
                self.assertIsNone(match)
                match = r
        return match

def get_unique_ip_record(self, name, addr, wtype=None):
        """Get an A or AAAA record on name with the matching data."""
        if wtype is None:
            addr, wtype = guess_wtype(addr)

recs = self.ldap_get_records(name)

# We need to use the internal dns_record_match because not all
        # forms always match on strings (e.g. IPv6)
        rec = dnsp.DnssrvRpcRecord()
        rec.wType = wtype
        rec.data = addr

match = None
        for r in recs:
            if dsdb_dns.records_match(r, rec):
                self.assertIsNone(match)
                match = r
        return match

def dns_query(self, name, qtype=dns.DNS_QTYPE_ALL):
        """make a query, which might help Windows notice LDAP changes"""
        p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
        fullname = "%s.%s" % (name, self.zone)
        q = self.make_name_question(fullname, qtype, dns.DNS_QCLASS_IN)
        self.finish_name_packet(p, [q])
        r, rp = self.dns_transaction_udp(p, host=SERVER_IP)

return r

def dns_update_non_text(self, name,
                            data,
                            wtype=None,
                            qclass=dns.DNS_QCLASS_IN):
        if wtype is None:
            data, wtype = guess_wtype(data)

if qclass == dns.DNS_QCLASS_IN:
            ttl = 123
        else:
            ttl = 0

fullname = "%s.%s" % (name, self.zone)
        p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
        u = self.make_name_question(self.zone,
                                    dns.DNS_QTYPE_SOA,
                                    dns.DNS_QCLASS_IN)
        self.finish_name_packet(p, [u])

r = dns.res_rec()
        r.name = fullname
        r.rr_type = wtype
        r.rr_class = qclass
        r.ttl = ttl
        if data is not None:
            r.length = 0xffff
            r.rdata = data
        else:
            r.length = 0

p.nscount = 1
        p.nsrecs = [r]

(code, response) = self.dns_transaction_udp(p, host=SERVER_IP)
        self.assert_dns_rcode_equals(code, dns.DNS_RCODE_OK)
        return response

def dns_delete(self, name, data, wtype=None):
        return self.dns_update_non_text(name,
                                        data,
                                        wtype,
                                        qclass=dns.DNS_QCLASS_NONE)

def dns_delete_type(self, name, wtype):
        return self.dns_update_non_text(name,
                                        None,
                                        wtype,
                                        qclass=dns.DNS_QCLASS_ANY)

def dns_update_record(self, name, txt, ttl=900):
        if isinstance(txt, str):
            txt = [txt]
        p = self.make_txt_update(name, txt, self.zone, ttl=ttl)
        (code, response) = self.dns_transaction_udp(p, host=SERVER_IP)
        if code.operation & dns.DNS_RCODE == dns.DNS_RCODE_REFUSED:
            # sometimes you might forget this
            print("\n\ngot DNS_RCODE_REFUSED\n")
            print("Are you running this in the fl2003 environment?\n")
            print("try `SELFTEST_TESTENV='fl2003dc:local' make testenv`\n\n")

self.assert_dns_rcode_equals(code, dns.DNS_RCODE_OK)
        return self.get_unique_txt_record(name, txt)

def rpc_update_record(self, name, txt, **kwargs):
        """Add the record that self.dns_update_record() would add, via the
        dnsserver RPC pipe.

As with DNS update, if the record already exists, we replace it.
        """
        if isinstance(txt, str):
            txt = [txt]

old = TXTRecord(txt)
        rec = TXTRecord(txt)
        for k, v in kwargs.items():
            setattr(rec, k, v)

try:
            self.rpc_replace(name, old, rec)
        except AssertionError as e:
            # we have caught and wrapped the WERRor inside
            if 'WERR_DNS_ERROR_RECORD_DOES_NOT_EXIST' not in str(e):
                raise
            self.rpc_replace(name, None, rec)

return self.get_unique_txt_record(name, txt)

def rpc_delete_txt(self, name, txt):
        if isinstance(txt, str):
            txt = [txt]
        old = TXTRecord(txt)
        self.rpc_replace(name, old, None)

def get_one_node(self, name):
        expr = f"(&(objectClass=dnsNode)(name={name}))"
        nodes = self.samdb.search(base=self.zone_dn,
                                  scope=ldb.SCOPE_SUBTREE,
                                  expression=expr,
                                  attrs=["dnsRecord", "dNSTombstoned", "name"])

if len(nodes) > 1:
            self.fail(
                f"expected 0 or 1 dnsNodes for {name}, found {len(nodes)}")

if len(nodes) == 0:
            return None
        return nodes[0]

def ldap_get_records(self, name):
        node = self.get_one_node(name)
        if node is None:
            return []

records = node.get('dnsRecord')
        return [ndr_unpack(dnsp.DnssrvRpcRecord, r) for r in records]

def ldap_get_non_tombstoned_records(self, name):
        all_records = self.ldap_get_records(name)
        records = []
        for r in all_records:
            if r.wType != dnsp.DNS_TYPE_TOMBSTONE:
                records.append(r)
        return records

def assert_tombstoned(self, name, tombstoned=True, timestamp=None):
        # If run with tombstoned=False, assert it isn't tombstoned
        # (and has no traces of tombstone). Otherwise assert it has
        # all the necessary bits.
        #
        # with timestamp=<non-zero number of hours>, we assert that
        # the nttime timestamp is about that time.
        #
        # with timestamp=None, we assert it is within a century or so.
        #
        # with timestamp=False (or 0), we don't assert on it.

node = self.get_one_node(name)
        if node is None:
            self.fail(f"no node named {name}")

dnsts = node.get("dNSTombstoned")
        if dnsts is None:
            is_tombstoned = False
        else:
            self.assertEqual(len(dnsts), 1)
            if dnsts[0] == b'TRUE':
                is_tombstoned = True
            else:
                is_tombstoned = False

if tombstoned != is_tombstoned:
            if is_tombstoned:
                self.fail(f"{name} is tombstoned")
            else:
                self.fail(f"{name} is not tombstoned")

recs = self.ldap_get_records(name)
        if is_tombstoned:
            self.assertEqual(len(recs), 1)
            self.assertEqual(recs[0].wType, dnsp.DNS_TYPE_TOMBSTONE)
            if timestamp is None:
                self.assert_nttime_in_hour_range(recs[0].data)
            elif timestamp:
                self.assert_nttime_in_hour_range(recs[0].data,
                                                 timestamp - 3,
                                                 timestamp + 3)

else:
            for r in recs:
                self.assertNotEqual(recs[0].wType, dnsp.DNS_TYPE_TOMBSTONE)

def ldap_replace_records(self, name, records):
        # We use raw ldap to avoid the "helpfulness" of dsdb_dns.replace()

dn = f'DC={name},{self.zone_dn}'

msg = ldb.Message.from_dict(self.samdb,
                                    {'dn': dn,
                                     'dnsRecord': [ndr_pack(r) for r in records]
                                    },
                                    ldb.FLAG_MOD_REPLACE)

try:
            self.samdb.modify(msg)
        except ldb.LdbError as e:
            if 'LDAP_NO_SUCH_OBJECT' not in e.args[1]:
                raise
            # We need to do an add
            msg["objectClass"] = ["top", "dnsNode"]
            msg["dnsRecord"].set_flags(ldb.FLAG_MOD_ADD)
            self.samdb.add(msg)

def ldap_update_core(self, name, wtype, data, **kwargs):
        """This one is not TXT specific."""
        records = self.ldap_get_records(name)

# default values
        rec = dnsp.DnssrvRpcRecord()
        rec.wType = wtype
        rec.rank = dnsp.DNS_RANK_ZONE
        rec.dwTtlSeconds = 900
        rec.dwSerial = 110
        rec.dwTimeStamp = 0
        rec.data = data

# override defaults, as required
        for k, v in kwargs.items():
            setattr(rec, k, v)

for i, r in enumerate(records[:]):
            if dsdb_dns.records_match(r, rec):
                records[i] = rec
                break
        else:  # record not found
            records.append(rec)

self.ldap_replace_records(name, records)
        return rec

def ldap_update_record(self, name, txt, **kwargs):
        """Add the record that self.dns_update_record() would add, via ldap,
        thus allowing us to set additional dnsRecord features like
        dwTimestamp.
        """
        rec = self.ldap_update_core(name,
                                    dnsp.DNS_TYPE_TXT,
                                    txt_s_list(txt),
                                    **kwargs)

recs = self.ldap_get_records(name)
        match = None
        for r in recs:
            if r.wType != rec.wType:
                continue
            if r.data.str == rec.data.str:
                self.assertIsNone(match, f"duplicate records for {name}")
                match = r
        self.assertEqual(match.rank, rec.rank & 255)
        self.assertEqual(match.dwTtlSeconds, rec.dwTtlSeconds)
        self.assert_timestamps_equal(match.dwTimeStamp, rec.dwTimeStamp)
        return match

def ldap_delete_record(self, name, data, wtype=dnsp.DNS_TYPE_TXT):
        rec = dnsp.DnssrvRpcRecord()
        if wtype == dnsp.DNS_TYPE_TXT:
            data = txt_s_list(data)

rec.wType = wtype
        rec.data = data
        records = self.ldap_get_records(name)
        for i, r in enumerate(records[:]):
            if dsdb_dns.records_match(r, rec):
                del records[i]
                break
        else:
            self.fail(f"record {data} not found")

self.ldap_replace_records(name, records)

def add_ip_record(self, name, addr, wtype=None, **kwargs):
        if wtype is None:
            addr, wtype = guess_wtype(addr)
        rec = self.ldap_update_core(name,
                                    wtype,
                                    addr,
                                    **kwargs)

recs = self.ldap_get_records(name)
        match = None
        for r in recs:
            if dsdb_dns.records_match(r, rec):
                self.assertIsNone(match, f"duplicate records for {name}")
                match = r
        self.assertEqual(match.rank, rec.rank & 255)
        self.assertEqual(match.dwTtlSeconds, rec.dwTtlSeconds)
        self.assert_timestamps_equal(match.dwTimeStamp, rec.dwTimeStamp)
        return match

def ldap_modify_timestamps(self, name, delta):
        records = self.ldap_get_records(name)
        for rec in records:
            rec.dwTimeStamp += delta
        self.ldap_replace_records(name, records)

def get_rpc_records(self, name, dns_type=None):
        if dns_type is None:
            dns_type = dnsp.DNS_TYPE_ALL
        select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA
        buflen, res = self.rpc_conn.DnssrvEnumRecords2(
            dnsserver.DNS_CLIENT_VERSION_LONGHORN,
            0,
            SERVER_IP,
            self.zone,
            name,
            None,
            dns_type,
            select_flags,
            None,
            None)
        recs = []
        if not res or res.count == 0:
            return []
        for rec in res.rec:
            recs.extend(rec.records)
        return recs

def dns_tombstone(self, name,
                      epoch_hours=DNS_TIMESTAMP_1981,
                      epoch_nttime=None):
        dn = f'DC={name},{self.zone_dn}'
        r = dnsp.DnssrvRpcRecord()
        r.wType = dnsp.DNS_TYPE_TOMBSTONE
        # r.dwTimeStamp is a 32 bit value in hours, and r.data is an
        # NTTIME (100 nanosecond intervals), both in the 1601 epoch. A
        # tombstome will have both, but expiration calculations use
        # the r.data NTTIME EntombedTime timestamp (see [MS-DNSP]).
        r.dwTimeStamp = epoch_hours
        if epoch_nttime is None:
            r.data = epoch_hours * 3600 * 10 * 1000 * 1000
        else:
            r.data = epoch_nttime

msg = ldb.Message.from_dict(self.samdb,
                                    {'dn': dn,
                                     'dnsRecord': [ndr_pack(r)],
                                     'dnsTombstoned': 'TRUE'
                                    },
                                    ldb.FLAG_MOD_REPLACE)
        try:
            self.samdb.modify(msg)
        except ldb.LdbError as e:
            if 'LDAP_NO_SUCH_OBJECT' not in e.args[1]:
                raise
            # We need to do an add
            msg["objectClass"] = ["top", "dnsNode"]
            self.samdb.add(msg)

def set_aging(self, enable=False):
        self.set_zone_int_params(Aging=int(bool(enable)))

def assert_timestamp_in_ballpark(self, rec):
        self.assertGreater(rec.dwTimeStamp, DNS_TIMESTAMP_1970)
        self.assertLess(rec.dwTimeStamp, DNS_TIMESTAMP_2101)

def assert_nttime_in_hour_range(self, t,
                                    hour_min=DNS_TIMESTAMP_1970,
                                    hour_max=DNS_TIMESTAMP_2101):
        t //= int(3600 * 1e7)
        self.assertGreater(t, hour_min)
        self.assertLess(t, hour_max)

def assert_soon_after(self, timestamp, reference):
        """Assert that a timestamp is the same or very slightly higher than a
        reference timestamp.

Typically we expect the timestamps to be identical, unless an
        hour has clicked over since the reference was taken. However
        we allow one more hour in case it happens during a daylight
        savings transition or something.
        """
        if hasattr(timestamp, 'dwTimeStamp'):
            timestamp = timestamp.dwTimeStamp
        if hasattr(reference, 'dwTimeStamp'):
            reference = reference.dwTimeStamp

diff = timestamp - reference
        days = abs(diff / 24.0)

if diff < 0:
            msg = f"timestamp is {days} days ({abs(diff)} hours) before reference"
        elif diff > 2:
            msg = f"timestamp is {days} days ({diff} hours) after reference"
        else:
            return
        raise AssertionError(msg)

def assert_timestamps_equal(self, ts1, ts2):
        """Just like assertEqual(), but tells us the difference, not the
        absolute values. e.g:

self.assertEqual(a, b)
        AssertionError: 3685491 != 3685371

self.assert_timestamps_equal(a, b)
        AssertionError: -120 (first is 5.0 days earlier than second)

Also, we turn a record into a timestamp if we need
        """
        if hasattr(ts1, 'dwTimeStamp'):
            ts1 = ts1.dwTimeStamp
        if hasattr(ts2, 'dwTimeStamp'):
            ts2 = ts2.dwTimeStamp

if ts1 == ts2:
            return

diff = ts1 - ts2
        days = abs(diff / 24.0)
        if ts1 == 0 or ts2 == 0:
            # when comparing to zero we don't want the number of days.
            msg = f"timestamp {ts1} != {ts2}"
        elif diff > 0:
            msg = f"{ts1} is {days} days ({diff} hours) after {ts2}"
        else:
            msg = f"{ts1} is {days} days ({abs(diff)} hours) before {ts2}"

raise AssertionError(msg)

def test_update_timestamps_aging_off_then_on(self):
        # we will add a record with aging off
        # it will have the current timestamp
        self.set_aging(False)
        name = 'timestamp-now'
        name2 = 'timestamp-eightdays'

rec = self.dns_update_record(name, [name])
        start_time = rec.dwTimeStamp
        self.assert_timestamp_in_ballpark(rec)
        # alter the timestamp -8 days using RPC
        # with aging turned off, we expect no change
        # when aging is on, we expect change
        eight_days_ago = start_time - 8 * 24
        rec = self.ldap_update_record(name2, [name2],
                                      dwTimeStamp=eight_days_ago)

self.assert_timestamps_equal(rec.dwTimeStamp, eight_days_ago)

# if aging was on, this would change
        rec = self.dns_update_record(name2, [name2])
        self.assert_timestamps_equal(rec.dwTimeStamp, eight_days_ago)

self.set_aging(True)
        rec = self.dns_update_record(name2, [name2])
        self.assertGreaterEqual(rec.dwTimeStamp, start_time)

def test_rpc_update_timestamps(self):
        # RPC always sets timestamps to zero on Windows.
        self.set_aging(False)
        name = 'timestamp-now'

rec = self.dns_update_record(name, [name])
        start_time = rec.dwTimeStamp
        self.assert_timestamp_in_ballpark(rec)
        # attempt to alter the timestamp to something close by.
        eight_days_ago = start_time - 8 * 24
        rec = self.rpc_update_record(name, [name],
                                     dwTimeStamp=eight_days_ago)
        self.assertEqual(rec.dwTimeStamp, 0)

# try again, with aging on
        self.set_aging(True)
        rec = self.rpc_update_record(name, [name],
                                     dwTimeStamp=eight_days_ago)
        self.assertEqual(rec.dwTimeStamp, 0)

# now that the record is static, a dns update won't change it
        rec = self.dns_update_record(name, [name])
        self.assertEqual(rec.dwTimeStamp, 0)

# but another record on the same node will behave normally
        # i.e. the node is not static, the record is.
        name2 = 'timestamp-eightdays'
        rec = self.dns_update_record(name2, [name2])
        self.assert_soon_after(rec.dwTimeStamp,
                               start_time)

def get_txt_timestamps(self, name, *txts):
        records = self.ldap_get_records(name)

ret = []
        for t in txts:
            for r in records:
                t2 = [x for x in r.data.str]
                if t == t2:
                    ret.append(r.dwTimeStamp)
        return ret

def test_update_aging_disabled_2(self):
        # With aging disabled, Windows updates the timestamps of all
        # records when one is updated.
        name = 'test'
        txt1 = ['test txt']
        txt2 = ['test', 'txt2']
        txt3 = ['test', 'txt3']

self.set_aging(False)

current_time = self.dns_update_record(name, txt1).dwTimeStamp

six_days_ago = current_time - 6 * 24
        eight_days_ago = current_time - 8 * 24
        fifteen_days_ago = current_time - 15 * 24
        hundred_days_ago = current_time - 100 * 24
        thousand_days_ago = current_time - 1000 * 24

for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago):
            # wind back
            self.ldap_update_record(name, txt1, dwTimeStamp=timestamp)
            self.assertEqual(self.get_txt_timestamps(name, txt1), [timestamp])

# no change here
            update_timestamp = self.dns_update_record(name, txt1).dwTimeStamp
            self.assert_timestamps_equal(update_timestamp, timestamp)

# adding a fresh record
        for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          100):
            # wind back
            timestamp1 = self.ldap_update_record(
                name,
                txt1,
                dwTimeStamp=timestamp).dwTimeStamp
            self.assert_timestamps_equal(timestamp1, timestamp)

self.dns_update_record(name, txt2)
            timestamps = self.get_txt_timestamps(name, txt1, txt2)
            self.assertEqual(timestamps, [timestamp, current_time])

self.ldap_delete_record(name, txt2)
            timestamps = self.get_txt_timestamps(name, txt1)
            self.assertEqual(timestamps, [timestamp])

# add record 2.
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          100):
            # wind back
            self.ldap_update_record(name, txt1, dwTimeStamp=timestamp)
            timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
            self.assert_timestamps_equal(timestamp1, timestamp)

timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
            # txt1 timestamp is now current time
            timestamps = self.get_txt_timestamps(name, txt1, txt2)
            self.assertEqual(timestamps, [timestamp, current_time])

# with 3 records, no change
        for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          10):
            # wind back
            self.ldap_update_record(name, txt1, dwTimeStamp=timestamp)
            self.ldap_update_record(name, txt2, dwTimeStamp=timestamp)
            self.ldap_update_record(name, txt3, dwTimeStamp=(timestamp + 30))
            timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
            self.assert_timestamps_equal(timestamp3, timestamp + 30)

self.dns_update_record(name, txt2).dwTimeStamp
            timestamps = self.get_txt_timestamps(name, txt1, txt2, txt3)
            self.assertEqual(timestamps, [timestamp,
                                          timestamp,
                                          timestamp + 30])

# with 3 records, one of which is static
        # first we set the updatee's timestamp to a recognisable number
        self.ldap_update_record(name, txt2, dwTimeStamp=999999)
        for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          10):
            # wind back
            self.ldap_update_record(name, txt1, dwTimeStamp=0)
            self.ldap_update_record(name, txt3, dwTimeStamp=(timestamp - 9))
            timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
            self.assert_timestamps_equal(timestamp3, timestamp - 9)

self.dns_update_record(name, txt2)
            timestamps = self.get_txt_timestamps(name, txt1, txt2, txt3)
            self.assertEqual(timestamps, [0,
                                          999999,
                                          timestamp - 9])

# with 3 records, updating one which is static
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          10):
            # wind back
            self.ldap_update_record(name, txt1, dwTimeStamp=0)
            self.ldap_update_record(name, txt2, dwTimeStamp=0)
            self.ldap_update_record(name, txt3, dwTimeStamp=(timestamp + 30))
            timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
            self.assert_timestamps_equal(timestamp3, timestamp + 30)

self.dns_update_record(name, txt2).dwTimeStamp
            timestamps = self.get_txt_timestamps(name, txt1, txt2, txt3)
            self.assertEqual(timestamps, [0,
                                          0,
                                          timestamp + 30])

# with 3 records, after the static nodes have been replaced
        self.ldap_update_record(name, txt1, dwTimeStamp=777777)
        self.ldap_update_record(name, txt2, dwTimeStamp=888888)
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        for timestamp in (current_time,
                          six_days_ago,
                          eight_days_ago,
                          fifteen_days_ago,
                          hundred_days_ago,
                          thousand_days_ago,
                          100000,
                          10):
            # wind back
            self.ldap_update_record(name, txt3, dwTimeStamp=(timestamp))
            timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
            self.assert_timestamps_equal(timestamp3, timestamp)

self.dns_update_record(name, txt2)
            timestamps = self.get_txt_timestamps(name, txt1, txt2, txt3)
            self.assertEqual(timestamps, [777777,
                                          888888,
                                          timestamp])

def _test_update_aging_disabled_n_days_ago(self, n_days):
        name = 'test'
        txt1 = ['1']
        txt2 = ['2']

self.set_aging(False)
        current_time = self.dns_update_record(name, txt1).dwTimeStamp

# rewind timestamp using ldap
        self.ldap_modify_timestamps(name, n_days * -24)
        n_days_ago = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assertGreater(current_time, n_days_ago)

# no change when updating this record
        update_timestamp = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(update_timestamp, n_days_ago)

# add another record, which should have the current timestamp
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

# get the original record timestamp. NOW it matches current_time
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp2)

# let's repeat that, this time with txt2 existing
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, n_days_ago)

# this update is not an add
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

# now timestamp1 is not changed
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, n_days_ago)

# delete record2, try again
        self.ldap_delete_record(name, txt2)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, n_days_ago)

# here we are re-adding the deleted record
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp

# It gets weird HERE.
        # note how the SIBLING of the deleted, re-added record differs
        # from the sibling of freshly added record, depending on the
        # time difference.
        if n_days <= 7:
            self.assert_timestamps_equal(timestamp1, n_days_ago)
        else:
            self.assert_timestamps_equal(timestamp1, timestamp2)

# re-timestamp record2, try again
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, n_days_ago)

# no change
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_timestamps_equal(timestamp2, n_days_ago)
        # also no change
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp2)

# let's introduce another record
        txt3 = ['3']
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp3, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp

if n_days <= 7:
            self.assert_timestamps_equal(timestamp1, n_days_ago)
        else:
            self.assert_timestamps_equal(timestamp1, timestamp3)

self.assert_timestamps_equal(timestamp2, timestamp3)

self.ldap_delete_record(name, txt3)
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp3, current_time)
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp

if n_days <= 7:
            self.assert_timestamps_equal(timestamp1, n_days_ago)
        else:
            self.assert_timestamps_equal(timestamp1, timestamp3)

self.assert_timestamps_equal(timestamp2, timestamp3)

# and here we'll make txt3 static
        txt4 = ['4']

# and here we'll make txt1 static
        self.ldap_update_record(name, txt1, dwTimeStamp=0)
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt3, dwTimeStamp=n_days_ago)
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        timestamp4 = self.dns_update_record(name, txt4).dwTimeStamp

self.assertEqual(timestamp1, 0)
        self.assert_timestamps_equal(timestamp2, n_days_ago)
        self.assert_timestamps_equal(timestamp3, n_days_ago)
        self.assert_soon_after(timestamp4, current_time)

def test_update_aging_disabled_in_no_refresh_window(self):
        self._test_update_aging_disabled_n_days_ago(4)

def test_update_aging_disabled_on_no_refresh_boundary(self):
        self._test_update_aging_disabled_n_days_ago(7)

def test_update_aging_disabled_in_refresh_window(self):
        self._test_update_aging_disabled_n_days_ago(9)

def test_update_aging_disabled_beyond_refresh_window(self):
        self._test_update_aging_disabled_n_days_ago(16)

def test_update_aging_disabled_in_eighteenth_century(self):
        self._test_update_aging_disabled_n_days_ago(100000)

def test_update_aging_disabled_static(self):
        name = 'test'
        txt1 = ['1']
        txt2 = ['2']

self.set_aging(False)

current_time = self.dns_update_record(name, txt1).dwTimeStamp
        self.ldap_update_record(name, txt1, dwTimeStamp=0)

# no change when updating this record
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assertEqual(timestamp1, 0)

# add another record, which should have the current timestamp
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_soon_after(timestamp1, current_time)

# let's repeat that, this time with txt2 existing
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

# delete record2, try again
        self.ldap_delete_record(name, txt2)
        self.ldap_update_record(name, txt1, dwTimeStamp=0)
        # no change when updating this record
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assertEqual(timestamp1, 0)

timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assertEqual(timestamp2, 0)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assertEqual(timestamp1, 0)
        # re-timestamp record2, try again
        self.ldap_update_record(name, txt2, dwTimeStamp=1)
        self.ldap_update_record(name, txt1, dwTimeStamp=0)
        # no change when updating this record
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_timestamps_equal(timestamp2, 1)

def test_update_aging_disabled(self):
        # With aging disabled, Windows updates the timestamps of all
        # records when one is updated.
        name = 'test'
        txt1 = ['test txt']
        txt2 = ['test', 'txt2']
        txt3 = ['test', 'txt3']
        minus_6 = -6 * 24
        minus_8 = -8 * 24

self.set_aging(False)

current_time = self.dns_update_record(name, txt1).dwTimeStamp

# rewind timestamp using ldap
        self.ldap_modify_timestamps(name, minus_6)
        after_mod = self.get_unique_txt_record(name, txt1)
        six_days_ago = after_mod.dwTimeStamp
        self.assert_timestamps_equal(six_days_ago, current_time + minus_6)

# no change
        update_timestamp = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(update_timestamp, six_days_ago)

self.check_query_txt(name, txt1, zone=self.zone)

# another record
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        # without aging, timestamp1 is changed!!
        self.assert_timestamps_equal(timestamp1, timestamp2)

# Set both records back to 8 days ago.
        self.ldap_modify_timestamps(name, minus_8)

eight_days_ago = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(eight_days_ago, current_time + minus_8)

update2 = self.dns_update_record(name, txt2)

# Without aging on, an update should not change the timestamps.
        self.assert_timestamps_equal(update2.dwTimeStamp, eight_days_ago)
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, eight_days_ago)

# Add another txt record. The new record should have the now
        # timestamp, and drag the others up with it.
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp3, current_time)
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp3)
        self.assert_timestamps_equal(timestamp2, timestamp3)

hundred_days_ago = current_time - 100 * 24
        thousand_days_ago = current_time - 1000 * 24
        record = self.ldap_update_record(name, txt1,
                                         dwTimeStamp=hundred_days_ago)
        self.assert_timestamps_equal(record.dwTimeStamp, hundred_days_ago)
        record = self.ldap_update_record(name, txt2,
                                         dwTimeStamp=thousand_days_ago)
        self.assert_timestamps_equal(record.dwTimeStamp, thousand_days_ago)

# update 3, will others change (because beyond RefreshInterval)? yes.
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp3, current_time)
        self.assert_timestamps_equal(timestamp1, hundred_days_ago)
        self.assert_timestamps_equal(timestamp2, thousand_days_ago)

fifteen_days_ago = current_time - 15 * 24
        self.ldap_update_record(name, txt3, dwTimeStamp=fifteen_days_ago)

timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        # DNS update has no effect because all records are old
        self.assert_timestamps_equal(timestamp2, thousand_days_ago)
        self.assert_timestamps_equal(timestamp1, hundred_days_ago)
        self.assert_timestamps_equal(timestamp3, fifteen_days_ago)

# Does update of old record affect timestamp of refreshable record? No.
        self.ldap_update_record(name, txt3, dwTimeStamp=eight_days_ago)
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        # DNS update has no effect because all records are old
        self.assert_timestamps_equal(timestamp2, thousand_days_ago)
        self.assert_timestamps_equal(timestamp1, hundred_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# RPC zeros timestamp, after which updates won't change it.
        # BUT it refreshes all others!
        self.rpc_update_record(name, txt2)

timestamp2 = self.dns_update_record(name, txt3).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        self.assertEqual(timestamp2, 0)
        self.assert_soon_after(timestamp1, current_time)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

def test_update_aging_enabled(self):
        name = 'test'
        txt1 = ['test txt']
        txt2 = ['test', 'txt2']
        txt3 = ['test', 'txt3']
        txt4 = ['4']

self.set_aging(True)

current_time = self.dns_update_record(name, txt2).dwTimeStamp

six_days_ago = current_time - 6 * 24
        eight_days_ago = current_time - 8 * 24
        fifteen_days_ago = current_time - 15 * 24
        hundred_days_ago = current_time - 100 * 24

self.ldap_update_record(name, txt1, dwTimeStamp=six_days_ago)

# with or without aging, a delta of -6 days does not affect
        # timestamps, because dwNoRefreshInterval is 7 days.
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp

self.assert_timestamps_equal(timestamp1, six_days_ago)
        self.assert_soon_after(timestamp2, current_time)

self.ldap_update_record(name, txt3, dwTimeStamp=eight_days_ago)
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# update 1, what happens to 2 and 3? Nothing?
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, six_days_ago)
        self.assert_soon_after(timestamp2, current_time)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# now set 1 to 8 days, and we should see changes
        self.ldap_update_record(name, txt1, dwTimeStamp=eight_days_ago)

# update 1, what happens to 2 and 3? Nothing?
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp1, current_time)
        self.assert_soon_after(timestamp2, current_time)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# next few ones use these numbers
        self.ldap_update_record(name, txt1, dwTimeStamp=fifteen_days_ago)
        self.ldap_update_record(name, txt2, dwTimeStamp=six_days_ago)
        self.ldap_update_record(name, txt3, dwTimeStamp=eight_days_ago)

# change even though 1 is outside the window
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp1, current_time)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# reset 1
        self.ldap_update_record(name, txt1, dwTimeStamp=fifteen_days_ago)

# no change, because 2 is outside the window
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, fifteen_days_ago)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# 3 changes, others do not
        timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, fifteen_days_ago)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_soon_after(timestamp3, current_time)

# reset 3 to 100 days
        self.ldap_update_record(name, txt3, dwTimeStamp=hundred_days_ago)

# reset 1 and 3 to 8 days. does update of 1 affect 3?
        self.ldap_update_record(name, txt1, dwTimeStamp=eight_days_ago)
        self.ldap_update_record(name, txt3, dwTimeStamp=eight_days_ago)

# 1 changes, others do not
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp1, current_time)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# Try an RPC update, zeroing 1 --> what happens to 3?
        timestamp1 = self.rpc_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assertEqual(timestamp1, 0)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

# with 2 and 3 at 8 days, does static record change things?
        self.ldap_update_record(name, txt2, dwTimeStamp=eight_days_ago)
        # 2 changes, but to zero!
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, 0)
        self.assert_timestamps_equal(timestamp2, 0)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)

self.ldap_update_record(name, txt2, dwTimeStamp=six_days_ago)
        self.ldap_update_record(name, txt1, dwTimeStamp=3000000)
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, 3000000)

# dns update remembers that node is static, even with no
        # static records.
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assertEqual(timestamp1, 0)

# Add another txt record. The new record should have the now
        # timestamp, and the others should remain unchanged.
        # BUT somehow record 1 is static!?
        timestamp4 = self.dns_update_record(name, txt4).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, 0)
        self.assert_timestamps_equal(timestamp2, six_days_ago)
        self.assert_timestamps_equal(timestamp3, eight_days_ago)
        self.assert_timestamps_equal(timestamp4, 0)

def _test_update_aging_enabled_n_days_ago(self, n_days):
        name = 'test'
        txt1 = ['1']
        txt2 = ['2']
        delta = n_days * -24

self.set_aging(True)
        current_time = self.dns_update_record(name, txt1).dwTimeStamp

# rewind timestamp using ldap
        self.ldap_modify_timestamps(name, delta)
        n_days_ago = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assertGreater(current_time, n_days_ago)

# update changes timestamp depending on time.
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        if n_days <= 7:
            self.assert_timestamps_equal(timestamp1, n_days_ago)
        else:
            self.assert_soon_after(timestamp1, current_time)

# add another record, which should have the current timestamp
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

# first record should not have changed
        timestamp1_b = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp1_b)

# let's repeat that, this time with txt2 existing
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp1_b)

# this update is not an add. record 2 is already up-to-date
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

# now timestamp1 is not changed
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp1_b)

# delete record2, try again
        self.ldap_delete_record(name, txt2)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        if n_days <= 7:
            self.assert_timestamps_equal(timestamp1, n_days_ago)
        else:
            self.assert_soon_after(timestamp1, current_time)

# here we are re-adding the deleted record
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_soon_after(timestamp2, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp

# re-timestamp record2, try again
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

# this should make no difference
        timestamp1_b = self.dns_update_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp1_b)

# no change
        timestamp2 = self.dns_update_record(name, txt2).dwTimeStamp
        self.assert_timestamps_equal(timestamp2, timestamp1)
        # also no change
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, timestamp2)

# let's introduce another record
        txt3 = ['3']
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt1, dwTimeStamp=n_days_ago)

timestamp3 = self.dns_update_record(name, txt3).dwTimeStamp
        self.assert_soon_after(timestamp3, current_time)

timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp

self.assert_timestamps_equal(timestamp1, n_days_ago)
        self.assert_timestamps_equal(timestamp2, n_days_ago)

txt4 = ['4']

# Because txt1 is static, txt4 is static
        self.ldap_update_record(name, txt1, dwTimeStamp=0)
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt3, dwTimeStamp=n_days_ago)
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        timestamp4 = self.dns_update_record(name, txt4).dwTimeStamp

self.assert_timestamps_equal(timestamp1, 0)
        self.assert_timestamps_equal(timestamp2, n_days_ago)
        self.assert_timestamps_equal(timestamp3, n_days_ago)
        self.assert_timestamps_equal(timestamp4, 0)

longer_ago = n_days_ago // 2

# remove all static records.
        self.ldap_delete_record(name, txt4)
        self.ldap_update_record(name, txt1, dwTimeStamp=longer_ago)
        self.ldap_update_record(name, txt2, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, txt3, dwTimeStamp=n_days_ago)
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, longer_ago)

timestamp4 = self.dns_update_record(name, txt4).dwTimeStamp
        timestamp2 = self.get_unique_txt_record(name, txt2).dwTimeStamp
        timestamp3 = self.get_unique_txt_record(name, txt3).dwTimeStamp
        timestamp1 = self.get_unique_txt_record(name, txt1).dwTimeStamp

# Here, although there is no record frm which to get the zero
        # timestamp, record 4 does it anyway.
        self.assert_timestamps_equal(timestamp1, longer_ago)
        self.assert_timestamps_equal(timestamp2, n_days_ago)
        self.assert_timestamps_equal(timestamp3, n_days_ago)
        self.assert_timestamps_equal(timestamp4, 0)

# and now record 1 wants to be static.
        self.ldap_update_record(name, txt4, dwTimeStamp=longer_ago)
        timestamp4 = self.get_unique_txt_record(name, txt4).dwTimeStamp
        self.assert_timestamps_equal(timestamp4, longer_ago)
        timestamp1 = self.dns_update_record(name, txt1).dwTimeStamp
        timestamp4 = self.get_unique_txt_record(name, txt4).dwTimeStamp
        self.assert_timestamps_equal(timestamp1, 0)
        self.assert_timestamps_equal(timestamp4, longer_ago)

def test_update_aging_enabled_in_no_refresh_window(self):
        self._test_update_aging_enabled_n_days_ago(4)

def test_update_aging_enabled_on_no_refresh_boundary(self):
        self._test_update_aging_enabled_n_days_ago(7)

def test_update_aging_enabled_in_refresh_window(self):
        self._test_update_aging_enabled_n_days_ago(9)

def test_update_aging_enabled_beyond_refresh_window(self):
        self._test_update_aging_enabled_n_days_ago(16)

def test_update_aging_enabled_in_eighteenth_century(self):
        self._test_update_aging_enabled_n_days_ago(100000)

def test_update_static_stickiness(self):
        name = 'test'
        A = ['A']
        B = ['B']
        C = ['C']
        D = ['D']

self.set_aging(False)
        self.dns_update_record(name, A).dwTimeStamp
        self.ldap_update_record(name, B, dwTimeStamp=0)
        self.dns_update_record(name, B)
        self.dns_update_record(name, C)
        ctime = self.get_unique_txt_record(name, C).dwTimeStamp
        self.assertEqual(ctime, 0)
        btime = self.get_unique_txt_record(name, B).dwTimeStamp
        self.assertEqual(btime, 0)

self.ldap_replace_records(name, [])

self.dns_update_record(name, D)
        dtime = self.get_unique_txt_record(name, D).dwTimeStamp
        self.assertEqual(dtime, 0)

def _test_update_timestamp_weirdness(self, n_days, aging=True):
        name = 'test'
        A = ['A']
        B = ['B']

self.set_aging(aging)

current_time = self.dns_update_record(name, A).dwTimeStamp

# rewind timestamp using ldap
        self.ldap_modify_timestamps(name, n_days * -24)
        n_days_ago = self.get_unique_txt_record(name, A).dwTimeStamp
        time_A = self.dns_update_record(name, A).dwTimeStamp
        # that dns_update should have reset the timestamp ONLY if
        # aging is on and the old timestamp is > noRefresh period (7
        # days)
        if n_days > 7 and aging:
            self.assert_soon_after(time_A, current_time)
        else:
            self.assert_timestamps_equal(time_A, n_days_ago)

# add another record, which should have the current timestamp
        time_B = self.dns_update_record(name, B).dwTimeStamp
        self.assert_soon_after(time_B, current_time)

time_A = self.get_unique_txt_record(name, A).dwTimeStamp
        if aging and n_days <= 7:
            self.assert_timestamps_equal(time_A, n_days_ago)
        else:
            self.assert_soon_after(time_A, current_time)

# delete B, try again
        self.ldap_delete_record(name, B)
        self.ldap_update_record(name, A, dwTimeStamp=n_days_ago)

time_A = self.dns_update_record(name, A).dwTimeStamp

# here we are re-adding the deleted record
        time_B = self.dns_update_record(name, B).dwTimeStamp
        self.assert_soon_after(time_B, current_time)

time_A = self.get_unique_txt_record(name, A).dwTimeStamp
        return n_days_ago, time_A, time_B

def test_update_timestamp_weirdness_no_refresh_no_aging(self):
        n_days_ago, time_A, time_B = \
            self._test_update_timestamp_weirdness(5, False)
        # the timestamp of the SIBLING of the deleted, re-added record
        # differs from the sibling of freshly added record.
        self.assert_timestamps_equal(time_A, n_days_ago)

def test_update_timestamp_weirdness_no_refresh_aging(self):
        n_days_ago, time_A, time_B = \
            self._test_update_timestamp_weirdness(5, True)
        # the timestamp of the SIBLING of the deleted, re-added record
        # differs from the sibling of freshly added record.
        self.assert_timestamps_equal(time_A, n_days_ago)

def test_update_timestamp_weirdness_refresh_no_aging(self):
        n_days_ago, time_A, time_B = \
            self._test_update_timestamp_weirdness(9, False)
        self.assert_timestamps_equal(time_A, time_B)

def test_update_timestamp_weirdness_refresh_aging(self):
        n_days_ago, time_A, time_B = \
            self._test_update_timestamp_weirdness(9, True)
        self.assert_timestamps_equal(time_A, time_B)

def test_aging_refresh(self):
        name, txt = 'agingtest', ['test txt']
        no_refresh = 200
        refresh = 160
        self.set_zone_int_params(NoRefreshInterval=no_refresh,
                                 RefreshInterval=refresh,
                                 Aging=1)
        before_mod = self.dns_update_record(name, txt)
        start_time = before_mod.dwTimeStamp

# go back 86 hours, which is in the no-refresh time (but
        # wouldn't be if we had stuck to the default of 168).
        self.ldap_modify_timestamps(name, -170)
        rec = self.dns_update_record(name, txt)
        self.assert_timestamps_equal(rec.dwTimeStamp,
                                     start_time - 170)

# back to -202 hours, into the refresh zone
        # the update should reset the timestamp to now.
        self.ldap_modify_timestamps(name, -32)
        rec = self.dns_update_record(name, txt)
        self.assert_soon_after(rec.dwTimeStamp, start_time)

# back to -362 hours, beyond the end of the refresh period.
        # Actually nothing changes at this time -- we can still
        # refresh, but the record is liable for scavenging.
        self.ldap_modify_timestamps(name, -160)
        rec = self.dns_update_record(name, txt)
        self.assert_soon_after(rec.dwTimeStamp, start_time)

def test_add_no_timestamp(self):
        # check zero timestamp is implicit
        self.set_aging(True)
        rec = self.ldap_update_record('ldap', 'test')
        self.assertEqual(rec.dwTimeStamp, 0)
        rec = self.rpc_update_record('rpc', 'test')
        self.assertEqual(rec.dwTimeStamp, 0)

def test_add_zero_timestamp(self):
        rec = self.ldap_update_record('ldap', 'test', dwTimeStamp=0)
        self.assertEqual(rec.dwTimeStamp, 0)
        rec = self.rpc_update_record('rpc', 'test', dwTimeStamp=0)
        self.assertEqual(rec.dwTimeStamp, 0)

def test_add_update_timestamp(self):
        # LDAP can change timestamp, RPC can't
        rec = self.ldap_update_record('ldap', 'test', dwTimeStamp=123456)
        self.assertEqual(rec.dwTimeStamp, 123456)
        rec = self.rpc_update_record('rpc', 'test', dwTimeStamp=123456)
        self.assertEqual(rec.dwTimeStamp, 0)
        # second time is a different code path (add vs update)
        rec = self.rpc_update_record('rpc', 'test', dwTimeStamp=123456)
        self.assertEqual(rec.dwTimeStamp, 0)
        # RPC update the one with timestamp, zeroing it.
        rec = self.rpc_update_record('ldap', 'test', dwTimeStamp=123456)
        self.assertEqual(rec.dwTimeStamp, 0)

def test_add_update_ttl(self):
        # RPC *can* set dwTtlSeconds.
        rec = self.ldap_update_record('ldap', 'test',
                                      dwTtlSeconds=1234)
        self.assertEqual(rec.dwTtlSeconds, 1234)
        rec = self.rpc_update_record('rpc', 'test', dwTtlSeconds=1234)
        self.assertEqual(rec.dwTtlSeconds, 1234)
        # does update work like add?
        rec = self.rpc_update_record('rpc', 'test', dwTtlSeconds=4321)
        self.assertEqual(rec.dwTtlSeconds, 4321)
        rec = self.rpc_update_record('ldap', 'test', dwTtlSeconds=5678)
        self.assertEqual(rec.dwTtlSeconds, 5678)

def test_add_update_ttl_serial(self):
        # when setting dwTtlSeconds, what happens to serial number?
        rec = self.ldap_update_record('ldap', 'test',
                                      dwTtlSeconds=1234,
                                      dwSerial=123)
        self.assertEqual(rec.dwTtlSeconds, 1234)
        self.assertEqual(rec.dwSerial, 123)
        rec = self.rpc_update_record('rpc', 'test', dwTtlSeconds=1234)
        self.assertEqual(rec.dwTtlSeconds, 1234)
        serial = rec.dwSerial
        self.assertLess(serial, 4)
        rec = self.rpc_update_record('rpc', 'test', dwTtlSeconds=4321)
        self.assertEqual(rec.dwTtlSeconds, 4321)
        self.assertEqual(rec.dwSerial, serial + 1)
        rec = self.rpc_update_record('ldap', 'test', dwTtlSeconds=5678)
        self.assertEqual(rec.dwTtlSeconds, 5678)
        self.assertEqual(rec.dwSerial, 124)

def test_add_update_dwFlags(self):
        # dwFlags splits into rank and flags.
        # according to [MS-DNSP] 2.3.2.2, flags MUST be zero
        rec = self.ldap_update_record('ldap', 'test', flags=22222, rank=222)
        self.assertEqual(rec.flags, 22222)
        self.assertEqual(rec.rank, 222)

rec = self.rpc_update_record('ldap', 'test', dwFlags=3333333)
        # rank != 3333333 & 0xff == 213
        self.assertEqual(rec.rank, 240)    # RPC fixes rank
        self.assertEqual(rec.flags, 0)

self.assertRaises(OverflowError,
                          self.ldap_update_record,
                          'ldap', 'test', flags=777777777, rank=777)

# reset to no default (rank overflows)
        rec = self.ldap_update_record('ldap', 'test', flags=7777, rank=777)
        self.assertEqual(rec.flags, 7777)
        self.assertEqual(rec.rank, 9)

# DNS update zeros flags, sets rank to 240 (RANK_ZONE)
        rec = self.dns_update_record('ldap', 'test', ttl=999)
        self.assertEqual(rec.flags, 0)
        self.assertEqual(rec.rank, 240)

rec = self.rpc_update_record('ldap', 'test', dwFlags=321)
        self.assertEqual(rec.flags, 0)
        self.assertEqual(rec.rank, 240)

# RPC adding a new record: fixed rank, zero flags
        rec = self.rpc_update_record('ldap', 'test 2', dwFlags=12345)
        self.assertEqual(rec.rank, 240)
        self.assertEqual(rec.flags, 0)

def test_add_update_dwReserved(self):
        # RPC does not change dwReserved.
        rec = self.ldap_update_record('ldap', 'test', dwReserved=54321)
        self.assertEqual(rec.dwReserved, 54321)
        rec = self.rpc_update_record('rpc', 'test', dwReserved=54321)
        self.assertEqual(rec.dwReserved, 0)
        rec = self.rpc_update_record('rpc', 'test', dwReserved=54321)
        self.assertEqual(rec.dwReserved, 0)
        rec = self.rpc_update_record('ldap', 'test', dwReserved=12345)
        self.assertEqual(rec.dwReserved, 54321)

def test_add_update_dwSerial(self):
        # On Windows the RPC record ends up with serial 2, on Samba
        # serial 3. Rather than knownfail this, we accept anything
        # below 4 (for now).
        rec = self.ldap_update_record('ldap', 'test', dwSerial=123)
        self.assertEqual(rec.dwSerial, 123)
        rec = self.rpc_update_record('rpc', 'test', dwSerial=123)
        self.assertLess(rec.dwSerial, 4)
        rec = self.rpc_update_record('rpc', 'test', dwSerial=123)
        self.assertLess(rec.dwSerial, 4)
        rec = self.dns_update_record('rpc', 'test')
        self.assertLess(rec.dwSerial, 4)
        rec = self.dns_update_record('dns-0', 'test')
        self.assertLess(rec.dwSerial, 5)

rec = self.dns_update_record('ldap', 'test')
        self.assertEqual(rec.dwSerial, 123)
        rec = self.rpc_update_record('ldap', 'test', dwSerial=123)
        self.assertEqual(rec.dwSerial, 123)
        rec = self.ldap_update_record('ldap', 'test', dwSerial=12)
        self.assertEqual(rec.dwSerial, 12)
        # when we dns-updated ldap/test, we alerted Windows to 123 as
        # a high water mark for the zone. (even though we have since
        # dropped the serial to 12, 123 is the base serial for new
        # records).
        rec = self.dns_update_record('dns', 'test')
        self.assertEqual(rec.dwSerial, 124)
        rec = self.dns_update_record('dns2', 'test')
        self.assertEqual(rec.dwSerial, 125)
        rec = self.rpc_update_record('rpc2', 'test')
        self.assertEqual(rec.dwSerial, 126)
        rec = self.dns_update_record('dns', 'test 2')
        self.assertEqual(rec.dwSerial, 127)

def test_add_update_dwSerial_2(self):
        # On Samba the RPC update resets the serial to a low number,
        # while Windows leaves it high.
        rec = self.ldap_update_record('ldap', 'test', dwSerial=123)
        self.assertEqual(rec.dwSerial, 123)
        rec = self.rpc_update_record('ldap', 'test', dwSerial=321)
        self.assertEqual(rec.dwSerial, 123)
        rec = self.dns_update_record('ldap', 'test')
        self.assertEqual(rec.dwSerial, 123)

def test_rpc_update_disparate_types(self):
        """Can we use update to replace a TXT with an AAAA?"""
        name = 'x'
        old = TXTRecord("x")
        new = ARecord("127.0.0.111")
        self.rpc_replace(name, None, old)
        recs = self.ldap_get_records(name)
        self.assertEqual(len(recs), 1)
        self.assertEqual(recs[0].wType, old.wType)

self.rpc_replace(name, old, new)
        recs = self.ldap_get_records(name)
        self.assertEqual(len(recs), 1)
        self.assertEqual(recs[0].wType, new.wType)

def test_add_update_many(self):
        # Samba fails often in this set, but we want to see how it
        # goes further down, so we print the problems and defer the
        # failure.
        failures = 0
        total = 0

def _defer_wrap(f):
            def _defer(*args):
                nonlocal failures, total
                total += 1
                try:
                    f(*args)
                except self.failureException as e:
                    from traceback import format_stack
                    print(f"{format_stack()[-2]} {e}\n")
                    failures += 1
            return _defer

defer_assertEqual = _defer_wrap(self.assertEqual)
        defer_assert_timestamp_in_ballpark = \
            _defer_wrap(self.assert_timestamp_in_ballpark)

self.set_aging(False)
        rec = self.ldap_update_record('ldap', 'test',
                                      version=11,
                                      rank=22,
                                      flags=33,
                                      dwSerial=44,
                                      dwTtlSeconds=55,
                                      dwReserved=66,
                                      dwTimeStamp=77)

self.assertEqual(rec.version, 5)  # disobeys request
        self.assertEqual(rec.rank, 22)
        self.assertEqual(rec.flags, 33)
        self.assertEqual(rec.dwSerial, 44)
        self.assertEqual(rec.dwTtlSeconds, 55)
        self.assertEqual(rec.dwReserved, 66)
        self.assertEqual(rec.dwTimeStamp, 77)
        # DNS updates first
        rec = self.dns_update_record('ldap', 'test', ttl=999)
        self.assertEqual(rec.version, 5)
        self.assertEqual(rec.rank, 240)          # rank gets fixed by DNS update
        defer_assertEqual(rec.flags, 0)          # flags gets fixed
        defer_assertEqual(rec.dwSerial, 45)      # serial increments
        self.assertEqual(rec.dwTtlSeconds, 999)  # TTL set
        defer_assertEqual(rec.dwReserved, 0)     # reserved fixed
        defer_assert_timestamp_in_ballpark(rec)  # changed on Windows ?!

self.set_aging(True)
        rec = self.dns_update_record('ldap', 'test', ttl=1111)
        self.assertEqual(rec.version, 5)
        self.assertEqual(rec.rank, 240)
        defer_assertEqual(rec.flags, 0)
        defer_assertEqual(rec.dwSerial, 46)
        self.assertEqual(rec.dwTtlSeconds, 1111)  # TTL set
        defer_assertEqual(rec.dwReserved, 0)
        self.assert_timestamp_in_ballpark(rec)

# RPC update
        rec = self.rpc_update_record('ldap', 'test',
                                     version=111,
                                     dwFlags=333,
                                     dwSerial=444,
                                     dwTtlSeconds=555,
                                     dwReserved=666,
                                     dwTimeStamp=777)

self.assertEqual(rec.version, 5)         # no change
        self.assertEqual(rec.rank, 240)          # no change
        defer_assertEqual(rec.flags, 0)           # no change
        defer_assertEqual(rec.dwSerial, 47)       # Serial increments
        self.assertEqual(rec.dwTtlSeconds, 555)  # TTL set
        defer_assertEqual(rec.dwReserved, 0)      # no change
        self.assertEqual(rec.dwTimeStamp, 0)     # timestamp zeroed

# RPC update, using default values
        rec = self.rpc_update_record('ldap', 'test')
        self.assertEqual(rec.version, 5)
        self.assertEqual(rec.rank, 240)
        defer_assertEqual(rec.flags, 0)
        defer_assertEqual(rec.dwSerial, 48)       # serial increments
        self.assertEqual(rec.dwTtlSeconds, 900)  # TTL changed
        defer_assertEqual(rec.dwReserved, 0)
        self.assertEqual(rec.dwTimeStamp, 0)

self.set_aging(False)
        rec = self.dns_update_record('ldap', 'test', ttl=888)
        self.assertEqual(rec.version, 5)
        self.assertEqual(rec.rank, 240)
        defer_assertEqual(rec.flags, 0)
        defer_assertEqual(rec.dwSerial, 49)       # serial increments
        self.assertEqual(rec.dwTtlSeconds, 888)  # TTL set
        defer_assertEqual(rec.dwReserved, 0)
        self.assertEqual(rec.dwTimeStamp, 0)     # timestamp stays zero

if failures:
            self.fail(f"failed {failures}/{total} defered assertions")

def test_static_record_dynamic_update(self):
        """Add a static record, then a dynamic record.
        The dynamic record should have a timestamp set."""
        name = 'test'
        txt = ['static txt']
        txt2 = ['dynamic txt']
        self.set_aging(True)
        rec = self.ldap_update_record(name, txt, dwTimeStamp=0)
        rec2 = self.dns_update_record(name, txt2)
        self.assert_timestamp_in_ballpark(rec2)
        ts2 = rec2.dwTimeStamp
        # update the first record. It should stay static (timestamp 0)
        rec = self.dns_update_record(name, txt)
        self.assertEqual(rec.dwTimeStamp, 0)
        # and rec2 should be unchanged.
        self.assertEqual(rec2.dwTimeStamp, ts2)

def test_dynamic_record_static_update(self):
        name = 'agingtest'
        txt1 = ['dns update before']
        txt2 = ['ldap update']
        txt3 = ['dns update after']
        self.set_aging(True)

self.dns_update_record(name, txt1)
        self.ldap_update_record(name, txt2)
        self.dns_update_record(name, txt3)

recs = self.get_rpc_records(name)
        for r in recs:
            d = [x.str for x in r.data.str]
            if d == txt1:
                self.assertNotEqual(r.dwTimeStamp, 0)
            elif d == txt2:
                self.assertEqual(r.dwTimeStamp, 0)
            elif d == txt3:
                self.assertNotEqual(r.dwTimeStamp, 0)

def test_tombstone_in_hours_and_nttime(self):
        # Until now Samba has measured tombstone timestamps in hours,
        # not ten-millionths of a second. After now, we want Samba to
        # handle both.

nh, oh, nn, on, on0, onf, nn0, nnf, _1601 = 'abcdefgij'
        now_hours = dsdb_dns.unix_to_dns_timestamp(int(time.time()))
        old_hours = now_hours - 24 * 90
        now_nttime = dsdb_dns.dns_timestamp_to_nt_time(now_hours)
        old_nttime = dsdb_dns.dns_timestamp_to_nt_time(old_hours)
        # calculations on hours might be based on the lower 32 bits,
        # so we test with these forced to extremes (the maximum change
        # is 429 seconds in NTTIME).
        old_nttime0 = old_nttime & 0xffffffff00000000
        old_nttimef = old_nttime | 0xffffffff
        now_nttime0 = now_nttime & 0xffffffff00000000
        now_nttimef = now_nttime | 0xffffffff
        self.dns_tombstone(nh, epoch_nttime=now_hours)
        self.dns_tombstone(oh, epoch_nttime=old_hours)
        self.dns_tombstone(nn, epoch_nttime=now_nttime)
        self.dns_tombstone(on, epoch_nttime=old_nttime)
        self.dns_tombstone(nn0, epoch_nttime=now_nttime0)
        self.dns_tombstone(nnf, epoch_nttime=now_nttimef)
        self.dns_tombstone(on0, epoch_nttime=old_nttime0)
        self.dns_tombstone(onf, epoch_nttime=old_nttimef)
        # this is our (arbitrary) threshold that will make us think in
        # NTTIME, not hours.
        self.dns_tombstone(_1601, epoch_nttime=(10 * 1000 * 1000 + 1))

try:
            file_samdb = get_file_samdb()
        except ldb.LdbError as e:
            raise AssertionError(
                f"failing because '{e}': this is Windows?") from None
        dsdb._dns_delete_tombstones(file_samdb)

# nh and nn should not be deleted
        for name in nh, nn, nn0, nnf:
            recs = self.ldap_get_records(name)
            self.assertEqual(len(recs), 1)
            self.assert_tombstoned(name, timestamp=False)

# oh and on should be GONE
        for name in oh, on, on0, onf, _1601:
            recs = self.ldap_get_records(name)
            self.assertEqual(len(recs), 0)

def test_dns_query_for_tombstoned_results(self):
        # This one fails on Windows, because the dns cache holds B
        # after it has been tombstoned behind its back.
        A = 'a'
        B = 'b'
        self.dns_tombstone(A)
        self.assert_tombstoned(A)
        r = self.dns_query(A, qtype=dns.DNS_QTYPE_TXT)
        self.assertEqual(r.ancount, 0)

self.dns_update_record(B, B)
        self.dns_tombstone(B)
        self.assert_tombstoned(B)
        r = self.dns_query(B, qtype=dns.DNS_QTYPE_TXT)
        self.assertEqual(r.ancount, 0)

def test_basic_scavenging(self):
        # NOTE: This one fails on Windows, because the RPC call to
        # prompt scavenging is not immediate. On Samba, in the
        # testenv, we don't have the RPC call but we can connect to
        # the database directly.

# just to be sure we have the right limits.
        self.set_zone_int_params(NoRefreshInterval=168,
                                 RefreshInterval=168,
                                 Aging=1)

ts1, ts2, ts3, ts4, ts5, ts6 = ('1', '2', '3', '4', '5', '6')
        self.dns_update_record(ts1, ts1)
        self.dns_update_record(ts2, ts2)
        # ts2 is tombstoned and timestamped in 1981
        self.dns_tombstone(ts2)
        # ts3 is tombstoned and timestamped in the future
        self.dns_tombstone(ts3, epoch_hours=(DNS_TIMESTAMP_2101 - 1))
        # ts4 is tombstoned and timestamped in the past
        self.dns_tombstone(ts4, epoch_hours=1111111)
        # ts5 is tombstoned in the past and timestamped in the future
        self.dns_tombstone(ts5, epoch_hours=5555555, epoch_nttime=int(1e10))

# ts2 and ts3 should now be tombstoned.
        self.assert_tombstoned(ts2)
        self.assert_tombstoned(ts3)

# let's un-tombstone ts2
        # ending up with dnsTombstoned: FALSE in Samba
        # and no dNSTombstoned in Windows.
        self.dns_update_record(ts2, "ts2 untombstoned")
        ts2_node = self.get_one_node(ts2)
        ts2_tombstone = ts2_node.get("dNSTombstoned")
        if ts2_tombstone is not None:
            self.assertEqual(ts2_tombstone[0], b"FALSE")

self.assert_tombstoned(ts2, tombstoned=False)

r = self.dns_update_record(ts6, ts6)

# put some records into the death zone.
        self.ldap_modify_timestamps(ts1, -15 * 24)
        self.ldap_modify_timestamps(ts2, -14 * 24 - 2)
        self.ldap_modify_timestamps(ts6, -14 * 24 + 2)

# ts1 will be saved by this record
        self.dns_update_record(ts1, "another record")

try:
            # Tell the server to clean-up records.
            # This is how it *should* work on Windows:
            self.rpc_conn.DnssrvOperation2(
                dnsserver.DNS_CLIENT_VERSION_LONGHORN,
                0,
                SERVER_IP,
                None,
                0,
                "StartScavenging",
                dnsserver.DNSSRV_TYPEID_NULL,
                None)
            # Samba won't get here (NOT_IMPLEMENTED error)
            # wait for Windows to do its cleanup.
            time.sleep(2)
        except WERRORError as e:
            if e.args[0] == werror.WERR_CALL_NOT_IMPLEMENTED:
                # This is the Samba way, talking to the file directly,
                # as if we were the server process. The direct
                # connection is needed because the tombstoning search
                # involves a magic system only filter.
                file_samdb = get_file_samdb()
                dsdb._scavenge_dns_records(file_samdb)
                dsdb._dns_delete_tombstones(file_samdb)
            else:
                raise

# Now what we should have:
        # ts1: alive: the old record is deleted, the new one not.
        # ts2: tombstoned
        # ts3: tombstoned
        # ts4: deleted. gone.
        # ts5: deleted. timestamp affects tombstoning, but not deletion.
        # ts6: alive
        #
        # We order our assertions to make the windows test
        # fail as late as possible (on ts4, ts5, ts2).
        r = self.get_unique_txt_record(ts1, ["another record"])
        self.assertIsNotNone(r)
        r = self.get_unique_txt_record(ts6, [ts6])
        self.assertIsNotNone(r)

self.assert_tombstoned(ts3)

n = self.get_one_node(ts4)
        self.assertIsNone(n)
        n = self.get_one_node(ts5)
        self.assertIsNone(n)

self.assert_tombstoned(ts2)

def test_samba_scavenging(self):
        # We expect this one to fail on Windows, because scavenging
        # and tombstoning cannot be performed on demand.

try:
            file_samdb = get_file_samdb()
        except ldb.LdbError as e:
            raise AssertionError(
                f"failing because '{e}': this is Windows?") from None

# let's try different limits.
        self.set_zone_int_params(NoRefreshInterval=30,
                                 RefreshInterval=20,
                                 Aging=1)

now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))

A, B, C, D = 'ABCD'
        # A has current time
        # B has safe, non-updateable time
        # C has safe time
        # D is scavengeable
        atime = self.dns_update_record(A, A).dwTimeStamp
        btime = self.ldap_update_record(B, B, dwTimeStamp=now-20).dwTimeStamp
        ctime = self.ldap_update_record(C, C, dwTimeStamp=now-40).dwTimeStamp
        dtime = self.ldap_update_record(D, D, dwTimeStamp=now-60).dwTimeStamp
        self.assert_soon_after(atime, now)
        self.assert_timestamps_equal(btime, now-20)
        self.assert_timestamps_equal(ctime, now-40)
        self.assert_timestamps_equal(dtime, now-60)

dsdb._scavenge_dns_records(file_samdb)

# D should be gone (tombstoned)
        r = self.get_unique_txt_record(D, D)
        self.assertIsNone(r)
        r = self.dns_query(D, qtype=dns.DNS_QTYPE_TXT)
        self.assertEqual(r.ancount, 0)
        recs = self.ldap_get_records(D)
        self.assertEqual(len(recs), 1)
        self.assert_tombstoned(recs[0])

# others unchanged.
        atime = self.get_unique_txt_record(A, A).dwTimeStamp
        btime = self.get_unique_txt_record(B, B).dwTimeStamp
        ctime = self.get_unique_txt_record(C, C).dwTimeStamp
        self.assert_soon_after(atime, now)
        self.assert_timestamps_equal(btime, now-20)
        self.assert_timestamps_equal(ctime, now-40)

btime = self.dns_update_record(B, B).dwTimeStamp
        ctime = self.dns_update_record(C, C).dwTimeStamp
        self.assert_timestamps_equal(btime, now-40)
        self.assert_soon_after(ctime, now)

# after this, D *should* still be a tombstone, because its
        # tombstone timestamp is not very old.
        dsdb._dns_delete_tombstones(file_samdb)
        recs = self.ldap_get_records(D)
        self.assertEqual(len(recs), 1)
        self.assert_tombstoned(recs[0])

# Let's delete C using rpc, and ensure it survives dns_delete_tombstones
        self.rpc_delete_txt(C, C)
        recs = self.ldap_get_records(C)
        self.assertEqual(len(recs), 1)
        self.assert_tombstoned(recs[0])
        dsdb._dns_delete_tombstones(file_samdb)
        recs = self.ldap_get_records(C)
        self.assertEqual(len(recs), 1)
        self.assert_tombstoned(recs[0])

# now let's wind A and B back to either side of the two week
        # threshold. A should survive, B should not.
        self.dns_tombstone(A, (now - 166))
        self.dns_tombstone(B, (now - 170))
        dsdb._dns_delete_tombstones(file_samdb)

recs = self.ldap_get_records(A)
        self.assertEqual(len(recs), 1)
        self.assert_tombstoned(recs[0])

recs = self.ldap_get_records(B)
        self.assertEqual(len(recs), 0)

def _test_A_and_AAAA_records(self, A, B, a_days, b_days, aging):
        self.set_aging(aging)

name = 'aargh'
        now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))
        a_initial = now - 24 * a_days
        b_initial = now - 24 * b_days

self.dns_update_non_text(name, A)
        self.ldap_modify_timestamps(name, a_days * -24)

rec_a = self.get_unique_ip_record(name, A)
        rec_b = self.add_ip_record(name, B, dwTimeStamp=b_initial)

self.assert_timestamps_equal(rec_a, a_initial)
        self.assert_timestamps_equal(rec_b, b_initial)

# touch the A record.
        self.dns_update_non_text(name, A)

# check the A timestamp, depending on norefresh
        rec_a = self.get_unique_ip_record(name, A)
        if aging and a_days > 7:
            time_a = now
            self.assert_soon_after(rec_a, now)
        elif a_days > 7:
            # when we have NO aging and are in the refresh window, the
            # timestamp now reads as a_initial, but will become now
            # after we manipulate B for a bit.
            time_a = now
            self.assert_timestamps_equal(rec_a, a_initial)
        else:
            time_a = a_initial
            self.assert_timestamps_equal(rec_a, a_initial)

# B timestamp should be unchanged?
        rec_b = self.get_unique_ip_record(name, B)
        self.assert_timestamps_equal(rec_b, b_initial)

# touch the B record.
        self.dns_update_non_text(name, B)

# check the B timestamp
        rec_b = self.get_unique_ip_record(name, B)
        if not aging:
            self.windows_variation(
                self.assert_soon_after, rec_b, now,
                msg="windows updates non-aging, samba does not")
        else:
            self.assert_soon_after(rec_b, now)

# rewind B
        rec_b = self.add_ip_record(name, B, dwTimeStamp=b_initial)

# NOW rec A might have changed! with no aging, and out of refresh.
        rec_a = self.get_unique_ip_record(name, A)
        self.assert_timestamps_equal(rec_a, time_a)

self.dns_update_non_text(name, A)

rec_a = self.get_unique_ip_record(name, B)
        self.assert_timestamps_equal(rec_b, b_initial)

# now delete A
        _, wtype = guess_wtype(A)
        self.ldap_delete_record(name, A, wtype=wtype)

# re-add it
        self.dns_update_non_text(name, A)

rec_a = self.get_unique_ip_record(name, A)
        self.assert_soon_after(rec_a, now)

rec_b = self.get_unique_ip_record(name, B)
        self.assert_timestamps_equal(rec_b, b_initial)

def test_A_5_days_AAAA_5_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 5, 5, aging=True)

def test_A_5_days_AAAA_5_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 5, 5, aging=False)

def test_A_5_days_AAAA_10_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 5, 10, aging=True)

def test_A_5_days_AAAA_10_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 5, 10, aging=False)

def test_A_10_days_AAAA_5_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 10, 5, aging=True)

def test_A_10_days_AAAA_5_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 10, 5, aging=False)

def test_A_10_days_AAAA_9_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 10, 9, aging=True)

def test_A_9_days_AAAA_10_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 9, 10, aging=False)

def test_A_20_days_AAAA_2_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 20, 2, aging=True)

def test_A_6_days_AAAA_40_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv6_ADDR, 6, 40, aging=False)

def test_A_5_days_A_5_days_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv4_ADDR_2, 5, 5, aging=True)

def test_A_5_days_A_10_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv4_ADDR, IPv4_ADDR_2, 5, 10, aging=False)

def test_AAAA_5_days_AAAA_6_days_aging(self):
        self._test_A_and_AAAA_records(IPv6_ADDR, IPv6_ADDR_2, 5, 6, aging=True)

def test_AAAA_5_days_AAAA_6_days_no_aging(self):
        self._test_A_and_AAAA_records(IPv6_ADDR, IPv6_ADDR_2, 5, 6, aging=False)

def _test_multi_records_delete(self, aging):
        # Batch deleting a type doesn't update other types timestamps.
        self.set_aging(aging)

name = 'aargh'
        now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))

back_5_days = now - 5 * 24
        back_10_days = now - 10 * 24
        back_25_days = now - 25 * 24

ip4s = {
            '1.1.1.1': now,
            '2.2.2.2': back_5_days,
            '3.3.3.3': back_10_days,
        }
        ip6s = {
            '::1': now,
            '::2': back_5_days,
            '::3': back_25_days,
        }

txts = {
            '1': now,
            '2': back_5_days,
            '3': back_25_days,
        }

# For windows, if we don't DNS update something, it won't know
        # there's anything.
        self.dns_update_record(name, '3')

for k, v in ip4s.items():
            r = self.add_ip_record(name, k, wtype=dns.DNS_QTYPE_A, dwTimeStamp=v)

for k, v in ip6s.items():
            r = self.add_ip_record(name, k, wtype=dns.DNS_QTYPE_AAAA, dwTimeStamp=v)

for k, v in txts.items():
            r = self.ldap_update_record(name, k, dwTimeStamp=v)

self.dns_delete_type(name, dnsp.DNS_TYPE_A)

r = self.dns_query(name, dns.DNS_QTYPE_A)
        self.assertEqual(r.ancount, 0)

r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        self.assertEqual(r.ancount, 3)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, set(txts))

r = self.dns_query(name, dns.DNS_QTYPE_AAAA)
        self.assertEqual(r.ancount, 3)
        rset = set(ipv6_normalise(x.rdata) for x in r.answers)
        self.assertEqual(rset, set(ip6s))

recs = self.ldap_get_records(name)
        self.assertEqual(len(recs), 6)
        for r in recs:
            if r.wType == dns.DNS_QTYPE_AAAA:
                k = ipv6_normalise(r.data)
                expected = ip6s[k]
            elif r.wType == dns.DNS_QTYPE_TXT:
                k = r.data.str[0]
                expected = txts[k]
            else:
                self.fail(f"unexpected wType {r.wType}")

self.assert_timestamps_equal(r.dwTimeStamp, expected)

def test_multi_records_delete_aging(self):
        self._test_multi_records_delete(True)

def test_multi_records_delete_no_aging(self):
        self._test_multi_records_delete(False)

def _test_dns_delete_times(self, n_days, aging=True):
        # In these tests, Windows replaces the records with
        # tombstones, while Samba just removes them. Both are
        # reasonable approaches (there is no reanimation pathway for
        # tombstones), but this means self.ldap_get_records() gets
        # different numbers for each. So we use
        # self.ldap_get_non_tombstoned_record().
        name = 'test'
        A = ['A']
        B = ['B']
        C = ['C']
        D = ['D']
        self.set_aging(aging)
        now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))
        n_days_ago = max(now - n_days * 24, 0)

self.dns_update_record(name, A)
        self.ldap_update_record(name, A, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, B, dwTimeStamp=n_days_ago)
        self.ldap_update_record(name, C, dwTimeStamp=n_days_ago)
        self.dns_update_record(name, D)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, set('ABCD'))

atime = self.get_unique_txt_record(name, A).dwTimeStamp
        btime = self.get_unique_txt_record(name, B).dwTimeStamp
        ctime = self.get_unique_txt_record(name, C).dwTimeStamp
        dtime = self.get_unique_txt_record(name, D).dwTimeStamp
        recs = self.ldap_get_records(name)
        self.assertEqual(len(recs), 4)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, set('ABCD'))

self.dns_delete(name, D)
        self.assert_timestamps_equal(atime, self.get_unique_txt_record(name, A))
        self.assert_timestamps_equal(btime, self.get_unique_txt_record(name, B))
        self.assert_timestamps_equal(ctime, self.get_unique_txt_record(name, C))
        recs = self.ldap_get_non_tombstoned_records(name)
        self.assertEqual(len(recs), 3)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, set('ABC'))

self.rpc_delete_txt(name, C)
        self.assert_timestamps_equal(atime, self.get_unique_txt_record(name, A))
        self.assert_timestamps_equal(btime, self.get_unique_txt_record(name, B))
        recs = self.ldap_get_non_tombstoned_records(name)
        self.assertEqual(len(recs), 2)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, set('AB'))

self.dns_delete(name, A)
        self.assert_timestamps_equal(btime, self.get_unique_txt_record(name, B))
        recs = self.ldap_get_records(name)
        self.assertEqual(len(recs), 1)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(rset, {'B'})

self.dns_delete(name, B)
        recs = self.ldap_get_non_tombstoned_records(name)
        # Windows leaves the node with zero records. Samba ends up
        # with a tombstone.
        self.assertEqual(len(recs), 0)
        r = self.dns_query(name, dns.DNS_QTYPE_TXT)
        rset = set(x.rdata.txt.str[0] for x in r.answers)
        self.assertEqual(len(rset), 0)

def test_dns_delete_times_5_days_aging(self):
        self._test_dns_delete_times(5, True)

def test_dns_delete_times_11_days_aging(self):
        self._test_dns_delete_times(11, True)

def test_dns_delete_times_366_days_aging(self):
        self._test_dns_delete_times(366, True)

def test_dns_delete_times_static_aging(self):
        self._test_dns_delete_times(1e10, True)

def test_dns_delete_times_5_days_no_aging(self):
        self._test_dns_delete_times(5, False)

def test_dns_delete_times_11_days_no_aging(self):
        self._test_dns_delete_times(11, False)

def test_dns_delete_times_366_days_no_aging(self):
        self._test_dns_delete_times(366, False)

def test_dns_delete_times_static_no_aging(self):
        self._test_dns_delete_times(1e10, False)

def _test_dns_delete_simple(self, a_days, b_days, aging=True, touch=False):
        # Here we show that with aging enabled, the timestamp of
        # sibling records is *not* modified when a record is deleted.
        #
        # With aging disabled, it *is* modified, if the dns server has
        # seen it updated before ldap set the time (that is, probably
        # the dns server overwrites AD). This happens even if AD
        # thinks the record is static.
        name = 'test'
        A = ['A']
        B = ['B']
        self.set_aging(aging)
        now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))
        a_days_ago = max(now - a_days * 24, 0)
        b_days_ago = max(now - b_days * 24, 0)

if touch:
            self.dns_update_record(name, A)
            self.dns_update_record(name, B)

self.ldap_update_record(name, A, dwTimeStamp=a_days_ago)
        self.ldap_update_record(name, B, dwTimeStamp=b_days_ago)

atime = self.get_unique_txt_record(name, A).dwTimeStamp

self.dns_delete(name, B)
        if not aging and touch:
            # this resets the timestamp even if it is a static record.
            self.assert_soon_after(self.get_unique_txt_record(name, A), now)
        else:
            self.assert_timestamps_equal(self.get_unique_txt_record(name, A), atime)

def test_dns_delete_simple_2_3_days_aging(self):
        self._test_dns_delete_simple(2, 3, True)

def test_dns_delete_simple_2_3_days_no_aging(self):
        self._test_dns_delete_simple(2, 3, False)

def test_dns_delete_simple_2_13_days_aging(self):
        self._test_dns_delete_simple(2, 13, True)

def test_dns_delete_simple_2_13_days_no_aging(self):
        self._test_dns_delete_simple(2, 13, False)

def test_dns_delete_simple_12_13_days_aging(self):
        self._test_dns_delete_simple(12, 13, True)

def test_dns_delete_simple_12_13_days_no_aging(self):
        self._test_dns_delete_simple(12, 13, False)

def test_dns_delete_simple_112_113_days_aging(self):
        self._test_dns_delete_simple(112, 113, True)

def test_dns_delete_simple_112_113_days_no_aging(self):
        self._test_dns_delete_simple(112, 113, False)

def test_dns_delete_simple_0_113_days_aging(self):
        # 1e9 hours ago evaluates to 0, i.e static
        self._test_dns_delete_simple(1e9, 113, True)

def test_dns_delete_simple_0_113_days_no_aging(self):
        self._test_dns_delete_simple(1e9, 113, False)

def test_dns_delete_simple_0_0_days_aging(self):
        self._test_dns_delete_simple(1e9, 1e9, True)

def test_dns_delete_simple_0_0_days_no_aging(self):
        self._test_dns_delete_simple(1e9, 1e9, False)

def test_dns_delete_simple_10_0_days_aging(self):
        self._test_dns_delete_simple(10, 1e9, True)

def test_dns_delete_simple_10_0_days_no_aging(self):
        self._test_dns_delete_simple(10, 1e9, False)

def test_dns_delete_simple_2_3_days_aging_touch(self):
        self._test_dns_delete_simple(2, 3, True, True)

def test_dns_delete_simple_2_3_days_no_aging_touch(self):
        self._test_dns_delete_simple(2, 3, False, True)

def test_dns_delete_simple_2_13_days_aging_touch(self):
        self._test_dns_delete_simple(2, 13, True, True)

def test_dns_delete_simple_2_13_days_no_aging_touch(self):
        self._test_dns_delete_simple(2, 13, False, True)

def test_dns_delete_simple_12_13_days_aging_touch(self):
        self._test_dns_delete_simple(12, 13, True, True)

def test_dns_delete_simple_12_13_days_no_aging_touch(self):
        self._test_dns_delete_simple(12, 13, False, True)

def test_dns_delete_simple_112_113_days_aging_touch(self):
        self._test_dns_delete_simple(112, 113, True, True)

def test_dns_delete_simple_112_113_days_no_aging_touch(self):
        self._test_dns_delete_simple(112, 113, False, True)

def test_dns_delete_simple_0_113_days_aging_touch(self):
        # 1e9 hours ago evaluates to 0, i.e static
        self._test_dns_delete_simple(1e9, 113, True, True)

def test_dns_delete_simple_0_113_days_no_aging_touch(self):
        self._test_dns_delete_simple(1e9, 113, False, True)

def test_dns_delete_simple_0_0_days_aging_touch(self):
        self._test_dns_delete_simple(1e9, 1e9, True, True)

def test_dns_delete_simple_0_0_days_no_aging_touch(self):
        self._test_dns_delete_simple(1e9, 1e9, False, True)

def test_dns_delete_simple_10_0_days_aging_touch(self):
        self._test_dns_delete_simple(10, 1e9, True, True)

def test_dns_delete_simple_10_0_days_no_aging_touch(self):
        self._test_dns_delete_simple(10, 1e9, False, True)

def windows_variation(self, fn, *args, msg=None, **kwargs):
        try:
            fn(*args, **kwargs)
        except AssertionError as e:
            print("Expected success on Windows only, failed as expected:\n" +
                  c_GREEN(e))
            return
        print(c_RED("known Windows failure"))
        if msg is not None:
            print(c_DARK_YELLOW(msg))
        print("Expected success on Windows:\n" +
              c_GREEN(f"{fn.__name__} {args} {kwargs}"))

def _test_dns_add_sibling(self, a_days, refresh, aging=True, touch=False):
        # Here we show that with aging enabled, the timestamp of
        # sibling records *is* modified when a record is added.
        #
        # With aging disabled, it *is* modified, if the dns server has
        # seen it updated before ldap set the time (that is, probably
        # the dns server overwrites AD). This happens even if AD
        # thinks the record is static.
        name = 'test'
        A = ['A']
        B = ['B']
        self.set_zone_int_params(RefreshInterval=int(refresh),
                                 NoRefreshInterval=7,
                                 Aging=int(aging))

now = dsdb_dns.unix_to_dns_timestamp(int(time.time()))
        a_days_ago = max(now - a_days * 24, 0)

if touch:
            self.dns_update_record(name, A)

self.ldap_update_record(name, A, dwTimeStamp=a_days_ago)

atime = self.get_unique_txt_record(name, A).dwTimeStamp

self.dns_update_record(name, B)
        a_rec = self.get_unique_txt_record(name, A)
        if not aging and touch:
            # On Windows, this resets the timestamp even if it is a
            # static record, though in that case it may be a
            # transitory effect of the DNS cache. We will insist on
            # the Samba behaviour of not changing (that is
            # un-static-ing) a zero timestamp, because that is the
            # sensible thing.
            if a_days_ago == 0:
                self.windows_variation(
                    self.assert_soon_after, a_rec, now,
                    msg="Windows resets static siblings (cache effect?)")
                self.assert_timestamps_equal(a_rec, 0)
            else:
                self.assert_soon_after(a_rec, now)
        else:
            self.assert_timestamps_equal(a_rec, atime)

b_rec = self.get_unique_txt_record(name, B)
        self.assert_soon_after(b_rec, now)

def test_dns_add_sibling_2_7_days_aging(self):
        self._test_dns_add_sibling(2, 7, True)

def test_dns_add_sibling_2_7_days_no_aging(self):
        self._test_dns_add_sibling(2, 7, False)

def test_dns_add_sibling_12_7_days_aging(self):
        self._test_dns_add_sibling(12, 7, True)

def test_dns_add_sibling_12_7_days_no_aging(self):
        self._test_dns_add_sibling(12, 7, False)

def test_dns_add_sibling_12_3_days_aging(self):
        self._test_dns_add_sibling(12, 3, True)

def test_dns_add_sibling_12_3_days_no_aging(self):
        self._test_dns_add_sibling(12, 3, False)

def test_dns_add_sibling_112_7_days_aging(self):
        self._test_dns_add_sibling(112, 7, True)

def test_dns_add_sibling_112_7_days_no_aging(self):
        self._test_dns_add_sibling(112, 7, False)

def test_dns_add_sibling_12_113_days_aging(self):
        self._test_dns_add_sibling(12, 113, True)

def test_dns_add_sibling_12_113_days_no_aging(self):
        self._test_dns_add_sibling(12, 113, False)

def test_dns_add_sibling_0_7_days_aging(self):
        # 1e9 days ago evaluates to 0, i.e static
        self._test_dns_add_sibling(1e9, 7, True)

def test_dns_add_sibling_0_7_days_no_aging(self):
        self._test_dns_add_sibling(1e9, 7, False)

def test_dns_add_sibling_0_0_days_aging(self):
        self._test_dns_add_sibling(1e9, 0, True)

def test_dns_add_sibling_0_0_days_no_aging(self):
        self._test_dns_add_sibling(1e9, 0, False)

def test_dns_add_sibling_10_0_days_aging(self):
        self._test_dns_add_sibling(10, 0, True)

def test_dns_add_sibling_10_0_days_no_aging(self):
        self._test_dns_add_sibling(10, 0, False)

def test_dns_add_sibling_2_7_days_aging_touch(self):
        self._test_dns_add_sibling(2, 7, True, True)

def test_dns_add_sibling_2_7_days_no_aging_touch(self):
        self._test_dns_add_sibling(2, 7, False, True)

def test_dns_add_sibling_12_7_days_aging_touch(self):
        self._test_dns_add_sibling(12, 7, True, True)

def test_dns_add_sibling_12_7_days_no_aging_touch(self):
        self._test_dns_add_sibling(12, 7, False, True)

def test_dns_add_sibling_12_3_days_aging_touch(self):
        self._test_dns_add_sibling(12, 3, True, True)

def test_dns_add_sibling_12_3_days_no_aging_touch(self):
        self._test_dns_add_sibling(12, 3, False, True)

def test_dns_add_sibling_112_7_days_aging_touch(self):
        self._test_dns_add_sibling(112, 7, True, True)

def test_dns_add_sibling_112_7_days_no_aging_touch(self):
        self._test_dns_add_sibling(112, 7, False, True)

def test_dns_add_sibling_12_113_days_aging_touch(self):
        self._test_dns_add_sibling(12, 113, True, True)

def test_dns_add_sibling_12_113_days_no_aging_touch(self):
        self._test_dns_add_sibling(12, 113, False, True)

def test_dns_add_sibling_0_7_days_aging_touch(self):
        self._test_dns_add_sibling(1e9, 7, True, True)

def test_dns_add_sibling_0_7_days_no_aging_touch(self):
        self._test_dns_add_sibling(1e9, 7, False, True)

def test_dns_add_sibling_0_0_days_aging_touch(self):
        self._test_dns_add_sibling(1e9, 0, True, True)

def test_dns_add_sibling_0_0_days_no_aging_touch(self):
        self._test_dns_add_sibling(1e9, 0, False, True)

def test_dns_add_sibling_10_0_days_aging_touch(self):
        self._test_dns_add_sibling(10, 0, True, True)

def test_dns_add_sibling_10_0_days_no_aging_touch(self):
        self._test_dns_add_sibling(10, 0, False, True)

TestProgram(module=__name__, opts=subunitopts)

test | Fairnetzt-Hochrhein

Home Events test

test

Dies ist ein Test

Datum

25 Nov. 2022

Vorbei!

Uhrzeit

13:00 - 18:00

JezK

Edit File: dns_aging.py

test

Datum

Uhrzeit

Teile diese Veranstaltung