Compatibility testing script for REST endpoints in the MediaWiki Reading Lists extension

reading-lists-compatibility-tests.py

#!/usr/bin/python

# This script destructively exercises the Reading Lists API for
# both the RESTBase implementation and the MW REST API implementation
# and compares the responses.
#
# This is for testing compatibility of the new MW REST endpoints with
# the existing RESTBase ones, to help ensure a smooth changeover. It is
# intended to be executed against WMF wikis.
#
# Note that this testing is DESTRUCTIVE. Any existing reading lists
# for the specified test user will be DESTROYED. (Unless the script
# or endpoints have a tragic bug, reading lists for other users will
# be unaffected).
#
# Some of the coding style herein is more akin to the PHP coding style
# used in MediaWiki than to the preferred Python coding style, just
# because that's what the author's fingers and eyes are currently used to.
# For a throwaway script, I make no apologies for this.
#
# Error handling is minimal. This is a testing script intended to be
# executed manually. Throwing an unhandled exception is acceptable.

import getpass
import json
import requests
import urllib.parse

OUTPUT_FILE = './rl_tests_output.txt'

# This hits live WMF production
# DEFAULT_SITE = 'https://en.wikipedia.org/w'
# DEFAULT_COOKIE_PREFIX = 'enwiki'
# DEFAULT_PROJECT = 'https://en.wikipedia.org'

# This hits WMF beta
# DEFAULT_SITE = 'https://meta.wikimedia.beta.wmflabs.org/w'
# DEFAULT_COOKIE_PREFIX = 'metawiki'
# DEFAULT_PROJECT = 'https://meta.wikimedia.org'

# This hits my local
DEFAULT_SITE = 'http://default.mediawiki.mwdd.localhost:8080/w'
DEFAULT_COOKIE_PREFIX = 'default'
DEFAULT_PROJECT = 'http://default.mediawiki.mwdd.localhost'

# ===========================================================
# Makes an HTTP GET call
# ===========================================================
def make_get_call(base_url, query_params, headers, cookie_jar):
	response = requests.get(base_url, headers=headers, params=query_params, cookies=cookie_jar)
	# if not response.content:
	# print('No response available for ' + base_url + '. Quitting.')
	# quit()
	if response.status_code != 200:
		print(response.text)
		quit()
	return response


# ===========================================================
# Makes an HTTP POST call. Body will be form-encoded.
# ===========================================================
def make_post_form_call(base_url, query_params, headers, cookie_jar, body):
	response = requests.post(base_url, headers=headers, params=query_params, cookies=cookie_jar, data=body)
	#if not response.content:
	#	print('No response available for ' + base_url + '. Quitting.')
	#	quit()
	return response


# ===========================================================
# Makes an HTTP POST call. Body will be json-encoded
# ===========================================================
def make_post_json_call(base_url, query_params, headers, cookie_jar, payload):
	response = requests.post(
		base_url,
		headers=headers | {'Content-Type': 'application/json'},
		params=query_params,
		cookies=cookie_jar,
		json=payload
	)
	# print(response)
	# if not response.content:
	#	print('No response available for ' + base_url + '. Quitting.')
	#	quit()
	return response


# ===========================================================
# Makes an HTTP PUT call. Body will be json-encoded
# ===========================================================
def make_put_json_call(base_url, query_params, headers, cookie_jar, payload):
	response = requests.put(
		base_url,
		headers=headers | {'Content-Type': 'application/json'},
		params=query_params,
		cookies=cookie_jar,
		json=payload
	)
	#if not response.content:
	#	print('No response available for ' + base_url + '. Quitting.')
	#	quit()
	return response


# ===========================================================
# Makes an HTTP DELETE call
# ===========================================================
def make_delete_call(base_url, query_params, headers, cookie_jar):
	response = requests.delete(
		base_url,
		headers=headers | {'Content-Type': 'application/json'},
		params=query_params,
		cookies=cookie_jar
	)
	#if not response.content:
	#	print('No response available for ' + base_url + '. Quitting.')
	#	quit()
	return response


