python类struct_passwd()的实例源码

def test_01__get_user_from_host(self, mock_pwd):
        """Test NixAuthentication()._get_user_from_host()."""
        self._init()
        usr = pwd.struct_passwd(["root", "*", "0", "0", "root usr",
                                 "/root", "/bin/bash"])
        mock_pwd.getpwuid.return_value = usr
        auth = udocker.NixAuthentication()
        (name, uid, gid, gecos, _dir, shell) = auth._get_user_from_host(0)
        self.assertEqual(name, usr.pw_name)
        self.assertEqual(uid, usr.pw_uid)
        self.assertEqual(gid, str(usr.pw_gid))
        self.assertEqual(gecos, usr.pw_gecos)
        self.assertEqual(_dir, usr.pw_dir)
        self.assertEqual(shell, usr.pw_shell)
        #
        mock_pwd.getpwnam.return_value = usr
        auth = udocker.NixAuthentication()
        (name, uid, gid, gecos, _dir, shell) = auth._get_user_from_host("root")
        self.assertEqual(name, usr.pw_name)
        self.assertEqual(uid, usr.pw_uid)
        self.assertEqual(gid, str(usr.pw_gid))
        self.assertEqual(gecos, usr.pw_gecos)
        self.assertEqual(_dir, usr.pw_dir)

def get_users(self):
        """Returns a list of all local users on the computer.

        Each user is represented as a dict with the keys: C{username},
        C{name}, C{uid}, C{enabled}, C{location}, C{work-phone} and
        C{home-phone}.
        """
        users = []
        found_usernames = set()
        for user in self.get_user_data():
            if not isinstance(user, struct_passwd):
                user = struct_passwd(user)
            if user.pw_name in found_usernames:
                continue
            gecos_data = [x or None for x in user.pw_gecos.split(",")[:4]]
            while len(gecos_data) < 4:
                gecos_data.append(None)
            name, location, work_phone, home_phone = tuple(gecos_data)
            enabled = user.pw_name not in self.locked_users
            users.append({"username": user.pw_name, "name": name,
                          "uid": user.pw_uid, "enabled": enabled,
                          "location": location, "work-phone": work_phone,
                          "home-phone": home_phone,
                          "primary-gid": user.pw_gid})
            found_usernames.add(user.pw_name)
        return users

def _getpwnam(self, real, name):
        return pwd.struct_passwd((name,) + self._users[name])

def test_addUser(self):
        """
        L{UserDatabase.addUser} accepts seven arguments, one for each field of
        a L{pwd.struct_passwd}, and makes the new record available via
        L{UserDatabase.getpwuid}, L{UserDatabase.getpwnam}, and
        L{UserDatabase.getpwall}.
        """
        username = 'alice'
        password = 'secr3t'
        uid = 123
        gid = 456
        gecos = 'Alice,,,'
        home = '/users/alice'
        shell = '/usr/bin/foosh'

        db = self.database
        db.addUser(username, password, uid, gid, gecos, home, shell)

        for [entry] in [[db.getpwuid(uid)], [db.getpwnam(username)],
                        db.getpwall()]:
            self.assertEqual(entry.pw_name, username)
            self.assertEqual(entry.pw_passwd, password)
            self.assertEqual(entry.pw_uid, uid)
            self.assertEqual(entry.pw_gid, gid)
            self.assertEqual(entry.pw_gecos, gecos)
            self.assertEqual(entry.pw_dir, home)
            self.assertEqual(entry.pw_shell, shell)

def test_addUser(self):
        """
        L{UserDatabase.addUser} accepts seven arguments, one for each field of
        a L{pwd.struct_passwd}, and makes the new record available via
        L{UserDatabase.getpwuid}, L{UserDatabase.getpwnam}, and
        L{UserDatabase.getpwall}.
        """
        username = 'alice'
        password = 'secr3t'
        lastChange = 17
        min = 42
        max = 105
        warn = 12
        inact = 3
        expire = 400
        flag = 3

        db = self.database
        db.addUser(username, password, lastChange, min, max, warn, inact,
                   expire, flag)

        for [entry] in [[db.getspnam(username)], db.getspall()]:
            self.assertEqual(entry.sp_nam, username)
            self.assertEqual(entry.sp_pwd, password)
            self.assertEqual(entry.sp_lstchg, lastChange)
            self.assertEqual(entry.sp_min, min)
            self.assertEqual(entry.sp_max, max)
            self.assertEqual(entry.sp_warn, warn)
            self.assertEqual(entry.sp_inact, inact)
            self.assertEqual(entry.sp_expire, expire)
            self.assertEqual(entry.sp_flag, flag)

def _getpwnam(username):
        v = struct_passwd()
        v.pw_name = username
        v.pw_uid = -1
        return v

def _getpwuid(uid):
        v = struct_passwd()
        v.pw_name = "unknown"
        v.pw_uid = uid
        return v

def load_passwd(dirpath):
        # check if we need to reload cache
        passwd_file = os.path.join(dirpath, "etc/passwd")
        passwd_stamp = os.stat(passwd_file).st_mtime
        if passwd_stamp <= users_lastupdate.get(dirpath, -1):
                return
        users[dirpath] = user = {}
        uids[dirpath] = uid = {}
        f = open(passwd_file)
        for line in f:
                arr = line.rstrip().split(":")
                if len(arr) != 7:
                        # Skip any line we can't make sense of.
                        continue
                try:
                        arr[2] = int(arr[2])
                        arr[3] = int(arr[3])
                except ValueError:
                        # Skip any line we can't make sense of.
                        continue
                pw_entry = pwd.struct_passwd(arr)

                user[pw_entry.pw_name] = pw_entry
                # Traditional systems allow multiple users to have the same
                # user id, so only the first one should be mapped to the
                # current pw_entry.
                uid.setdefault(pw_entry.pw_uid, pw_entry)

        users_lastupdate[dirpath] = passwd_stamp
        f.close()