From 0c5dc26ddc0602c8d2eeb7599efd8a9e0f252ffe Mon Sep 17 00:00:00 2001
From: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Date: Mon, 14 Jun 2021 23:41:14 +1200
Subject: [PATCH] pytest: dns_aging: add Samba-specific scavenging test

We can't make scavenging happen on demand on Windows, so we just
concentrate on Samba here.

Signed-off-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
---
 python/samba/tests/dns_aging.py | 85 +++++++++++++++++++++++++++++++++
 selftest/knownfail.d/dns-aging  |  1 +
 2 files changed, 86 insertions(+)

diff --git a/python/samba/tests/dns_aging.py b/python/samba/tests/dns_aging.py
index 5a6c7346d5f..769ed724d2b 100644
--- a/python/samba/tests/dns_aging.py
+++ b/python/samba/tests/dns_aging.py
@@ -2085,6 +2085,91 @@ class TestDNSAging(DNSTest):
 
         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
+        btime = 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 = dns_query(self, 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)
 
diff --git a/selftest/knownfail.d/dns-aging b/selftest/knownfail.d/dns-aging
index 3f890f0125d..daf13e731ef 100644
--- a/selftest/knownfail.d/dns-aging
+++ b/selftest/knownfail.d/dns-aging
@@ -45,3 +45,4 @@ samba.tests.dns_aging.+test_A_5_days_A_10_days_no_aging
 samba.tests.dns_aging.+test_A_5_days_A_5_days_aging
 samba.tests.dns_aging.+test_A_6_days_AAAA_40_days_no_aging
 samba.tests.dns_aging.+test_A_9_days_AAAA_10_days_no_aging
+samba.tests.dns_aging.+test_samba_scavenging
-- 
2.47.3