# ===========================================================
# Outputs a response
# ===========================================================
def output_response(response):
	context_line = str(response.status_code) + ': ' + response.url;
	print(context_line)
	for_file = [context_line]
	if response.text == '':
		empty_response_str = '<empty response>'
		print(empty_response_str)
		for_file.append(empty_response_str)
	else:
		try:
			for_file.append(print_json(response.json()))
		except Exception as err:
			print(response.text)
			raise err
		save_output_file(for_file)

# ===========================================================
# Prints a json value. Returns the value that was printed.
# ===========================================================
def print_json(json_to_print):
	json_str = json.dumps(json_to_print, sort_keys=True, indent=4)
	print(json_str)
	return json_str


# ===========================================================
# Gets session cookie from jar.
# Returns empty string if not found
# ===========================================================
def get_cookie(cookie_prefix, postfixes, jar):
	cookie = ''

	for postfix in postfixes:
		if cookie_prefix + postfix in jar:
			cookie = jar[cookie_prefix + postfix]
			break

	return cookie


# ===========================================================
# Writes data to the output file, appending if the file already exists
# IS THIS GOING TO BE USED IN THE NEW ARRANGEMENT? DELETE IF NOT.
# ===========================================================
def save_output_file(lines):
	with open(OUTPUT_FILE, 'a') as result_file:
		for line in lines:
			result_file.write(line)
			result_file.write('\n')
		result_file.close()


# ===========================================================
# returns a cookie jar representing a login session (or False on failure)
# ===========================================================
def get_logged_in_cookie_jar(action_api_url):
	cookie_prefix = input("Enter cookie prefix (or hit <Enter> to use '" + DEFAULT_COOKIE_PREFIX + "'): ")
	if not cookie_prefix:
		cookie_prefix = DEFAULT_COOKIE_PREFIX

	user_name = input("Enter user name.: ")
	print('WARNING: ANY EXISTING READING LIST DATA FOR ' + user_name + ' WILL BE DELETED.')
	print('Hit ^c to terminate this script without deleting reading list data.\n')

	password = getpass.getpass("Enter password for user " + user_name + ":")

	print('')
	print('Getting (logged-out) session cookies and corresponding CSRF token...')
	response = make_get_call(
		action_api_url,
		{
			'action': 'query',
			'meta': 'tokens',
			'type': 'login',
			'format': 'json'
		},
		{},
		{}
	)
	logintoken = response.json()['query']['tokens']['logintoken']
	jar = response.cookies
	session_cookie = get_cookie(cookie_prefix, {'_session', 'Session'}, jar)
	print('Retrieved logintoken: ' + logintoken)
	print('Retrieved session cookie: ' + session_cookie)

	print('')
	print("Confirming (logged-out) session persists")
	response = make_get_call(
		action_api_url,
		{
			'action': 'query',
			'meta': 'siteinfo',
			'siprop': 'dbrepllag',
			'sishowalldb': '',
			'format': 'json'
		},
		{},
		jar
	)
	if get_cookie(cookie_prefix, {'_session', 'Session'}, response.cookies) != '':
		print('Session cookie not persisted. Quitting.')
		quit()

	print('')
	print("Getting (logged-in) session cookies via login...")
	response = make_post_form_call(
		action_api_url,
		{
			'action': 'clientlogin',
			'format': 'json'
		},
		{
			'content-type': 'application/x-www-form-urlencoded'
		},
		jar,
		{
			'username': user_name,
			'password': password,
			'logintoken': logintoken,
			'loginreturnurl': 'https://localhost/no_client_site_needed.php'
		}
	)
	jar = response.cookies
	session_cookie = get_cookie(cookie_prefix, {'_session', 'Session'}, jar)
	userid_cookie = get_cookie(cookie_prefix, {'UserID'}, jar)
	username_cookie = get_cookie(cookie_prefix, {'UserName'}, jar)
	print('Retrieved session cookie: ' + session_cookie)
	print('Retrieved userid cookie: ' + userid_cookie)
	print('Retrieved username cookie: ' + username_cookie)

	if not session_cookie or not userid_cookie or not username_cookie:
		jar = False

	return jar


