# -*- encoding: UTF8 -*- # Test harness for CreateSwarmReview.py from __future__ import print_function import sys import unittest import os import re sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import testp4 import P4 import logging import subprocess from CreateSwarmReview import CreateSwarmReview os.environ["LOGS"] = "." LOGGER_NAME = "TestCreateSwarmReview" LOG_FILE = "log-CreateSwarmReview.log" python3 = sys.version_info[0] >= 3 if python3: from unittest.mock import patch, Mock else: from mock import patch, Mock def ensureDirectory(directory): if not os.path.isdir(directory): os.makedirs(directory) def localDirectory(root, *dirs): "Create and ensure it exists" dir_path = os.path.join(root, *dirs) ensureDirectory(dir_path) return dir_path def create_file(file_name, contents): "Create file with specified contents" ensureDirectory(os.path.dirname(file_name)) if python3: contents = bytes(contents.encode()) with open(file_name, 'wb') as f: f.write(contents) def append_to_file(file_name, contents): "Append contents to file" if python3: contents = bytes(contents.encode()) with open(file_name, 'ab+') as f: f.write(contents) class TestCreateSwarmReview(unittest.TestCase): def __init__(self, methodName='runTest'): super(TestCreateSwarmReview, self).__init__(methodName=methodName) self.logger = logging.getLogger(LOGGER_NAME) self.logger.setLevel(logging.DEBUG) logformat = '%(levelname)s [%(asctime)s] [%(filename)s : %(lineno)d] - %(message)s' logging.basicConfig(format=logformat, filename=LOG_FILE, level=logging.DEBUG) def assertRegex(self, *args, **kwargs): if python3: return super(TestCreateSwarmReview, self).assertRegex(*args, **kwargs) else: return super(TestCreateSwarmReview, self).assertRegexpMatches(*args, **kwargs) def setUp(self): pass def tearDown(self): pass @patch('CreateSwarmReview.requests') def testCreateSwarmReview(self, requests_mock): """trigger fires and sends expected info to Swarm""" self.server = testp4.P4Server() trig_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "CreateSwarmReview.py") config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_config.yaml") p4 = self.server.p4 p4.logger = self.logger # # This works if no spaces in server root pathname! # port = p4.port.replace('"', '') # self.logger.debug("port: |%s|" % port) # triggers = p4.fetch_triggers() # triggers['Triggers'] = ['create_swarm_review change-commit //... " python ' + trig_path + # " -p %quote%" + port + "%quote% -u " + p4.user + # " -c %s " % config_path + # ' %change% "'] # self.logger.debug(triggers) # p4.save_triggers(triggers) # # Reconnect to pick up changes p4.disconnect() p4.connect() inside = localDirectory(self.server.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') p4.run('add', inside_file1) result = p4.run('submit', '-d', 'inside_file1 added') self.assertEquals("1", result[-1]['submittedChange']) outside = localDirectory(self.server.client_root, "outside") outside_file1 = os.path.join(outside, "outside_file1") create_file(outside_file1, 'Test content') p4.run('add', outside_file1) result = p4.run('submit', '-d', 'outside_file1 added') self.assertEquals("2", result[-1]['submittedChange']) p4.run('edit', inside_file1) append_to_file(inside_file1, "New line") result = p4.run('submit', '-d', 'inside_file1 edited') self.assertEquals("3", result[-1]['submittedChange']) p4.run('edit', inside_file1) append_to_file(inside_file1, "New line") result = p4.run('submit', '-d', 'inside_file1 edited again') self.assertEquals("4", result[-1]['submittedChange']) trig_args = ["-p", p4.port, "-u", p4.user, "-c", config_path, "--test-mode", "1"] get_call_count = 0 post_call_count = 0 # Test exclusionary mapping with open(config_path, "w") as f: f.write(""" api: "api/v6" user: swarmtest ticket: A123453 review_description: - "Please review me!" - "Don't forget to check YYYY" projects: - name: ProjectA create_review: y require_job: y update_review: n depot_paths: - //depot/inside/... - "-//depot/inside/*_file1" default_reviewers: - user1 - user2 """) trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) # Hasn't been called self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) with open(config_path, "w") as f: f.write(""" api: "api/v6" user: swarmtest ticket: A123453 review_description: - "Please review me!" - "Don't forget to check YYYY" projects: - name: ProjectA create_review: y require_job: y update_review: y depot_paths: - //depot/inside/... default_reviewers: - user1 - user2 """) #---------------------------------------- # For a path config file defines as inside we should call. trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) post_call_count += 1 # Expect an update self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) args, kwargs = requests_mock.post.call_args_list[post_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': {'reviewers': ['user1', 'user2'], 'description': "Please review me!\nDon't forget to check YYYY", 'change': '1'}}, kwargs) #---------------------------------------- # For a path config file defined as outside the project we should not call Swarm trig_args[-1] = "2" trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) # Hasn't been called again! self.assertEqual(1, requests_mock.post.call_count) #---------------------------------------- # If no job exists then same file results in a new review trig_args[-1] = "3" trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) post_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) args, kwargs = requests_mock.post.call_args_list[post_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': {'reviewers': ['user1', 'user2'], 'description': "Please review me!\nDon't forget to check YYYY", 'change': '3'}}, kwargs) #---------------------------------------- # With a job, review is updated job = p4.fetch_job() job['Description'] = 'test job' p4.save_job(job) p4.run('fix', '-c', '1', 'job000001') p4.run('fix', '-c', '4', 'job000001') fixes = p4.run('fixes', '-j', 'job000001') self.assertEquals(2, len(fixes)) # Construct our mock response object, giving it relevant expected behaviours mock_response = Mock() review_response = {"lastSeen":813, "reviews":[ {"id":815,"author":"perforce","changes":[814,816,817],"comments":[0,0],"commits":[817], "commitStatus":[],"created":1509095267,"deployDetails":[],"deployStatus":None, "description":"asdfasdf","groups":["longlived"], "participants":{"perforce":[]},"pending":False,"projects":[],"state":"approved", "stateLabel":"Approved","testDetails":[],"testStatus":None,"type":"default", "updated":1509096701,"updateDate":"2017-10-27T09:31:41+00:00"}], "totalCount":1} mock_response.json.return_value = review_response # Assign our mock response as the result of our patched function requests_mock.get.return_value = mock_response trig_args[-1] = "4" trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) get_call_count += 1 post_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) args, kwargs = requests_mock.post.call_args_list[post_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews/815/changes/',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': {'change': '4', 'id': '815'}}, kwargs) #---------------------------------------- # Now just create another job and submit a change # Test description templating with open(config_path, "w") as f: f.write(""" api: "api/v6" user: swarmtest ticket: A123453 review_description: - "$jobDescription" - "$changeDescription" - "Extra text" projects: - name: ProjectA create_review: y require_job: y update_review: y depot_paths: - //depot/inside/... default_reviewers: - user1 - user2 """) p4.run('edit', inside_file1) append_to_file(inside_file1, "New line") result = p4.run('submit', '-d', 'inside_file1 edited again') self.assertEquals("5", result[-1]['submittedChange']) job = p4.fetch_job() job['Description'] = 'test job2' p4.save_job(job) p4.run('fix', '-c', '5', 'job000002') trig_args[-1] = "5" trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) post_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) args, kwargs = requests_mock.post.call_args_list[post_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': {'reviewers': ['user1', 'user2'], 'description': "test job2\n\ninside_file1 edited again\nExtra text", 'change': '5'}}, kwargs) if __name__ == '__main__': unittest.main()
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#6 | 23150 | Robert Cowham | Refactor to move tests into subdir | ||
#5 | 23138 | Robert Cowham | Refactored tests | ||
#4 | 23132 | Robert Cowham |
Search for jobs and update existing review Also add default reviewers |
||
#3 | 23122 | Robert Cowham | Add functionality to allow depot paths to be specified in groups | ||
#2 | 23073 | Robert Cowham | Use mocks for unit tests | ||
#1 | 23070 | Robert Cowham | Initial versions |