Added a 'set' method for overwriting existing users with new ones.

pyolite/models/lists/users.py

@@ -21,7 +21,6 @@ def decorated(self, string_user, *args, **kwargs):
                 user = User(self.repository_model.path,
                             self.repository_model.git,
                             string_user)
-
             return func(self, user, *args, **kwargs)
 
         return decorated
@@ -70,6 +69,31 @@ def remove(self, user):
                                          (user.name,
                                           self.repository_model.name))
 
+    @with_user
+    def get_or_create(self, user):
+        return user
+
+    def set(self, users=None):
+        users_serialized = ""
+        if isinstance(users, dict):
+            users = users.iteritems()
+
+        if users:
+            for user, permission in users:
+                if not hasattr(user, 'name'):
+                    user = self.get_or_create(user)
+
+                users_serialized += "    %s     =    %s\n" % (permission,
+                                                              user.name)
+
+        self.repo.overwrite(users_serialized)
+
+        users = ", ".join((user for user, permission in users))
+        commit_message = "Initialized repository %s with users: %s" % (
+            self.repository_model.name, users
+        )
+        self.repository_model.git.commit(['conf'], commit_message)
+
     def __iter__(self):
         for user in self._user:
             yield user

pyolite/repo.py

@@ -39,3 +39,9 @@ def write(self, string):
             fcntl.flock(f, fcntl.LOCK_EX)
             f.write(string)
             fcntl.flock(f, fcntl.LOCK_UN)
+
+    def overwrite(self, string):
+        with open(self.path, 'w') as f:
+            fcntl.flock(f, fcntl.LOCK_EX)
+            f.write(string)
+            fcntl.flock(f, fcntl.LOCK_UN)

tests/lists/test_users.py

@@ -1,5 +1,6 @@
 from mock import MagicMock, patch, call
 from nose.tools import raises
+from pyolite.models.user import User
 from spec import Spec
 
 from pyolite.models.lists.users import ListUsers
@@ -122,3 +123,56 @@ def test_user_edit_permissions(self):
 
             message = "User another_user has R permission for repository test_repo"
             mocked_repository.git.commit.has_calls([call(['conf'], message)])
+
+    def user_init(self, *args, **kwargs):
+        self.name = args[2]
+
+    @patch('pyolite.models.user.User.__init__', new=user_init)
+    def test_user_get_or_create(self):
+        mocked_repo = MagicMock()
+        mocked_repository = MagicMock()
+
+        mocked_repository.name = 'test_repo'
+        mocked_repository.path = 'path'
+
+        with patch.multiple('pyolite.models.lists.users',
+                            Repo=MagicMock(return_value=mocked_repo)):
+            found_user = object()
+            mocked_user_get = MagicMock(return_value=found_user)
+
+            with patch.multiple('pyolite.models.user.User', get=mocked_user_get):
+                repo_users = ListUsers(mocked_repository)
+
+                # user found
+                user = repo_users.get_or_create('test_user')
+                assert user is found_user
+
+                # user created
+                mocked_user_get.side_effect = ValueError
+                user = repo_users.get_or_create('test_user')
+                assert user.name is 'test_user'
+
+    def test_users_set(self):
+        mocked_repo = MagicMock()
+        mocked_repository = MagicMock()
+        mocked_user = MagicMock()
+
+        mock_single_user = MagicMock()
+        mock_single_user.name = 'user'
+
+        mocked_repository.name = 'test_repo'
+        mocked_repository.path = 'path'
+
+        mocked_user.get.return_value = mock_single_user
+
+        with patch.multiple('pyolite.models.lists.users',
+                            Repo=MagicMock(return_value=mocked_repo),
+                            User=mocked_user):
+            repo_users = ListUsers(mocked_repository)
+            repo_users.set((('mocked', 'R'), ('mocked', 'RW+')))
+
+            serialized_users = "    R     =    user\n    RW+     =    user\n"
+            mocked_repo.overwrite.assert_called_once_with(serialized_users)
+
+            message = "Initialized repository test_repo with users: test, user"
+            mocked_repository.git.commit.has_calls([call(['conf'], message)])

tests/test_repo.py

@@ -43,7 +43,7 @@ def test_it_should_retrieve_all_users_from_repo(self):
             repo = Repo(mocked_path)
             eq_(repo.users, ['user1', 'user2'])
 
-    def test_it_shoujld_write_to_repo_config(self):
+    def test_it_should_write_to_repo_config(self):
         path = 'tests/fixtures/empty_repo.conf'
 
         Repo(path).write('some_text')
@@ -55,6 +55,20 @@ def test_it_shoujld_write_to_repo_config(self):
             f.write('')
             f.truncate()
 
+    def test_it_should_overwrite_the_repo_config(self):
+        path = 'tests/fixtures/empty_repo.conf'
+
+        Repo(path).write('some_text')
+
+        Repo(path).overwrite('another_text')
+
+        with open(path, 'r+') as f:
+            eq_('another_text', f.read())
+
+            f.seek(0)
+            f.write('')
+            f.truncate()
+
     def test_replace_filelocking(self):
         mocked_re = MagicMock()
         mocked_fcntl = MagicMock()
@@ -145,3 +159,32 @@ def test_write_filelocking(self):
                 # asserts lock has been removed after writing
                 mocked_fcntl.flock.assert_called_once_with(manager,
                                                            mocked_fcntl.LOCK_UN)
+
+    def test_overwrite_filelocking(self):
+        path = 'tests/fixtures/empty_repo.conf'
+        mocked_path = MagicMock()
+        mocked_path.__str__ = lambda x: path
+
+        mocked_fcntl = MagicMock()
+        mocked_open = MagicMock()
+
+        with patch('__builtin__.open', mocked_open):
+            manager = mocked_open.return_value.__enter__.return_value
+
+            # asserts file locking has been put in place before writing
+            manager.write = lambda text: ([
+                mocked_fcntl.flock.assert_called_once_with(
+                    manager, mocked_fcntl.LOCK_EX
+                ),
+                mocked_fcntl.reset_mock()
+            ])
+
+            with patch.multiple('pyolite.repo', fcntl=mocked_fcntl):
+                repo = Repo(path)
+
+                mocked_fcntl.reset_mock()
+                repo.overwrite('some_text')
+
+                # asserts lock has been removed after writing
+                mocked_fcntl.flock.assert_called_once_with(manager,
+                                                           mocked_fcntl.LOCK_UN)