# ===========================================================
# returns a csrf token
# ===========================================================
def get_logged_in_csrf_token(action_api_url, jar):
	print('')
	print('Getting (logged-in) CSRF token...')
	response = make_get_call(
		action_api_url,
		{
			'action': 'query',
			'meta': 'tokens',
			'type': 'csrf',
			'format': 'json'
		},
		{},
		jar
	)
	csrf_token = response.json()['query']['tokens']['csrftoken']
	print('Retrieved csrf token: ' + csrf_token)
	return csrf_token


# ===========================================================
#
# ===========================================================
def test_action_api_endpoints(action_api_url, jar):
	# Test with api.php watchlistraw endpoint
	print('')
	print("Getting logged-in user watchlist info...")
	response = make_get_call(
		action_api_url,
		{
			'action': 'query',
			'list': 'watchlistraw',
			'format': 'json'
		},
		{},
		jar
	)
	print_json(response.json())


# ===========================================================
# RESTBase requires the csrf token to be sent as a query parameter
# by the name "csrf_token". The REST API endpoints support this for
# compatibility, but prefer receiving it as "token" in the body.
# ===========================================================
def test_reading_list_endpoints(reading_lists_url, project, csrf_token, jar):
	include_restbase_compatible_slashes = True

	new_list_data_one = {
		'name': 'planets',
		'description': 'planets of the solar system'
	}

	new_list_data_two = {
		'name': 'Planets',
		'description': 'Planets of the solar system'
	}

	new_lists_batch = {
		'batch': [
			{'name': 'dogs', 'description': 'Woof!'},
			{'name': 'cats', 'description': 'meow'}
		]
	}

	# It is not necessary for the page to actually exist
	list_entry_data = {
		'project': project,
		'title': 'Earth'
	}

	print('')
	print("Tearing down lists to start clean. This will harmlessly fail if lists are not set up for this user")
	response = make_post_json_call(
		reading_lists_url + 'lists/teardown',
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token}
	)
	output_response(response)

	print('')
	print("Set up reading lists for this user. This will create a default list.")
	response = make_post_json_call(
		reading_lists_url + 'lists/setup',
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token}
	)
	output_response(response)

	print('')
	print("Retrieve the new default list")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Create a new non-default list")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_post_json_call(
		url_to_use,
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token} | new_list_data_one
	)
	output_response(response)
	new_list_id = response.json()['id']

	print('')
	print("Modify the new non-default list")
	response = make_put_json_call(
		reading_lists_url + 'lists/' + str(new_list_id),
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token} | new_list_data_two
	)
	output_response(response)

	print('')
	print("Create two more non-default lists using batch mode")
	response = make_post_json_call(
		reading_lists_url + 'lists/batch',
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token} | new_lists_batch
	)
	output_response(response)

	print('')
	print("Retrieve all the lists")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{},
		{},
		jar
	)
	output_response(response)


	print('')
	print("Retrieve only one of the lists")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{'limit': 1},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Retrieve only the next list (using pagination)")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{'limit': 1, 'next': response.json()['next']},
		{},
		jar
	)
	output_response(response)


	print('')
	print("Add an entry to the non-default list")
	url_to_use = reading_lists_url + 'lists/' + str(new_list_id) + '/entries'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_post_json_call(
		url_to_use,
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token} | list_entry_data
	)
	output_response(response)

	print('')
	print("List the entry we just added")
	url_to_use = reading_lists_url + 'lists/' + str(new_list_id) + '/entries'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{},
		{},
		jar
	)
	output_response(response)
	new_list_entry_id = response.json()['entries'][0]['id']

	print('')
	print("Retrieve only lists containing pages by a specific title")
	encoded_project = urllib.parse.quote_plus(project)
	response = make_get_call(
		reading_lists_url + 'lists/pages/' + encoded_project + '/Earth',
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Retrieve only changes since a certain time (for this call, this should be all changes to all lists)")
	encoded_timestamp = urllib.parse.quote_plus('2024-01-01T00:00:00Z')
	response = make_get_call(
		reading_lists_url + 'lists/changes/since/' + encoded_timestamp,
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Deletes the entry we just added")
	response = make_delete_call(
		reading_lists_url + 'lists/' + str(new_list_id) + '/entries/' + str(new_list_entry_id),
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Confirm the entry is gone")
	url_to_use = reading_lists_url + 'lists/' + str(new_list_id) + '/entries'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Delete the Planets list")
	response = make_delete_call(
		reading_lists_url + 'lists/' + str(new_list_id),
		{},
		{},
		jar
	)
	output_response(response)

	print('')
	print("Confirm the Planets list is gone")
	url_to_use = reading_lists_url + 'lists'
	if include_restbase_compatible_slashes:
		url_to_use += '/'
	response = make_get_call(
		url_to_use,
		{},
		{},
		jar
	)
	output_response(response)

# ============= FIDDLING
	if False:
		new_lists_batch_one_object = {
			'batch': {'name': 'sheep', 'descriptionx': 'baa'}
		}
		print('')
		print("Create a batch of lists, but not really because this is just an object")
		response = make_post_json_call(
			reading_lists_url + 'lists/batch',
			{'csrf_token': csrf_token},
			{},
			jar,
			{'token': csrf_token} | new_lists_batch_one_object
		)
		output_response(response)

		new_lists_batch_two_lists = {
			'batch': [
				{'name': 'pigs', 'descriptionx': 'oink'},
				{'name': 'cows', 'description': 'moo'}
			]
		}

		print('')
		print("Create two new lists")
		response = make_post_json_call(
			reading_lists_url + 'lists/batch',
			{'csrf_token': csrf_token},
			{},
			jar,
			{'token': csrf_token} | new_lists_batch_two_lists
		)
		output_response(response)

		print('')
		print("Retrieve a list that doesn't exist")
		url_to_use = reading_lists_url + 'lists'
		if include_restbase_compatible_slashes:
			url_to_use += '/invalidid'
		response = make_get_call(
			url_to_use,
			{},
			{},
			jar
		)
		output_response(response)

		print('')
		print("Retrieve a list with an invalid title")
		encoded_project = urllib.parse.quote_plus(project)
		response = make_get_call(
			reading_lists_url + 'lists/pages/' + encoded_project + '/%Earth',
			{},
			{},
			jar
		)
		output_response(response)

# ============= END FIDDLING

	print('')
	print("Tearing down lists")
	response = make_post_json_call(
		reading_lists_url + 'lists/teardown',
		{'csrf_token': csrf_token},
		{},
		jar,
		{'token': csrf_token}
	)
	output_response(response)

# ===========================================================
# main script routine
# ===========================================================
def main():
	print('------- begin reading-lists-compatibility-tests -------')

	site = input("Enter site (or hit <Enter> to use '" + DEFAULT_SITE + "'): ")
	if not site:
		site = DEFAULT_SITE

	action_api_url = site + '/api.php'
	rest_api_url = site + '/rest.php'
	restbase_url = site[0: site.rindex('/')] + '/api/rest_v1'

	project = input("Enter reading lists project (or hit <Enter> to use '" + DEFAULT_PROJECT + "'): ")
	if not project:
		project = DEFAULT_PROJECT

	jar = get_logged_in_cookie_jar(action_api_url)
	if not jar:
		print('Login failure. Quitting.')
		quit()

	csrf_token = get_logged_in_csrf_token(action_api_url, jar)
	if not csrf_token:
		print('Unable to retrieve csrf token. Quitting.')
		quit()

	# Not really for Reading Lists, just an example in case we need to do something like this.
	test_action_api_endpoints(action_api_url, jar)

	# This is the actual point of the script. Everything before this was setup/confirmation.
	# reading_lists_url = rest_api_url + '/readinglists/v0/'
	# reading_lists_url = restbase_url + '/data/'
	reading_lists_url = rest_api_url + '/readinglists.v0/'
	test_reading_list_endpoints(reading_lists_url, project, csrf_token, jar)

	print('------- end reading-lists-compatibility-tests -------')


# ===========================================================
#  script entry point
# ===========================================================
main()