# -*- encoding: UTF8 -*- # Test harness for CreateSwarmReview.py from __future__ import print_function import sys import unittest import os from p4testutils import TestCase, P4Server, localDirectory, create_file, append_to_file parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, parent_dir) 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 class TestCreateSwarmReview(TestCase): def __init__(self, methodName='runTest'): super(TestCreateSwarmReview, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName) 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 = P4Server() config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "~test_config.yaml") p4 = self.server.p4 p4.logger = self.logger 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 patch_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) self.assertEqual(patch_call_count, requests_mock.patch.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. mock_post_response = Mock() mock_post_response.json.return_value = {"review": { "id":123, "author": "swarmtest", "commits": [817], "commitStatus":[],"created":1509095267,"deployDetails":[],"deployStatus":None, "description":"asdfasdf","groups":["longlived"], "participants": {"swarmtest":[], "user1":[], "user2":[]}, "state": "needsReview"}} # Assign our mock response as the result of our patched function requests_mock.post.return_value = mock_post_response trig = CreateSwarmReview(*trig_args) result = trig.run() self.assertEqual(0, result) post_call_count += 1 # Expect an update patch_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) self.assertEqual(patch_call_count, requests_mock.patch.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': [('change', '1'), ('description', "Please review me!\nDon't forget to check YYYY"), ('reviewers[]', 'user1'), ('reviewers[]', 'user2')]}, kwargs) args, kwargs = requests_mock.patch.call_args_list[patch_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews/123',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': [('author', p4.user)]}, 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 patch_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) self.assertEqual(patch_call_count, requests_mock.patch.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': [('change', '3'), ('description', "Please review me!\nDon't forget to check YYYY"), ('reviewers[]', 'user1'), ('reviewers[]', 'user2')]}, kwargs) args, kwargs = requests_mock.patch.call_args_list[patch_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews/123',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': [('author', p4.user)]}, 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) self.assertEqual(patch_call_count, requests_mock.patch.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 patch_call_count += 1 self.assertEqual(get_call_count, requests_mock.get.call_count) self.assertEqual(post_call_count, requests_mock.post.call_count) self.assertEqual(patch_call_count, requests_mock.patch.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': [('change', '5'), ('description', "test job2\ninside_file1 edited again\nExtra text"), ('reviewers[]', 'user1'), ('reviewers[]', 'user2')]}, kwargs) args, kwargs = requests_mock.patch.call_args_list[patch_call_count - 1] self.assertEqual(('http://swarm.dev/api/v6/reviews/123',), args) self.assertDictEqual({'auth': ('swarmtest', 'A123453'), 'data': [('author', p4.user)]}, kwargs) if __name__ == '__main__': unittest.main()