#!/usr/share/ucs-test/runner python
## desc: Test S4 SRV record maintainance
## roles: [domaincontroller_slave]
## tags: [apptest,ucsschool]
## exposure: dangerous
## packages:
##    - ucs-school-slave | ucs-school-nonedu-slave
##    - univention-samba4
##    - ucs-school-s4-branch-site

import os
import sys
sys.dont_write_bytecode = True
import subprocess
import time
from ldap.filter import escape_filter_chars
import univention.testing.utils as testing_utils
import univention.testing.ucr as testing_ucr
import univention.testing.udm as testing_udm
from univention.admin.uldap import explodeDn
from ucsschool.lib.schoolldap import set_credentials, LDAP_Connection, MACHINE_READ
from ucsschool.lib.models import School
import univention.admin.uexceptions as udm_errors
import univention.testing.strings as uts
import univention.config_registry

try:
	from univention.testing.ucs_samba import wait_for_drs_replication
except ImportError:
	print "WARNING: Using dummy implementation for wait_for_drs_replication."

	def wait_for_drs_replication(ldap_filter, attrs=None):
		print "wait_for_drs_replication: sleeping 15 seconds"
		time.sleep(15)

# Initialize globals
ucr = testing_ucr.UCSTestConfigRegistry()
ucr.load()
TESTS_DOMAINADMIN_ACCOUNT = ucr.get('tests/domainadmin/account')
if TESTS_DOMAINADMIN_ACCOUNT:
	TESTS_DOMAINADMIN = explodeDn(TESTS_DOMAINADMIN_ACCOUNT, 1)[0]
TESTS_DOMAINADMIN_PWD = ucr.get('tests/domainadmin/pwd')
set_credentials(TESTS_DOMAINADMIN_ACCOUNT, TESTS_DOMAINADMIN_PWD)

listener_name = "ucsschool-s4-branch-site"

LISTENER_BASEDIR = "/usr/lib/univention-directory-listener"
HOOKS_BASEDIR = os.path.join(LISTENER_BASEDIR, "hooks")
LISTENER_HOOKS_BASEDIR = os.path.join(HOOKS_BASEDIR, "%s.d" % (listener_name,))

module_template = '''
__package__='' 	# workaround for PEP 366
import univention.debug as ud
import univention.config_registry
import listener

ud.debug(ud.LISTENER, ud.ERROR, "%(modulename)s load")

def handler(dn, new, old, command):
	ud.debug(ud.LISTENER, ud.ERROR, "%(modulename)s handler %(secret)s called")
	if "%(modulename)s" == "01_hook":
		ud.debug(ud.LISTENER, ud.ERROR, " ".join((dn, new, old, unknown)))	## SHOULD FAIL
	else:
		ud.debug(ud.LISTENER, ud.ERROR, " ".join((dn, str(new), str(old), command)))

def postrun():
	ud.debug(ud.LISTENER, ud.ERROR, "%(modulename)s postrun %(secret)s called")
'''


