python/samba/tests/blackbox/netads_dns.py

   1 # Blackbox tests for the "net ads dns async" commands
   2 #
   3 # Copyright (C) Samuel Cabrero <scabrero@samba.org> 2022
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 import os
  19 import dns.resolver
  20 import re
  21
  22 from samba.tests import BlackboxTestCase
  23
  24 SERVER = os.environ["DC_SERVER"]
  25 REALM = os.environ["REALM"]
  26 COMMAND = "bin/net ads"
  27
  28 class NetAdsDnsTests(BlackboxTestCase):
  29
  30     def setUp(self):
  31         super().setUp()
  32         nameserver = os.environ["DC_SERVER_IP"]
  33         # filename=None will disable reading /etc/resolv.conf. The file might
  34         # not exist e.g. on build or CI systems.
  35         self.resolver = dns.resolver.Resolver(filename=None)
  36         self.resolver.nameservers = [nameserver]
  37
  38     def parse_output(self, output):
  39         v4 = []
  40         v6 = []
  41         for line in output.split("\n"):
  42             m = re.search(r'^.*IPv4addr = (.*)$', line)
  43             if m:
  44                 v4.append(m.group(1))
  45             m = re.search(r'^.*IPv6addr = (.*)$', line)
  46             if m:
  47                 v6.append(m.group(1))
  48         return (v4, v6)
  49
  50     def test_async_dns(self):
  51         host = "%s.%s" % (SERVER, REALM)
  52
  53         sync_v4 = []
  54         answers = self.resolver.query(host, 'A')
  55         for rdata in answers:
  56             sync_v4.append(rdata.address)
  57         self.assertGreaterEqual(len(sync_v4), 1)
  58
  59         sync_v6 = []
  60         answers = self.resolver.query(host, 'AAAA')
  61         for rdata in answers:
  62             sync_v6.append(rdata.address)
  63         self.assertGreaterEqual(len(sync_v6), 1)
  64
  65         async_v4 = []
  66         async_v6 = []
  67         argv = "%s dns async %s.%s " % (COMMAND, SERVER, REALM)
  68         try:
  69             out = self.check_output(argv)
  70             (async_v4, async_v6) = self.parse_output(out.decode('utf-8'))
  71         except samba.tests.BlackboxProcessError as e:
  72             self.fail("Error calling [%s]: %s" % (argv, e))
  73
  74         self.assertGreaterEqual(len(async_v4), 1)
  75         self.assertGreaterEqual(len(async_v6), 1)
  76
  77         sync_v4.sort()
  78         async_v4.sort()
  79         self.assertStringsEqual(' '.join(sync_v4), ' '.join(async_v4))
  80
  81         sync_v6.sort()
  82         async_v6.sort()
  83         self.assertStringsEqual(' '.join(sync_v6), ' '.join(async_v6))