Attachment #8953 for bug #42465




#!/usr/share/ucs-test/runner python
## -*- coding: utf-8 -*-
## desc: test UsernameHandler with --dry-run (Bug 42465)
## tags: [apptest,ucsschool,skip_in_multiserver,ucsschool_import]
## roles: [domaincontroller_master]
## exposure: dangerous
## packages:
##   - ucs-school-import
## bugs: [42465]

import copy
import string

from univention.testing.ucs_samba import wait_for_drs_replication
from univention.admin.uldap import getAdminConnection
from univention.admin.uexceptions import noObject, ldapError
from ucsschool.importer.utils.username_handler import UsernameHandler
from ucsschool.importer.exceptions import FormatError
from essential.importusers import Person
from essential.importusers_cli_v2 import CLI_Import_v2_Tester


				self.log.error("DN %r -> %s", dn, exc)
		super(Test, self).cleanup()

	def test(self):
		"""
		Bug #42313: remove illegal characters from username
		* "Username must only contain numbers, letters and dots, and may not be 'admin'!"

				name = name.strip(".")
			return name.translate(None, bad_chars)

		random_puncts = list(string.punctuation)
		random.shuffle(random_puncts)
		lastname = uts.random_username(20)
		self.unique_basenames_to_remove.append(lastname)
		self.log.info("Testing with dry_run=False and lastname=%r which will be shortened to %r...", lastname, lastname[:12])
			lastnames.append(".{}.{}.".format(uts.random_username(4), uts.random_username(4)))
			lastnames.append("{}[{}]{}".format(uts.random_username(3), uts.random_username(3), uts.random_username(3)))
			lastnames.append(uts.random_username(40))
			self.unique_basenames_to_remove.extend(lastnames)
			self.log.info('*** Importing new users with role %r and the following lastnames:\n%r', role, lastnames)

		role = 'student'
		record_uid = uts.random_name()
		person = Person(self.ou_A.name, role)
		source_uid = 'sourceUID-%s' % (uts.random_string(),)
		person.update(record_uid=record_uid, source_uid=source_uid, lastname=lastname)
			config.update_entry('csv:mapping:recordUID', 'record_uid')
			config.update_entry('scheme:username:default', "<lastname>[ALWAYSCOUNTER]")

		config = copy.deepcopy(self.default_config)
		config.update_entry('sourceUID', source_uid)
		config.update_entry('scheme:recordUID', '<record_uid>')
		config.update_entry('user_role', role)
		config.update_entry('csv:mapping:recordUID', 'record_uid')
		config.update_entry('scheme:username:default', "<lastname>[ALWAYSCOUNTER]")
		fn_csv = self.create_csv_file(person_list=[person], mapping=config['csv']['mapping'])
		config.update_entry('input:filename', fn_csv)
		fn_config = self.create_config_json(config=config)
			config.update_entry('input:filename', fn_csv)
			fn_config = self.create_config_json(config=config)

		self.save_ldap_status()
		self.run_import(['-c', fn_config, '-i', fn_csv])
		utils.wait_for_replication()
		self.check_new_and_removed_users(1, 0)
		person.update(username=remove_bad_chars("{}1".format(lastname[:12])))
		utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)
		wait_for_drs_replication('cn={}'.format(escape_filter_chars(person.username)))  # wait for creation before deletion
			# check usernames
			for person, lastname in names.items():
				person.update(username=remove_bad_chars("{}1".format(lastname[:12])))
				utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)
				wait_for_drs_replication('cn={}'.format(escape_filter_chars(person.username)))  # wait for creation before deletion

		# delete user
		self.log.info('*** Deleting user with lastname=%r', lastname)
		fn_csv = self.create_csv_file(person_list=[], mapping=config['csv']['mapping'])
		config.update_entry('input:filename', fn_csv)
		fn_config = self.create_config_json(config=config)
		self.save_ldap_status()
		self.run_import(['-c', fn_config, '-i', fn_csv])
		utils.wait_for_replication()

		# recreate user with dry_run
		self.log.info('*** Importing user with the same lastname with --dry_run')
		# Prevent 'The email address is already taken by another user. Please change the email address.'
		# because of slow s4 replication. Username is the same, so it doesn't change what's tested.
		person.mail = '{}{}'.format(uts.random_username(4), person.mail)
		fn_csv = self.create_csv_file(person_list=[person], mapping=config['csv']['mapping'])
		config.update_entry('input:filename', fn_csv)
		fn_config = self.create_config_json(config=config)
		self.save_ldap_status()
		self.run_import(['-c', fn_config, '-i', fn_csv, '--dry-run'])
		utils.wait_for_replication()
		utils.verify_ldap_object(person.dn, should_exist=False)
		person.update(username=remove_bad_chars("{}2".format(lastname[:12])))
		utils.verify_ldap_object(person.dn, should_exist=False)
			self.run_import(['-c', fn_config, '-i', fn_csv])
			utils.wait_for_replication()
			for person, lastname in names.items():
				person.update(username=remove_bad_chars("2{}".format(lastname[:12])))
				utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)

		lo, po = getAdminConnection()
		utils.verify_ldap_object(
			"cn={},cn=unique-usernames,cn=ucsschool,cn=univention,{}".format(escape_dn_chars(lastname[:12]), lo.base),
			expected_attr={'ucsschoolUsernameNextNumber': ['2']},
			strict=False,
			should_exist=True)
			"": None,
			"..[ALWAYSCOUNTER]..": None,
			"[ALWAYSCOUNTER]": None,
			"[FOObar]": "FOObar",
			"{}.[COUNTER2]".format(name12): name12,
			".": None,
			'M'*14 + '.' + 'A': 'M' * 14,
		}
		for input_name, expected in usernames.items():
			try:
				out = unh.format_username(input_name)
				self.unique_basenames_to_remove.append(expected)
				if out != expected:
					self.fail("UsernameHandler.format_username(%r) returned %r, expected %r." % (input_name, out, expected))
			except FormatError:
				if expected is not None:
					self.fail("UsernameHandler.format_username(%r) raise a FormatError, expected it to return %r." % (input_name, expected))
				continue
		self.log.info('*** Starting unit test for UsernameHandler.format_username() (2/5)')
		for i in range(1000):
			name = uts.random_username(15)
			self.unique_basenames_to_remove.append(name)
			out = unh.format_username(name)
			if out != name:
				self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
		self.log.info('*** Starting unit test for UsernameHandler.format_username() (3/5)')
		for i in range(1000):
			name = uts.random_name(20)
			self.unique_basenames_to_remove.append(name)
			out = unh.format_username(name)
			if out.startswith(".") or out.endswith(".") or len(out) > 15:
				self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
		self.log.info('*** Starting unit test for UsernameHandler.format_username() (4/5)')
		for i in range(1000):
			name = uts.random_name_special_characters(20)
			name = name.translate(None, "[]")  # those are reserved for counter vars
			self.unique_basenames_to_remove.append(name)
			out = unh.format_username(name)
			if out.startswith(".") or out.endswith(".") or len(out) > 15:
				self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
		self.log.info('*** Starting unit test for UsernameHandler.format_username() (5/5)')
		usernames = [
			('Max[ALWAYSCOUNTER].Mustermann', 'Max1.Mustermann'),
			('Max[ALWAYSCOUNTER].Mustermann', 'Max2.Mustermann'),
			('Max[ALWAYSCOUNTER].Mustermann', 'Max3.Mustermann'),
			('Max[ALWAYSCOUNTER].Mustermann', 'Max4.Mustermann'),
			('Maria[ALWAYSCOUNTER].Musterfrau', 'Maria1.Musterfrau'),
			('Moritz[COUNTER2]', 'Moritz'),
			('Moritz[COUNTER2]', 'Moritz2')
		]
		self.unique_basenames_to_remove.extend(["Max.Mustermann", "Maria.Musterfrau", "Moritz"])
		unh = UsernameHandler(20)
		for input_name, expected in usernames:
			out = unh.format_username(input_name)
			if out != expected:
				self.fail("UsernameHandler.format_username(%r) returned %r, expected %r." % (input_name, out, expected))

		self.log.info('*** OK: Import with --dry_run didn\'t raise the counter.')


def main():
	tester = Test()
	try:

Lines 1-3 Link Here

(-)ucs-test-ucsschool/debian/changelog (+6 lines)
		1	ucs-test-ucsschool (3.0.17-63) unstable; urgency=low
		2
		3	* Bug #42465: add test for dry-run and username counter
		4
		5	-- Daniel Troeder <troeder@univention.de> Thu, 22 Jun 2017 10:42:55 +0200
		6
1	ucs-test-ucsschool (3.0.17-62) unstable; urgency=low	7	ucs-test-ucsschool (3.0.17-62) unstable; urgency=low
2		8
3	* Bug #44588: add support for multiple schools to test User	9	* Bug #44588: add support for multiple schools to test User

Return to bug 42465

Lines 1-12 Link Here

(-)ucs-test-ucsschool/90_ucsschool/215_import-users_username_dry-run (-134 / +59 lines)
1	#!/usr/share/ucs-test/runner python	1	#!/usr/share/ucs-test/runner python
2	## -- coding: utf-8 --	2	## -- coding: utf-8 --
3	## desc: remove illegal characters from username (Bug 42313)	3	## desc: test UsernameHandler with --dry-run (Bug 42465)
4	## tags: [apptest,ucsschool,skip_in_multiserver,ucsschool_import]	4	## tags: [apptest,ucsschool,skip_in_multiserver,ucsschool_import]
5	## roles: [domaincontroller_master]	5	## roles: [domaincontroller_master]
6	## exposure: dangerous	6	## exposure: dangerous
7	## packages:	7	## packages:
8	## - ucs-school-import	8	## - ucs-school-import
9	## bugs: [42313]	9	## bugs: [42465]
10		10
11	import copy	11	import copy
12	import string	12	import string
Lines 18-25 Link Here
18	from univention.testing.ucs_samba import wait_for_drs_replication	18	from univention.testing.ucs_samba import wait_for_drs_replication
19	from univention.admin.uldap import getAdminConnection	19	from univention.admin.uldap import getAdminConnection
20	from univention.admin.uexceptions import noObject, ldapError	20	from univention.admin.uexceptions import noObject, ldapError
21	from ucsschool.importer.utils.username_handler import UsernameHandler
22	from ucsschool.importer.exceptions import FormatError
23	from essential.importusers import Person	21	from essential.importusers import Person
24	from essential.importusers_cli_v2 import CLI_Import_v2_Tester	22	from essential.importusers_cli_v2 import CLI_Import_v2_Tester
25		23
Lines 46-52 Link Here
46	self.log.error("DN %r -> %s", dn, exc)	44	self.log.error("DN %r -> %s", dn, exc)
47	super(Test, self).cleanup()	45	super(Test, self).cleanup()
48		46
49	def test(self): # formally test_create_with_illegal_chars_in_username()	47	def test(self):
50	"""	48	"""
51	Bug #42313: remove illegal characters from username	49	Bug #42313: remove illegal characters from username
52	* "Username must only contain numbers, letters and dots, and may not be 'admin'!"	50	* "Username must only contain numbers, letters and dots, and may not be 'admin'!"
Lines 62-204 Link Here
62	name = name.strip(".")	60	name = name.strip(".")
63	return name.translate(None, bad_chars)	61	return name.translate(None, bad_chars)
64		62
65	for role in ('student', 'teacher', 'teacher_and_staff'):	63	random_puncts = list(string.punctuation)
66	random_puncts = list(string.punctuation)	64	random.shuffle(random_puncts)
67	random.shuffle(random_puncts)	65	lastname = uts.random_username(20)
68	lastnames = ["{}{}{}".format(uts.random_username(5), x, uts.random_username(5)) for x in random_puncts[:4]]	66	self.unique_basenames_to_remove.append(lastname)
69	lastnames.append(".{}".format(uts.random_username()))	67	self.log.info("Testing with dry_run=False and lastname=%r which will be shortened to %r...", lastname, lastname[:12])
70	lastnames.append(".{}.{}.".format(uts.random_username(4), uts.random_username(4)))
71	lastnames.append("{}[{}]{}".format(uts.random_username(3), uts.random_username(3), uts.random_username(3)))
72	lastnames.append(uts.random_username(40))
73	self.unique_basenames_to_remove.extend(lastnames)
74	self.log.info('*** Importing new users with role %r and the following lastnames:\n%r', role, lastnames)
75		68
76	config = copy.deepcopy(self.default_config)	69	role = 'student'
77	source_uid = 'sourceUID-%s' % (uts.random_string(),)	70	record_uid = uts.random_name()
78	config.update_entry('sourceUID', source_uid)	71	person = Person(self.ou_A.name, role)
79	config.update_entry('scheme:recordUID', '<record_uid>')	72	source_uid = 'sourceUID-%s' % (uts.random_string(),)
80	config.update_entry('user_role', role)	73	person.update(record_uid=record_uid, source_uid=source_uid, lastname=lastname)
81	config.update_entry('csv:mapping:recordUID', 'record_uid')
82	config.update_entry('scheme:username:default', "<lastname>[ALWAYSCOUNTER]")
83		74
84	persons = list()	75	config = copy.deepcopy(self.default_config)
85	names = dict()	76	config.update_entry('sourceUID', source_uid)
86	for lastname in lastnames:	77	config.update_entry('scheme:recordUID', '<record_uid>')
87	record_uid = uts.random_name()	78	config.update_entry('user_role', role)
88	person = Person(self.ou_A.name, role)	79	config.update_entry('csv:mapping:recordUID', 'record_uid')
89	person.update(record_uid=record_uid, source_uid=source_uid, lastname=lastname)	80	config.update_entry('scheme:username:default', "<lastname>[ALWAYSCOUNTER]")
90	persons.append(person)	81	fn_csv = self.create_csv_file(person_list=[person], mapping=config['csv']['mapping'])
91	names[person] = lastname	82	config.update_entry('input:filename', fn_csv)
92	fn_csv = self.create_csv_file(person_list=persons, mapping=config['csv']['mapping'])	83	fn_config = self.create_config_json(config=config)
93	config.update_entry('input:filename', fn_csv)
94	fn_config = self.create_config_json(config=config)
95		84
96	# save ldap state for later comparison	85	self.save_ldap_status()
97	self.save_ldap_status()	86	self.run_import(['-c', fn_config, '-i', fn_csv])
98	# start import	87	utils.wait_for_replication()
99	self.run_import(['-c', fn_config, '-i', fn_csv])	88	self.check_new_and_removed_users(1, 0)
100	utils.wait_for_replication()	89	person.update(username=remove_bad_chars("{}1".format(lastname[:12])))
101	# check for new users in LDAP	90	utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)
102	self.check_new_and_removed_users(len(persons), 0)	91	wait_for_drs_replication('cn={}'.format(escape_filter_chars(person.username))) # wait for creation before deletion
103	# check usernames
104	for person, lastname in names.items():
105	person.update(username=remove_bad_chars("{}1".format(lastname[:12])))
106	utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)
107	wait_for_drs_replication('cn={}'.format(escape_filter_chars(person.username))) # wait for creation before deletion
108		92
109	# delete users	93	# delete user
110	self.log.info('*** Deleting users with role %r and the following lastnames:\n%r', role, lastnames)	94	self.log.info('*** Deleting user with lastname=%r', lastname)
111	fn_csv = self.create_csv_file(person_list=[], mapping=config['csv']['mapping'])	95	fn_csv = self.create_csv_file(person_list=[], mapping=config['csv']['mapping'])
112	config.update_entry('input:filename', fn_csv)	96	config.update_entry('input:filename', fn_csv)
113	fn_config = self.create_config_json(config=config)	97	fn_config = self.create_config_json(config=config)
114	self.save_ldap_status()	98	self.save_ldap_status()
115	self.run_import(['-c', fn_config, '-i', fn_csv])	99	self.run_import(['-c', fn_config, '-i', fn_csv])
116	utils.wait_for_replication()	100	utils.wait_for_replication()
117		101
118	# recreate users, but move position of COUNTER variable	102	# recreate user with dry_run
119	self.log.info(	103	self.log.info('*** Importing user with the same lastname with --dry_run')
120	'*** Importing users with role %r and the same lastnames, but move position of COUNTER variable:\n%r',	104	# Prevent 'The email address is already taken by another user. Please change the email address.'
121	role,	105	# because of slow s4 replication. Username is the same, so it doesn't change what's tested.
122	lastnames)	106	person.mail = '{}{}'.format(uts.random_username(4), person.mail)
123	config.update_entry('scheme:username:default', "[ALWAYSCOUNTER]<lastname>")	107	fn_csv = self.create_csv_file(person_list=[person], mapping=config['csv']['mapping'])
124	for person in persons:	108	config.update_entry('input:filename', fn_csv)
125	# Prevent 'The email address is already taken by another user. Please change the email address.'	109	fn_config = self.create_config_json(config=config)
126	# because of slow s4 replication. Username is the same, so it doesn't change what's tested.	110	self.save_ldap_status()
127	person.mail = '{}{}'.format(uts.random_username(4), person.mail)	111	self.run_import(['-c', fn_config, '-i', fn_csv, '--dry-run'])
128	fn_csv = self.create_csv_file(person_list=persons, mapping=config['csv']['mapping'])	112	utils.wait_for_replication()
129	config.update_entry('input:filename', fn_csv)	113	utils.verify_ldap_object(person.dn, should_exist=False)
130	fn_config = self.create_config_json(config=config)	114	person.update(username=remove_bad_chars("{}2".format(lastname[:12])))
131	self.save_ldap_status()	115	utils.verify_ldap_object(person.dn, should_exist=False)
132	self.run_import(['-c', fn_config, '-i', fn_csv])
133	utils.wait_for_replication()
134	for person, lastname in names.items():
135	person.update(username=remove_bad_chars("2{}".format(lastname[:12])))
136	utils.verify_ldap_object(person.dn, expected_attr={'uid': [person.username]}, strict=False, should_exist=True)
137		116
138	self.log.info('*** Starting unit test for UsernameHandler.format_username() (1/5)')	117	lo, po = getAdminConnection()
139	unh = UsernameHandler(15) # 20 - len("exam-")	118	utils.verify_ldap_object(
140	name12 = uts.random_username(12) # 15 - 3	119	"cn={},cn=unique-usernames,cn=ucsschool,cn=univention,{}".format(escape_dn_chars(lastname[:12]), lo.base),
141	usernames = {	120	expected_attr={'ucsschoolUsernameNextNumber': ['2']},
142	".abc.def.": "abc.def",	121	strict=False,
143	"...abc...def...": "abc...def",	122	should_exist=True)
144	"": None,
145	"..[ALWAYSCOUNTER]..": None,
146	"[ALWAYSCOUNTER]": None,
147	"[FOObar]": "FOObar",
148	"{}.[COUNTER2]".format(name12): name12,
149	".": None,
150	'M'14 + '.' + 'A': 'M' 14,
151	}
152	for input_name, expected in usernames.items():
153	try:
154	out = unh.format_username(input_name)
155	self.unique_basenames_to_remove.append(expected)
156	if out != expected:
157	self.fail("UsernameHandler.format_username(%r) returned %r, expected %r." % (input_name, out, expected))
158	except FormatError:
159	if expected is not None:
160	self.fail("UsernameHandler.format_username(%r) raise a FormatError, expected it to return %r." % (input_name, expected))
161	continue
162	self.log.info('*** Starting unit test for UsernameHandler.format_username() (2/5)')
163	for i in range(1000):
164	name = uts.random_username(15)
165	self.unique_basenames_to_remove.append(name)
166	out = unh.format_username(name)
167	if out != name:
168	self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
169	self.log.info('*** Starting unit test for UsernameHandler.format_username() (3/5)')
170	for i in range(1000):
171	name = uts.random_name(20)
172	self.unique_basenames_to_remove.append(name)
173	out = unh.format_username(name)
174	if out.startswith(".") or out.endswith(".") or len(out) > 15:
175	self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
176	self.log.info('*** Starting unit test for UsernameHandler.format_username() (4/5)')
177	for i in range(1000):
178	name = uts.random_name_special_characters(20)
179	name = name.translate(None, "[]") # those are reserved for counter vars
180	self.unique_basenames_to_remove.append(name)
181	out = unh.format_username(name)
182	if out.startswith(".") or out.endswith(".") or len(out) > 15:
183	self.fail("UsernameHandler.format_username(%r) returned %r." % (name, out))
184	self.log.info('*** Starting unit test for UsernameHandler.format_username() (5/5)')
185	usernames = [
186	('Max[ALWAYSCOUNTER].Mustermann', 'Max1.Mustermann'),
187	('Max[ALWAYSCOUNTER].Mustermann', 'Max2.Mustermann'),
188	('Max[ALWAYSCOUNTER].Mustermann', 'Max3.Mustermann'),
189	('Max[ALWAYSCOUNTER].Mustermann', 'Max4.Mustermann'),
190	('Maria[ALWAYSCOUNTER].Musterfrau', 'Maria1.Musterfrau'),
191	('Moritz[COUNTER2]', 'Moritz'),
192	('Moritz[COUNTER2]', 'Moritz2')
193	]
194	self.unique_basenames_to_remove.extend(["Max.Mustermann", "Maria.Musterfrau", "Moritz"])
195	unh = UsernameHandler(20)
196	for input_name, expected in usernames:
197	out = unh.format_username(input_name)
198	if out != expected:
199	self.fail("UsernameHandler.format_username(%r) returned %r, expected %r." % (input_name, out, expected))
200		123
		124	self.log.info('*** OK: Import with --dry_run didn\'t raise the counter.')
201		125
		126
202	def main():	127	def main():
203	tester = Test()	128	tester = Test()
204	try:	129	try: