Search This Blog

Sunday, March 14, 2021

Writing Pre-Commit Hook Scripts in Python for Subversion

It is pretty simple to write hook scripts for subversion in python even without using the subversion library. 

Here in this post i show how i wrote a pre-commit hook script to tackle three of the requirements.

  1. Length of the commit message should not be less than 50 characters
  2. The log message should have the work item id written so the commit can be linked to the work items in the Rational Change and Configuration Management Tool
  3. As the repository will only contain python scripts so nothing else should be committed into it.
  4. The changes should be recorded in the work item id mentioned.

Input Arguments

As this script will run on the server side. The inputs to this will be the transaction and the repository to which  this transaction is happening.

We tackle that by taking that as input arguments:

1
2
3
def run(argv):
	repo = argv[1]
	tranx = argv[2]

Exit First

Now we will put all the conditions where we want to exit first. We start with the log messages.
We will call the check_len_log_message function that will return the status as 0 or 1. A status 1 means error and the script needs to fail and exit.

1
2
3
4
5
	cmd = ('svnlook', 'log', repo, '-t', tranx)
	out = check_output(cmd)
	status = check_len_log_message(out)
	if status:
		sys.exit(1)

The check_len_log_message is a simple function that checks whether the len of log message is greater than 50 characters.
  
1
2
3
4
5
6
7
def check_len_log_message(log_message):
	""" Perform length check here
	"""
	stream = sys.stderr
	if len(log_message) <= 50:
		stream.write("The length of the commit message is less than 50. Exiting.")
		return 1

The Visual SVN uses the standard error stream to display messages. We are going to do the same use the Std.Error as a stream and write our error message there. Moving to the next functional check is see if work item ID is written in the correct way.
  
1
2
3
	status = check_log_message(out)
	if status:
		sys.exit(1)

The function check_log_message looks as below:
   
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def check_log_message(log_message):
	""" Perform Check here 
	"""
	stream = sys.stderr
	match = re.search(r"^Work\sItem:[0-9](.*)",log_message)
	if match:
		stream.write("This is a debug Message your commit will be linked to " + match.group(0))
		sys.exit(0)
	else:
		stream.write("Failed write 'Work Item:<ID>' space between work and item but no spaces between : and <ID>.")
		sys.exit(1)

After this we check the file type that is being modified. It should have an extension of .py for the hook to allow any change to it.
  
1
2
3
4
5
6
	cmd = ('svnlook','changed',repo,'-t',tranx)
	out = check_output(cmd)
	changed_paths = [ line[4:] for line in out.split('\n') if len(line) > 4 ]
	status = check_fileType(changed_paths)
	if status:
		sys.exit(1)

The check_fileType function can be simplified to go through all the paths and exit when it first encounters a wrong file with incorrect extension. When that happens the function exits with status 1. Otherwise if no conflict found it exits with status 0.
   
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def check_fileType(changed_list):
	"""Perform File Type Check here
	"""
	stream = sys.stderr
	for path in changed_list:
		match = re.search(r".py$",path)
		if not match:
			stream.write("You are trying to commit a non python file. Repository doesnt take non python files")
			return 1
	return 0;

Now the last thing to do is extract the ID number of the work item so that the changed paths can be populated there. The ids are extracted using the id function.
  
1
2
3
4
def extract_id(log_message):
	"return the work itemid"
	match = re.search(r"\b\d+",log_message)
	return match.group(0)

This is a simple regex that outputs the number in the log message.

Write to RTC Work Item

