Add failing (regression) test case which verifies non remote user roles
without is_remote field are correctly deleted when syncing user role
assignments.
# See the License for the specific language governing permissions and
# limitations under the License.
+from pymongo import MongoClient
+
from st2tests.base import CleanDbTestCase
from st2common.persistence.auth import User
from st2common.persistence.rbac import Role
@@ -22,6 +24,7 @@ from st2common.models.db.auth import UserDB
from st2common.models.api.rbac import RoleDefinitionFileFormatAPI
from st2common.models.api.rbac import UserRoleAssignmentFileFormatAPI
from st2common.services.rbac import get_roles_for_user
+from st2common.services.rbac import get_role_assignments_for_user
from st2common.services.rbac import create_role
from st2common.services.rbac import assign_role_to_user
from st2common.services.rbac import create_group_to_role_map
@@ -57,6 +60,10 @@ class BaseRBACDefinitionsDBSyncerTestCase(CleanDbTestCase):
user_3_db = User.add_or_update(user_3_db)
self.users['user_3'] = user_3_db
+ user_5_db = UserDB(name='user_5')
+ user_5_db = User.add_or_update(user_5_db)
+ self.users['user_5'] = user_5_db
+
class RBACDefinitionsDBSyncerTestCase(BaseRBACDefinitionsDBSyncerTestCase):
def test_sync_roles_no_definitions(self):
@@ -171,6 +178,33 @@ class RBACDefinitionsDBSyncerTestCase(BaseRBACDefinitionsDBSyncerTestCase):
role_dbs = get_roles_for_user(user_db=self.users['user_1'])
self.assertItemsEqual(role_dbs, [self.roles['role_1']])
+ def test_sync_user_assignments_assignments_without_is_remote_are_deleted(self):
+ # Test case which verifies roles without "is_remote" field (pre v2.3) are removed from the
+ # database
+ syncer = RBACDefinitionsDBSyncer()
+
+ # Note: User use pymongo to insert mock data because we want to insert a
+ # raw document and skip mongoengine to leave is_remote field unpopulated
+ client = MongoClient()
+ db = client['st2-test']
+ db.user_role_assignment_d_b.insert_one({'user': 'user_5', 'role': 'role_1'})
+ db.user_role_assignment_d_b.insert_one({'user': 'user_5', 'role': 'role_2'})
+ db.user_role_assignment_d_b.insert_one({'user': 'user_5', 'role': 'role_3',
+ 'is_remote': False})
+ db.user_role_assignment_d_b.insert_one({'user': 'user_5', 'role': 'role_4',
+ 'is_remote': True})
+
+ # Initial state, 4 roles
+ role_assignment_dbs = get_role_assignments_for_user(user_db=self.users['user_5'])
+ self.assertEqual(len(role_assignment_dbs), 4)
+
+ # All 3 non remote roles should have been deleted
+ syncer.sync_users_role_assignments(role_assignment_apis=[])
+ role_assignment_dbs = get_role_assignments_for_user(user_db=self.users['user_5'])
+ self.assertEqual(len(role_assignment_dbs), 1)
+ self.assertEqual(role_assignment_dbs[0].name, 'role_4')
+ self.assertEqual(role_assignment_dbs[0].is_remote, True)
+
def test_sync_user_assignments_multiple_custom_roles_assignments(self):
syncer = RBACDefinitionsDBSyncer()
