Source

vasm-users / model.py

Full commit
# -*- encoding: utf-8 -*-

"""A model for the users VASM module. Provides a way for accessing users data
from the system, and for adding, deleting and editing accounts.
"""


__author__ = "rbistolfi"
__mail__ = "moc.liamg@iflotsibr"[::-1]
__date__ = "14/06/2010"


import pwd


class UserAddError(Exception):
    """Exception raised when user addition fails."""
    pass


class User(object):
    """A class representing a Linux user account."""

    __slots__ = [ "uid",
            "name",
            "group",
            "groups",
            "home",
            "shell",
            "comment",
            "password",
            "expire",
            "inactive" ]


class UserFactory(object):
    """A Users factory. Query the system for existing users and produce a list
    of User instances.

    """

    def _create(self, user_data):
        """Return a User object initialized with user_data."""

        user_obj = User()
        user_obj.name = user_data[0]
        user_obj.password = user_data[1]
        user_obj.uid = user_data[2]
        user_obj.group = user_data[3]
        user_obj.comment = user_data[4]
        user_obj.home = user_data[5]
        user_obj.shell = user_data[6]

        return user_obj

    def fetch_all(self):
        """Return a list containing one User instance for each existing user in
        the system.

        >>> factory = UserFactory()
        >>> all_users = factory.fetch_all()
        >>> root = [ i for i in all_users if i.uid == 0 ][0]
        >>> root.uid
        0
        >>> root.name
        'root'
        >>> root.home
        '/root'
        
        """

        users = []
        for user in pwd.getpwall():
            user_obj = self._create(user)
            users.append(user_obj)
        return users

    def from_uid(self, uid):
        """Creates a User instance from a user id.
        
        >>> factory = UserFactory()
        >>> root = factory.from_uid(0)
        >>> root.uid
        0
        >>> root.name
        'root'
        >>> root.home
        '/root'
 
        """

        return self._create(pwd.getpwuid(uid))

    def from_name(self, name):
        """Creates a User instance from username.
        
        >>> factory = UserFactory()
        >>> root = factory.from_name('root')
        >>> root.uid
        0
        >>> root.name
        'root'
        >>> root.home
        '/root'
        
        """

        return self._create(pwd.getpwnam(name))


class Observable(object):
    """Implement some methods for subscribing callables to events."""

    signals = []
    
    def __init__(self):
        self._observers = {}

    def connect(self, signal, func, *args, **kwargs):
        """Connect a callable object to a signal.

        Parameters:
            signal: a signal registered in cls.signal
            func: a callable object that will be executed when signal is
                  emited.

        Any extra parameters will be passed to the connected callable,
        after the object affected by the signal.

        >>> ul = UsersList()
        >>> tmp = []
        >>> def on_user_added(obj, extra):
        ...     tmp.append(obj)
        >>> ul.connect("user-added", on_user_added, "This is an extra parameter")
        >>> user = User()
        >>> ul.append(user)
        >>> tmp[0] is user
        True

        """
       
        assert callable(func), "%r should be a callable" % func
        assert signal in self.signals, "Unknown signal: %s" % signal
        
        self._observers.setdefault(signal, []).append([func, (args, kwargs)])

    def disconnect(self, signal, func):
        """Disconnect a callable conected to the given signal.

        Parameters:
            signal: a signal registered in cls.signal
            func: a connected callable object.

        >>> ul = UsersList()
        >>>
        >>> def on_user_deleted(user):
        ...     return 1
        >>>
        >>> ul.connect("user-deleted", on_user_deleted)
        >>> "user-deleted" in ul._observers
        True
        >>> on_user_deleted in ul._observers["user-deleted"][-1]
        True
        >>> ul.disconnect("user-deleted", on_user_deleted)
        >>> ul._observers["user-deleted"]
        []
        
       
        """

        if signal in self._observers:
            for connected in self._observers[signal]:
                if func in connected:
                    self._observers[signal].remove(connected)

    def emit(self, signal, obj):
        """Emit a signal and execute its callbacks."""
        
        if signal in self._observers:
            for func, arguments in self._observers[signal]:
                args, kwargs = arguments
                func(obj, *args, **kwargs)


class UsersList(list, Observable):
    """A container for User instances.
    This is an observable container, you can connect callables with any signal
    present in cls.signals:

        - "changed":
            Emited when any of the append, remove or __setitem__ methods of the
            container are called.
        
        - "user-added":
            Emited when a User instance is appended to the container.

        - "user-deleted":
            Emited when a User instance is removed from the container with the
            remove method.

        - "user-updated":
            Emited when a new User instance is assigned to a container position
            using a subscript.
    
    >>> users = UsersList()
    >>> def on_user_added(user):
    ...     pass
    >>> users.connect("user-added", on_user_added)
    >>> a_user = User()
    >>> users.append(a_user) # on_user_added is called

    If the factory class attribute has been defined, it will be used to
    initialize the instance, populating the container with the objects 
    returned by the factory. This initialization will not emit any signal and
    can be used for representing the initial state of the system accounts.
 
    """
    
    signals = ["changed", "user-added", "user-updated", "user-deleted"]
    factory = UserFactory

    def __init__(self):
        """UsersList initialization. Populates the List with existing users
        without emiting any signal.
        
        """
        
        list.__init__(self)
        Observable.__init__(self)

        # populate the list with existing users
        if self.factory:
            factory = self.factory()
            for user in factory.fetch_all():        
                assert isinstance(user, User), \
                        "A User instance was expected, got %r instead" % user

                super(UsersList, self).append(user)
        
    def __setitem__(self, index, user):
        """Override list.__setitem__ for triggering the needed signals.
        
        >>> users_list = UsersList()
        >>> user = User()
        >>> user.name = 'Guido'
        >>> users_list[0] = user
        >>> users_list[0].name
        'Guido'
        >>> users_list[1] = 1        
        Traceback (most recent call last):
            ...
        AssertionError: A User instance was expected, got 1 instead

        """

        assert isinstance(user, User), "A User instance was expected," \
                " got %r instead" % user

        super(UsersList, self).__setitem__(index, user)

        self.emit("changed", self)
        self.emit("user-updated", user)

    def append(self, user):
        """Add a user to the users list.
        
        >>> user = User()
        >>> users_list = UsersList()
        >>> users_list.append(user)
        >>> user in users_list
        True
        >>> users_list.append(user)
        Traceback (most recent call last):
            ...
        UserAddError: User exists
        >>> users_list.append(1)
        Traceback (most recent call last):
            ...
        AssertionError: A User instance was expected, got 1 instead

        """

        assert isinstance(user, User), "A User instance was expected," \
                " got %r instead" % user


        if user not in self:
            super(UsersList, self).append(user)
        else:
            raise UserAddError("User exists")

        self.emit("changed", self)
        self.emit("user-added", user)

    def remove(self, user):
        """Remove a user from the users list.

        >>> user = User()
        >>> users_list = UsersList()
        >>> users_list.append(user)
        >>> user in users_list
        True
        >>> users_list.remove(user)
        >>> user in users_list
        False
        >>> users_list.remove(1)
        Traceback (most recent call last):
            ...
        AssertionError: A User instance was expected, got 1 instead

        """

        assert isinstance(user, User), "A User instance was expected," \
                " got %r instead" % user

        super(UsersList, self).remove(user)

        self.emit("changed", self)
        self.emit("user-deleted", user)

if __name__ == '__main__':
    import doctest
    doctest.testmod(verbose=True)