The code looks something like the following:
  
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
	session = requests.Session()
	session.verify = False
	session.allow_redirects = True
	session.headers = {'accept':'application/json'}
	session.auth = (username,password)
	# authenticated user
	response = session.get(base_url + auth_url)
	#print(str(response.headers))
	#print(str(response.status_code))
	if 'x-com-ibm-team-repository-web-auth-msg' in response.headers and response.headers['x-com-ibm-team-repository-web-auth-msg'] == 'authrequired':
		print("Not authenticated yet")
		login_response = session.post(base_url + '/j_security_check', data={'j_username':username, 'j_password':password})
		print(str(login_response.headers))
		print(str(login_response.status_code))
		if 'x-com-ibm-team-repository-web-auth-msg' in login_response.headers and login_response.headers['x-com-ibm-team-repository-web-auth-msg'] == 'authrequired':
			print(str(login_response.status_code))
			print("Exit HEre")
			stream.write(str(login_response.status_code))
			stream.write("Couldnt Login Try again ")
			sys.exit(1)
		response = session.get(base_url + auth_url)
	#print(str(response.headers))
	#print(str(response.status_code))
	response = session.get(bt_url + number + '.json')
	#print(str(response.status_code))
	#print(str(response.headers))
	#print(str(response.json()))
	json_data = response.json()
	task_type = json_data["dc:type"]["rdf:resource"].split('/')
	task_type = task_type[len(task_type)-1]
	# GEt List of files
	cmd = ('svnlook', 'changed', repo, '-t', tranx)
	out = check_output(cmd)
	cmd = ('svnlook', 'author', repo, '-t', tranx)
	authorid = check_output(cmd)
	my_dict = {}
	if task_type == "task":
		data = json_data["rtc_cm:com.bombardier.team.workitem.attribute.result-large"]
		my_dict["rtc_cm:com.bombardier.team.workitem.attribute.result-large"] = data + "\n" + authorid + out
	if task_type == "change_request" or task_type == "feature_request" or task_type == "defect":
		data = json_data["rtc_cm:com.bombardier.team.workitem.attribute.log.implementation"]
		my_dict["rtc_cm:com.bombardier.team.workitem.attribute.log.implementation"] = data + "\n" + authorid + out
	session.headers = {'Content-Type':'application/x-oslc-cm-change-request+json'}
	response = session.put(bt_url + number + '.json',data=json.dumps(my_dict))
	if response.status_code != 200:
		stream.write("Couldnt Update the RTC ID")
		sys.exit(1)

  1. The session is opened. 
  2. The redirect allow is set to true as the most of the users will have their Jazz application redirecting to the form URL.
  3. We first attempt to get the authentication from the authentication URL
  4. In case the authentication is not done prior it attempts to do that by posing the credentials to the redirected form URL
  5. Once authentication is successful we get the json form of the extracted id.
  6. There we check which type of work item it is. If it is task we take in the result attribute and append the changed path and the author information to that box
  7. If the task type is any of the formal items like requests then the implementation log is taken into the account and the changed path and author information is appended to that box.
  8. After it is appended the payload is prepared and posted to the same work item JSON URL.
  9. If we get a success code from the POST request we gracefully exit and if not we throw the error and exit with status 1 preventing any commits to the repository.
