# -*- encoding: UTF8 -*-
# Test harness for swarm_rename_users.py
from __future__ import print_function
import sys
import unittest
import os
import json
sys.path.append(os.path.join('..', '..', 'triggers', 'tests'))
sys.path.append('..')
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)
import swarm_rename_users as sru
os.environ["LOGS"] = "."
LOGGER_NAME = "TestSwarmRenameUsers"
LOG_FILE = "log-SwarmRenameUsers.log"
python3 = sys.version_info[0] >= 3
class TestSwarmRenameUsersDetails(TestCase):
def __init__(self, methodName='runTest'):
super(TestSwarmRenameUsersDetails, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName)
def setUp(self):
pass
def tearDown(self):
pass
def testActivities(self):
json1 = """
{"type":"change","link":["change",{"change":2072659}],
"user":"olduser","action":"committed",
"behalfOf":null,"projects":[]}
"""
json2 = """
{"type":"change","link":["change",{"change":2072659}],
"user":"fred","action":"committed",
"behalfOf":"olduser","projects":[]}
"""
json3 = """
{"type":"change","link":["change",{"change":2072659}],
"user":"fred","action":"committed",
"behalfOf":null,"projects":[]}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Activity(json.loads(json1))
self.assertEqual('olduser', obj.user())
obj.Update(usersToRename)
self.assertEqual('newuser', obj.user())
self.assertTrue(obj.updated)
obj = sru.Activity(json.loads(json2))
self.assertEqual('fred', obj.user())
self.assertEqual('olduser', obj.behalfOf())
obj.Update(usersToRename)
self.assertEqual('fred', obj.user())
self.assertEqual('newuser', obj.behalfOf())
self.assertTrue(obj.updated)
obj = sru.Activity(json.loads(json3))
self.assertEqual('fred', obj.user())
self.assertEqual(None, obj.behalfOf())
obj.Update(usersToRename)
self.assertEqual('fred', obj.user())
self.assertEqual(None, obj.behalfOf())
self.assertFalse(obj.updated)
def testComments(self):
json1 = """
{"id":null,"topic":"changes\/577295","context":null,
"user":"olduser","time":1358803378,"body":"Adding comment"}
"""
json2 = """
{"id":null,"topic":"changes\/577295","context":null,
"user":"olduser","readBy":[],"time":1358803378,"body":"Adding comment"}
"""
json3 = """
{"id":null,"topic":"changes\/577295","context":null,
"user":"jim","readBy":["olduser","fred"],"time":1358803378,"body":"Adding comment"}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Comment(json.loads(json1))
self.assertEqual('olduser', obj.user())
self.assertEqual(None, obj.readBy())
obj.Update(usersToRename)
self.assertEqual('newuser', obj.user())
self.assertEqual(None, obj.readBy())
self.assertTrue(obj.updated)
obj = sru.Comment(json.loads(json2))
self.assertEqual('olduser', obj.user())
self.assertEqual([], obj.readBy())
obj.Update(usersToRename)
self.assertEqual('newuser', obj.user())
self.assertEqual([], obj.readBy())
self.assertTrue(obj.updated)
obj = sru.Comment(json.loads(json3))
self.assertEqual('jim', obj.user())
self.assertEqual(['olduser', 'fred'], obj.readBy())
obj.Update(usersToRename)
self.assertEqual('jim', obj.user())
self.assertEqual(['newuser', 'fred'], obj.readBy())
self.assertTrue(obj.updated)
def testProjects(self):
# Test project with no renames for owners/moderators
json1 = """
{"name": "Test Project",
"defaults": {
"reviewers": []
},
"description": "Test for demo purposes",
"owners": [
"fred",
"bill"
],
"moderators": [
"fred",
"bill"
]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json1))
self.assertEqual(['fred', 'bill'], obj.owners())
self.assertEqual(['fred', 'bill'], obj.moderators())
obj.Update(usersToRename)
self.assertEqual(['fred', 'bill'], obj.owners())
self.assertEqual(['fred', 'bill'], obj.moderators())
self.assertFalse(obj.updated)
# Test owners/moderators updated
json2 = """
{"name": "Test Project",
"defaults": {
"reviewers": []
},
"description": "Test for demo purposes",
"owners": [
"olduser",
"bill"
],
"moderators": [
"olduser",
"bill"
]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json2))
self.assertEqual(['olduser', 'bill'], obj.owners())
self.assertEqual(['olduser', 'bill'], obj.moderators())
obj.Update(usersToRename)
self.assertEqual(['newuser', 'bill'], obj.owners())
self.assertEqual(['newuser', 'bill'], obj.moderators())
self.assertTrue(obj.updated)
# Test default:reviewers not updated
json3 = """
{"name": "Test Project",
"defaults": {
"reviewers": ["jim"]
},
"description": "Test for demo purposes",
"owners": ["fred"],
"moderators": ["jim"]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json3))
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(['jim'], obj.defaultReviewers())
obj.Update(usersToRename)
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(['jim'], obj.defaultReviewers())
self.assertFalse(obj.updated)
# Test default:reviewers are updated
json4 = """
{"name": "Test Project",
"defaults": {
"reviewers": ["olduser", "jim"]
},
"description": "Test for demo purposes",
"owners": ["fred"],
"moderators": ["jim"]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json4))
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(['olduser', 'jim'], obj.defaultReviewers())
obj.Update(usersToRename)
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(['newuser', 'jim'], obj.defaultReviewers())
self.assertTrue(obj.updated)
# Test branches:defaults:reviewers not updated
json5 = """
{"name": "Test Project",
"description": "Test for demo purposes",
"owners": ["fred"],
"moderators": ["jim"],
"branches": [
{
"id": "main",
"defaults": {
"reviewers": []
},
"moderators": []
}
]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json5))
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': []}], obj.branchDefaultReviewers())
self.assertEqual([{'main': []}], obj.branchModerators())
obj.Update(usersToRename)
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': []}], obj.branchModerators())
self.assertEqual([{'main': []}], obj.branchDefaultReviewers())
self.assertFalse(obj.updated)
json6 = """
{"name": "Test Project",
"description": "Test for demo purposes",
"owners": ["fred"],
"moderators": ["jim"],
"branches": [
{
"id": "main",
"defaults": {
"reviewers": ["jim"]
},
"moderators": ["john"]
}
]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json6))
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': ["jim"]}], obj.branchDefaultReviewers())
self.assertEqual([{'main': ["john"]}], obj.branchModerators())
obj.Update(usersToRename)
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': ["jim"]}], obj.branchDefaultReviewers())
self.assertEqual([{'main': ["john"]}], obj.branchModerators())
self.assertFalse(obj.updated)
# Test branches:defaults:reviewers and branches:moderators are updated
json7 = """
{"name": "Test Project",
"description": "Test for demo purposes",
"owners": ["fred"],
"moderators": ["jim"],
"branches": [
{
"id": "main",
"defaults": {
"reviewers": ["jim", "olduser"]
},
"moderators": ["john", "olduser"]
}
]
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Project(json.loads(json7))
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': ["john", "olduser"]}], obj.branchModerators())
self.assertEqual([{'main': ["jim", "olduser"]}], obj.branchDefaultReviewers())
obj.Update(usersToRename)
self.assertEqual(['fred'], obj.owners())
self.assertEqual(['jim'], obj.moderators())
self.assertEqual(None, obj.defaultReviewers())
self.assertEqual([{'main': ["john", "newuser"]}], obj.branchModerators())
self.assertEqual([{'main': ["jim", "newuser"]}], obj.branchDefaultReviewers())
self.assertTrue(obj.updated)
def testReviews(self):
# Test review with no renames for versions/author/approvals/participants
json1 = """
{
"changes": [12345],
"commits": [123456],
"versions": [{
"difference": 1,
"stream": null,
"user": "fred",
"time": 1611858237,
"pending": true,
"addChangeMode": "replace"}
],
"author": "fred",
"approvals": {
"mike": [1, 3],
"fred": [3]
},
"participants": {
"mike": {"vote": {"value": 1,"version": 3}},
"fred": [],
"swarm-group-test": {
"minimumRequired": "0"
}
}
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Review(json.loads(json1))
self.assertEqual('fred', obj.author())
self.assertEqual(['mike', 'fred'], obj.approvals())
self.assertEqual(['mike', 'fred', 'swarm-group-test'], obj.participants())
self.assertEqual(['fred'], obj.versionUsers())
obj.Update(usersToRename)
self.assertEqual('fred', obj.author())
self.assertEqual(['mike', 'fred'], obj.approvals())
self.assertEqual(['mike', 'fred', 'swarm-group-test'], obj.participants())
self.assertEqual(['fred'], obj.versionUsers())
self.assertFalse(obj.updated)
# Test review with renames for versions/author/approvals/participants
json2 = """
{
"changes": [12345],
"commits": [123456],
"versions": [{
"difference": 1,
"stream": null,
"user": "olduser",
"time": 1611858237,
"pending": true,
"addChangeMode": "replace"}
],
"author": "olduser",
"approvals": {
"olduser": [1, 3],
"fred": [3]
},
"participants": {
"olduser": {"vote": {"value": 1,"version": 3}},
"fred": [],
"swarm-group-test": {
"minimumRequired": "0"
}
}
}
"""
obj = sru.Review(json.loads(json2))
self.assertEqual('olduser', obj.author())
self.assertEqual(['olduser', 'fred'], obj.approvals())
self.assertEqual(['olduser', 'fred', 'swarm-group-test'], obj.participants())
self.assertEqual(['olduser'], obj.versionUsers())
obj.Update(usersToRename)
self.assertEqual('newuser', obj.author())
self.assertEqual(['fred', 'newuser'], obj.approvals())
self.assertEqual(['fred', 'swarm-group-test', 'newuser'], obj.participants())
self.assertEqual(['newuser'], obj.versionUsers())
self.assertEqual([1, 3], obj.json['approvals']['newuser'])
self.assertEqual({'vote': {'value': 1, 'version': 3}}, obj.json['participants']['newuser'])
self.assertTrue(obj.updated)
def testFileInfo(self):
# Test review with no renames for readBy
json1 = """
{"readBy":{
"jim":{"version":28,"digest":"EEFF792157ADBB2311938D7358F0588B"},
"mike":{"version":30,"digest":"1E43325205374FBBE0E72DAB5930F8DB"}
}
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.FileInfo(json.loads(json1))
self.assertEqual(['jim', 'mike'], obj.readBy())
obj.Update(usersToRename)
self.assertEqual(['jim', 'mike'], obj.readBy())
self.assertFalse(obj.updated)
# Test review with no renames for readBy
json1 = """
{"readBy":{
"olduser":{"version":28,"digest":"EEFF792157ADBB2311938D7358F0588B"},
"mike":{"version":30,"digest":"1E43325205374FBBE0E72DAB5930F8DB"}
}
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.FileInfo(json.loads(json1))
self.assertEqual(['olduser', 'mike'], obj.readBy())
obj.Update(usersToRename)
self.assertEqual(['mike', 'newuser'], obj.readBy())
self.assertEqual({'version': 28, 'digest': 'EEFF792157ADBB2311938D7358F0588B'}, obj.json['readBy']['newuser'])
self.assertTrue(obj.updated)
def testWorkflow(self):
# Test workflow with no renames for owners
json1 = """
{"on_submit":{"with_review":{"rule":"approved","mode":"inherit"},
"without_review":{"rule":"reject","mode":"inherit"}},
"owners":["swarm-group-dev","dave", "scott"],
"end_rules":{"update":{"rule":"no_checking","mode":"inherit"}}
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Workflow(json.loads(json1))
self.assertEqual(['swarm-group-dev', 'dave', 'scott'], obj.owners())
obj.Update(usersToRename)
self.assertEqual(['swarm-group-dev', 'dave', 'scott'], obj.owners())
self.assertFalse(obj.updated)
# Test workflow with renames for owners
json1 = """
{"on_submit":{"with_review":{"rule":"approved","mode":"inherit"},
"without_review":{"rule":"reject","mode":"inherit"}},
"owners":["swarm-group-dev","olduser", "scott"],
"end_rules":{"update":{"rule":"no_checking","mode":"inherit"}}
}
"""
usersToRename = {'olduser': 'newuser'}
obj = sru.Workflow(json.loads(json1))
self.assertEqual(['swarm-group-dev', 'olduser', 'scott'], obj.owners())
obj.Update(usersToRename)
self.assertEqual(['swarm-group-dev', 'newuser', 'scott'], obj.owners())
self.assertTrue(obj.updated)
class TestSwarmRenameUsers(TestCase):
def __init__(self, methodName='runTest'):
super(TestSwarmRenameUsers, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName)
def setUp(self):
pass
def tearDown(self):
pass
def setupServer(self):
self.server = P4Server()
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)
return p4
def testRename(self):
"""Test renaming of users"""
p4 = self.setupServer()
usersToRename = {'olduser': 'newuser'}
data = [{"key": "swarm-activity-ffe71695",
"val":'{"type":"change","link":["change",{"change":2072657}],"user":"olduser","action": "committed","behalfOf":null,"projects":[]}'},
{"key": "swarm-activity-ffe71696",
"val": '{"type":"change","link":["change",{"change":2072658}],"user":"fred","action": "committed","behalfOf":"olduser","projects":[]}'},
{"key": "swarm-activity-ffe71697",
"val": '{"type":"change","link":["change",{"change":2072659}],"user":"fred","action":"committed","behalfOf":null,"projects":[]}'},
{"key": 'swarm-comment-0000010435',
"val": '{"id":null,"topic":"changes/577295","context":null,"user":"olduser","time":1358803378,"body":"Adding comment"}'},
{"key": 'swarm-comment-0000010437',
"val": '{"id":null,"topic":"changes/577295","context":null,"user":"jim","readBy":["olduser","fred"],"time":1358803378,"body":"Adding comment"}'},
{"key": 'swarm-project-test',
"val": '{"name":"Test project","defaults":{"reviewers":[]},"description":"test desc","owners":["fred","olduser"],"branches":[{"id":"main","name":"Main","workflow":null,"paths":["//test/main/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]},{"id":"p1","name":"p1","workflow":null,"paths":["//test/p16.1/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]},{"id":"r16-1","name":"r16.1","workflow":null,"paths":["//test/r1/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]}],"jobview":"","emailFlags":{"change_email_project_users":"1","review_email_project_members":"1"},"tests":{"enabled":false,"url":"","postBody":"","postFormat":"url"},"deploy":{"enabled":false,"url":""},"deleted":false,"private":false,"workflow":null,"retainDefaultReviewers":false,"minimumUpVotes":null,"upgrade":1,"id":"test-proj"}'},
{"key": 'swarm-user-fred',
"val": '{"follows":[],"delayedComments":null,"user_notification_settings":null'},
{"key": 'swarm-user-olduser',
"val": '{"follows":[],"delayedComments":null,"user_notification_settings":null'},
]
for d in data:
p4.run_key(d["key"], d["val"])
obj = sru.SwarmRenameUsers()
obj.renameUsers(p4, usersToRename)
expected = {"swarm-activity-ffe71695": '{"type":"change","link":["change",{"change":2072657}],"user":"newuser","action":"committed","behalfOf":null,"projects":[]}',
"swarm-activity-ffe71696": '{"type":"change","link":["change",{"change":2072658}],"user":"fred","action":"committed","behalfOf":"newuser","projects":[]}',
"swarm-activity-ffe71697": '{"type":"change","link":["change",{"change":2072659}],"user":"fred","action":"committed","behalfOf":null,"projects":[]}'
}
fdata = p4.run_keys("-e", "swarm-activity-*")
for d in fdata:
self.assertTrue((d["key"] in expected))
self.assertEqual(expected[d["key"]], d["value"])
expected = {'swarm-comment-0000010435':
'{"id":null,"topic":"changes/577295","context":null,"user":"newuser","time":1358803378,"body":"Adding comment"}',
'swarm-comment-0000010437':
'{"id":null,"topic":"changes/577295","context":null,"user":"jim","readBy":["newuser","fred"],"time":1358803378,"body":"Adding comment"}',
}
fdata = p4.run_keys("-e", "swarm-comment-*")
for d in fdata:
self.assertTrue((d["key"] in expected))
self.assertEqual(expected[d["key"]], d["value"])
self.maxDiff = None
expected = {'swarm-project-test': '{"name":"Test project","defaults":{"reviewers":[]},"description":"test desc","owners":["fred","newuser"],"branches":[{"id":"main","name":"Main","workflow":null,"paths":["//test/main/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]},{"id":"p1","name":"p1","workflow":null,"paths":["//test/p16.1/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]},{"id":"r16-1","name":"r16.1","workflow":null,"paths":["//test/r1/..."],"defaults":{"reviewers":[]},"minimumUpVotes":null,"retainDefaultReviewers":false,"moderators":[],"moderators-groups":[]}],"jobview":"","emailFlags":{"change_email_project_users":"1","review_email_project_members":"1"},"tests":{"enabled":false,"url":"","postBody":"","postFormat":"url"},"deploy":{"enabled":false,"url":""},"deleted":false,"private":false,"workflow":null,"retainDefaultReviewers":false,"minimumUpVotes":null,"upgrade":1,"id":"test-proj"}'}
fdata = p4.run_keys("-e", "swarm-project-*")
for d in fdata:
self.assertTrue((d["key"] in expected))
self.assertEqual(expected[d["key"]], d["value"])
expected = {'swarm-user-fred': '{"follows":[],"delayedComments":null,"user_notification_settings":null',
'swarm-user-newuser':'{"follows":[],"delayedComments":null,"user_notification_settings":null'}
fdata = p4.run_keys("-e", "swarm-user-*")
for d in fdata:
self.assertTrue((d["key"] in expected))
self.assertEqual(expected[d["key"]], d["value"])
if __name__ == '__main__':
unittest.main()