class Test():

	@classmethod
	def setup_hook_file(cls, filename, secret_str):
		modulename = filename[:-3].replace('-', '_')
		filepath = os.path.join(LISTENER_HOOKS_BASEDIR, filename)
		with open(filepath, 'w') as f:
			f.write(module_template % {'modulename': modulename, 'secret': secret_str})

	def __enter__(self):
		self.secret_str = uts.random_name()
		Test.setup_hook_file("02_hook.py", self.secret_str)
		Test.setup_hook_file("01_hook.py", "_dummy_")
		return self

	def __exit__(self, type, value, traceback):
		os.unlink(os.path.join(LISTENER_HOOKS_BASEDIR, "01_hook.py"))
		os.unlink(os.path.join(LISTENER_HOOKS_BASEDIR, "02_hook.py"))
		cmd = ["/etc/init.d/univention-directory-listener", "restart"]
		p1 = subprocess.Popen(cmd)
		p1.wait()

	@LDAP_Connection(MACHINE_READ)
	def run(self, ldap_machine_read=None, ldap_position=None):
		status = 100
		positive_test_fqdn_list = []
		negative_test_fqdn_list = []

		try:
			res = ldap_machine_read.search(base=ucr['ldap/hostdn'], scope='base', attr=('univentionService',))
		except udm_errors.ldapError, e:
			testing_utils.fail(log_message="Error accessing LDAP: %s" % (e,))

		(record_dn, obj) = res[0]
		services = obj['univentionService']
		_ucsschool_services = set(('UCS@school Education', 'UCS@school Administration'))
		for service_id in _ucsschool_services:
			if service_id in services:
				_local_ucsschool_service = service_id
				_ucsschool_services.remove(service_id)
				_not_local_ucsschool_service = _ucsschool_services.pop()
				break

		try:
			ldap_filter = "(&(objectClass=univentionDomainController)(!(univentionService=%s)))" \
				% _local_ucsschool_service
			attrs = ['cn', 'associatedDomain']
			res = ldap_machine_read.search(base=ldap_position.getDn(), filter=ldap_filter, attr=attrs)
		except udm_errors.ldapError, e:
			testing_utils.fail(log_message="Error accessing LDAP: %s" % (e,))

		for (record_dn, obj) in res:
			if 'associatedDomain' in obj:
				domainname = obj['associatedDomain'][0]
			else:
				domainname = ucr['domainname']
			negative_test_fqdn_list.append(".".join((obj['cn'][0], domainname)))

		with testing_udm.UCSTestUDM() as udm:
			for school in School.get_all(ldap_machine_read):
				positive_test_hostname = uts.random_name()
				udm.create_object(
					"computers/domaincontroller_slave",
					name=positive_test_hostname,
					position="cn=dc,cn=server,cn=computers,%s" % (school.dn,),
					domain=ucr.get('domainname'),
					service=("S4 SlavePDC", _local_ucsschool_service),
					groups=("cn=DC-Edukativnetz,cn=ucsschool,cn=groups,%(ldap/base)s" % ucr)
				)

				positive_test_fqdn = ".".join((positive_test_hostname, ucr.get('domainname')))
				positive_test_fqdn_list.append(positive_test_fqdn)

				negative_test_hostname = uts.random_name()
				udm.create_object(
					"computers/domaincontroller_slave",
					name=negative_test_hostname,
					position="cn=dc,cn=server,cn=computers,%s" % (school.dn,),
					domain=ucr.get('domainname'),
					service=("S4 SlavePDC", _not_local_ucsschool_service),
					groups=("cn=DC-Edukativnetz,cn=ucsschool,cn=groups,%(ldap/base)s" % ucr)
				)

				negative_test_fqdn = ".".join((negative_test_hostname, ucr.get('domainname')))

				# wait for postrun and S4 Connector, check by creating a dummy user and waiting for that object to appear
				dummy_username = uts.random_username()
				dummy_password = uts.random_string()
				udm.create_user(username=dummy_username, password=dummy_password)
				testing_utils.wait_for_replication_and_postrun()
				wait_for_drs_replication("(sAMAccountName=%s)" % (escape_filter_chars(dummy_username),), attrs="sAMAccountName")

				# The replication in a multi shool environment need more time
				time.sleep(15)

				# verify that the positive test DC is present in the UCR variables
				ucr2 = univention.config_registry.ConfigRegistry()
				ucr2.load()
				test_srv_record = '_kerberos._tcp'
				ucr_var = 'connector/s4/mapping/dns/srv_record/%s.%s/location' % (test_srv_record, ucr.get('domainname'))
				test_value = ucr2.get(ucr_var, '')
				if test_value.find(positive_test_fqdn) == -1:
					testing_utils.fail(log_message="%s not found in %s" % (positive_test_fqdn, ucr_var))

				# verify that the negative test DC is NOT present in the UCR variables
				if test_value.find(negative_test_fqdn) != -1:
					testing_utils.fail(log_message="%s found in %s" % (negative_test_fqdn, ucr_var))

				# verify that the positive_test DC is present in DNS/Samba4
				time.sleep(3)
				p1 = subprocess.Popen(['host', '-t', 'srv', test_srv_record], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
				(stdout, stderr) = p1.communicate()

				if stdout.find(positive_test_fqdn) == -1:
					testing_utils.fail(log_message="%s not found in DNS SRV record %s" % (positive_test_fqdn, test_srv_record))

				# verify that the negative_test DC is NOT present in DNS/Samba4
				if stdout.find(negative_test_fqdn) != -1:
					testing_utils.fail(log_message="%s found in DNS SRV record %s" % (negative_test_fqdn, test_srv_record))

			# restart listener to load the test hooks before the test DCs get removed
			cmd = ["/etc/init.d/univention-directory-listener", "restart"]
			p1 = subprocess.Popen(cmd)
			p1.wait()

		# Ok, now the UCSTestUDM() context got destroyed via UCSTestUDM().__exit__()
		# wait for postrun and S4 Connector, check by creating a dummy user and waiting for that object to appear
		with testing_udm.UCSTestUDM() as udm:
			dummy_username = uts.random_username()
			dummy_password = uts.random_string()
			udm.create_user(username=dummy_username, password=dummy_password)
			testing_utils.wait_for_replication_and_postrun()
			wait_for_drs_replication("(sAMAccountName=%s)" % (escape_filter_chars(dummy_username),), attrs="sAMAccountName")
		# The replication in a multi shool environment need more time
		time.sleep(15)

		# verify that the postitive test DCs are removed from DNS/Samba4
		p1 = subprocess.Popen(['host', '-t', 'srv', '_kerberos._tcp'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
		(stdout, stderr) = p1.communicate()
		for postitive_test_fqdn in positive_test_fqdn_list:
			if stdout.find(postitive_test_fqdn) != -1:
				testing_utils.fail(log_message="%s still found in DNS SRV record %s" % (postitive_test_fqdn, test_srv_record))

		# verify that the "negative" DCs of the other Role (Education/Administration) are not in DNS/Samba4
		for negative_test_fqdn in negative_test_fqdn_list:
			if stdout.find(negative_test_fqdn) != -1:
				testing_utils.fail(log_message="%s present in DNS SRV record %s" % (negative_test_fqdn, test_srv_record))

		# verify that the local DC is still in DNS/Samba4
		local_fqdn = ".".join((ucr.get("hostname"), ucr.get("domainname")))
		if stdout.find(local_fqdn) == -1:
			testing_utils.fail(log_message="%s not present in DNS SRV record %s" % (local_fqdn, test_srv_record))

		# verify that the positive test DCs are removed from the UCR variable
		ucr2 = univention.config_registry.ConfigRegistry()
		ucr2.load()
		test_srv_record = '_kerberos._tcp'
		ucr_var = 'connector/s4/mapping/dns/srv_record/%s.%s/location' % (test_srv_record, ucr.get('domainname'))
		test_value = ucr2.get(ucr_var, '')
		for positive_test_fqdn in positive_test_fqdn_list:
			if test_value.find(positive_test_fqdn) != -1:
				testing_utils.fail(log_message="%s still found in UCR variable %s" % (positive_test_fqdn, ucr_var))

		# verify that the "negative" DCs are not in the UCR variable:
		for negative_test_fqdn in negative_test_fqdn_list:
			if test_value.find(negative_test_fqdn) != -1:
				testing_utils.fail(log_message="%s found in UCR variable %s" % (negative_test_fqdn, ucr_var))

		# verify that the local DC is still in the UCR variable
		if test_value.find(local_fqdn) == -1:
			testing_utils.fail(log_message="%s not present in UCR variable %s" % (local_fqdn, ucr_var))

		# check that the listener hooks of 02_hook have been run:
		for attr in ("handler", "postrun"):
			search_string = "02_hook %s %s called" % (attr, self.secret_str)
			cmd = ["grep", "-q", search_string, "/var/log/univention/listener.log"]
			p1 = subprocess.Popen(cmd, shell=False, close_fds=True)
			p1.wait()
			if p1.returncode:
				testing_utils.fail(log_message="hook not called for '%s()'" % (attr,))

		return status


if __name__ == '__main__':

	with Test() as test:
		rc = test.run()

	sys.exit(rc)

# vim: set filetype=py