import os
import sys
from subprocess import check_output
import re
bt_url = "https://localhost:9443/ccm/oslc/workitems/"
bt_auth_url = "https://localhost:9443/ccm/"
base_url = "https://localhost:9443/ccm"
auth_url = '/authenticated/identity'
username = "username"
password = "password"
def check_log_message(log_message):
""" Perform Check here
"""
stream = sys.stderr
match = re.search(r"^Work\sItem:[0-9](.*)",log_message)
if match:
stream.write("This is a debug Message your commit will be linked to " + match.group(0))
sys.exit(0)
else:
stream.write("Failed write 'Work Item:<ID>' space between work and item but no spaces between : and <ID>.")
sys.exit(1)
def check_len_log_message(log_message):
""" Perform length check here
"""
stream = sys.stderr
if len(log_message) <= 50:
stream.write("The length of the commit message is less than 50. Exiting.")
return 1
def check_fileType(changed_list):
"""Perform File Type Check here
"""
stream = sys.stderr
for path in changed_list:
match = re.search(r".py$",path)
if not match:
stream.write("You are trying to commit a non python file. Repository doesnt take non python files")
return 1
return 0;
def extract_id(log_message):
"return the work itemid"
match = re.search(r"\b\d+",log_message)
return match.group(0)
def run(argv):
repo = argv[1]
tranx = argv[2]
cmd = ('svnlook', 'log', repo, '-t', tranx)
out = check_output(cmd)
status = check_len_log_message(out)
if status:
sys.exit(1)
status = check_log_message(out)
if status:
sys.exit(1)
number = extract_id(out)
stream = sys.stderr
stream.write(number)
cmd = ('svnlook','changed',repo,'-t',tranx)
out = check_output(cmd)
changed_paths = [ line[4:] for line in out.split('\n') if len(line) > 4 ]
status = check_fileType(changed_paths)
if status:
sys.exit(1)
# Here
session = requests.Session()
session.verify = False
session.allow_redirects = True
session.headers = {'accept':'application/json'}
session.auth = (username,password)
# authenticated user
response = session.get(base_url + auth_url)
#print(str(response.headers))
#print(str(response.status_code))
if 'x-com-ibm-team-repository-web-auth-msg' in response.headers and response.headers['x-com-ibm-team-repository-web-auth-msg'] == 'authrequired':
print("Not authenticated yet")
login_response = session.post(base_url + '/j_security_check', data={'j_username':username, 'j_password':password})
print(str(login_response.headers))
print(str(login_response.status_code))
if 'x-com-ibm-team-repository-web-auth-msg' in login_response.headers and login_response.headers['x-com-ibm-team-repository-web-auth-msg'] == 'authrequired':
print(str(login_response.status_code))
print("Exit HEre")
stream.write(str(login_response.status_code))
stream.write("Couldnt Login Try again ")
sys.exit(1)
response = session.get(base_url + auth_url)
#print(str(response.headers))
#print(str(response.status_code))
response = session.get(bt_url + number + '.json')
#print(str(response.status_code))
#print(str(response.headers))
#print(str(response.json()))
json_data = response.json()
task_type = json_data["dc:type"]["rdf:resource"].split('/')
task_type = task_type[len(task_type)-1]
# GEt List of files
cmd = ('svnlook', 'changed', repo, '-t', tranx)
out = check_output(cmd)
cmd = ('svnlook', 'author', repo, '-t', tranx)
authorid = check_output(cmd)
my_dict = {}
if task_type == "task":
data = json_data["rtc_cm:com.bombardier.team.workitem.attribute.result-large"]
my_dict["rtc_cm:com.bombardier.team.workitem.attribute.result-large"] = data + "\n" + authorid + out
if task_type == "change_request" or task_type == "feature_request" or task_type == "defect":
data = json_data["rtc_cm:com.bombardier.team.workitem.attribute.log.implementation"]
my_dict["rtc_cm:com.bombardier.team.workitem.attribute.log.implementation"] = data + "\n" + authorid + out
session.headers = {'Content-Type':'application/x-oslc-cm-change-request+json'}
response = session.put(bt_url + number + '.json',data=json.dumps(my_dict))
if response.status_code != 200:
stream.write("Couldnt Update the RTC ID")
sys.exit(1)
if __name__ == '__main__':
run(sys.argv)
view raw log-police.py hosted with ❤ by GitHub
py -2 log-police.py %1 %2
REM "Minimum Character size of 50"
"%VISUALSVN_SERVER%\bin\VisualSVNServerHooks.exe" check-logmessage "%1" -t "%2" --min-size 50
IF ERRORLEVEL 1 exit /b 1
view raw svn_pre_hook_visualsvn.bat hosted with ❤ by GitHub
The full code for this is captured in the gist below.
Posted by at 9:10 PM
Labels: ,

No comments:

Post a Comment

Subscribe to: Post Comments (Atom)