Adding upstream version 2:4.20.0+dfsg.upstream/2%4.20.0+dfsg

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 244 insertions, 0 deletions

diff --git a/source3/script/tests/test_smbcquota.py b/source3/script/tests/test_smbcquota.py
new file mode 100755
index 0000000..02229d7
--- /dev/null
+++ b/source3/script/tests/test_smbcquota.py
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+# Unix SMB/CIFS implementation.
+# Tests for smbcquotas
+# Copyright (C) Noel Power 2017
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os, subprocess, sys
+import traceback
+import logging
+import shutil
+
+USER_QUOTAS = 1
+USER_DEFAULT_QUOTAS = 2
+GROUP_QUOTAS = 3
+GROUP_DEFAULT_QUOTAS = 4
+BLOCK_SIZE = 1024
+DEFAULT_SOFTLIM = 2
+DEFAULT_HARDLIM = 4
+
+class test_env:
+    def __init__(self):
+        self.server = None
+        self.domain = None
+        self.username = None
+        self.password = None
+        self.envdir = None
+        self.quota_db = None
+        self.smbcquotas = None
+        self.users = []
+
+class user_info:
+    def __init__(self):
+        self.uid = 0
+        self.username = ""
+        self.softlim = 0
+        self.hardlim = 0
+
+class Quota:
+    def __init__(self):
+        self.flags = 0
+        self.quotatype = USER_DEFAULT_QUOTAS
+        self.uid = 0
+        self.usedblocks = 0
+        self.softlimit = 0
+        self.hardlimit = 0
+        self.hardlimit = 0
+        self.usedinodes = 0
+        self.slimitinodes = 0
+        self.hlimitinodes = 0
+
+def init_quota_db(users, output_file):
+    filecontents = open(output_file,"w+")
+    lines = ""
+    default_values = "0 " + str(DEFAULT_SOFTLIM) + " " + str(DEFAULT_HARDLIM) + " 0 0 0"
+    for user in users:
+        lines = lines + user.uid + " " + default_values + "\n"
+    filecontents.write(lines)
+    filecontents.close()
+
+def load_quotas(input_file):
+    fileContents = open(input_file,"r")
+    lineno = 0
+    quotas = []
+    for line in fileContents:
+        if line.strip().startswith("#"):
+            continue
+        content = line.strip().split()
+        quota = Quota()
+        if len(content) < 7:
+            logging.debug("ignoring line %d, doesn't have enough fields\n"%lineno)
+        else:
+            quota.flags = 2
+            quota.uid = content[0]
+            quota.usedblocks = content[1]
+            quota.softlimit = content[2]
+            quota.hardlimit = content[3]
+            quota.usedinodes = content[4]
+            quota.slimitinodes = content[5]
+            quota.hlimitinodes = content[6]
+            quotas.append(quota)
+
+    fileContents.close()
+    return quotas
+
+def get_quotas(uid, quota_list):
+    for quota in quota_list:
+        if quota.uid == uid:
+            return quota
+    return None
+
+def get_users():
+    output = subprocess.Popen(['getent', 'passwd'],
+                              stdout=subprocess.PIPE).communicate()[0].decode("utf-8").split('\n')
+    users = []
+    for line in output:
+        info = line.split(':')
+        if len(info) > 3 and info[0]:
+            user = user_info()
+            user.username = info[0]
+            user.uid = info[2]
+            logging.debug("Adding user ->%s<-\n"%user.username)
+            users.append(user)
+    return users
+
+
+
+def smbcquota_output_to_userinfo(output):
+    infos = []
+    for line in output:
+        if len(line) > 1:
+            username = line.strip(':').split()[0]
+            quota_info = line.split(':')[1].split('/')
+            if len(quota_info) > 2:
+                info = user_info()
+                info.username = username.strip()
+                info.softlim = int(quota_info[1].strip()) / BLOCK_SIZE
+                info.hardlim = int(quota_info[2].strip()) / BLOCK_SIZE
+                infos.append(info)
+    return infos
+
+def check_quota_limits(infos, softlim, hardlim):
+    if len(infos) < 1:
+        logging.debug("no users info to check :-(\n")
+        return False
+    for info in infos:
+        if int(info.softlim) != softlim:
+            logging.debug("expected softlimit %s got ->%s<-\n"%(softlim, info.softlim))
+            return False
+        if int(info.hardlim) != hardlim:
+            logging.debug("expected hardlimit limit %s got %s\n"%(hardlim,info.hardlim))
+            return False
+    return True
+
+class test_base:
+    def __init__(self, env):
+        self.env = env
+    def run(self, protocol):
+        pass
+
+class listtest(test_base):
+    def run(self, protocol):
+        init_quota_db(self.env.users, self.env.quota_db)
+        quotas = load_quotas(self.env.quota_db)
+        args = [self.env.smbcquotas]
+        remaining_args = ['-U' + self.env.username + "%" + self.env.password, '-L', '//' + self.env.server + '/quotadir']
+        if protocol == 'smb2':
+            args.append('-m smb2')
+        args.extend(remaining_args)
+        output = subprocess.Popen([self.env.smbcquotas, '-U' + self.env.username + "%" + self.env.password, '-L', '//' + self.env.server + '/quotadir'], stdout=subprocess.PIPE).communicate()[0].decode("utf-8").split('\n')
+        infos = smbcquota_output_to_userinfo(output)
+        return check_quota_limits(infos, DEFAULT_SOFTLIM, DEFAULT_HARDLIM)
+def get_uid(name, users):
+    for user in users:
+        if user.username == name:
+            return user.uid
+    return None
+
+class gettest(test_base):
+    def run(self, protocol):
+        init_quota_db(self.env.users, self.env.quota_db)
+        quotas = load_quotas(self.env.quota_db)
+        uid = get_uid(self.env.username, self.env.users)
+        output = subprocess.Popen([self.env.smbcquotas, '-U' + self.env.username + "%" + self.env.password, '-u' + self.env.username, '//' + self.env.server + '/quotadir'], stdout=subprocess.PIPE).communicate()[0].decode("utf-8").split('\n')
+        user_infos = smbcquota_output_to_userinfo(output)
+        db_user_info = get_quotas(uid, quotas)
+        # double check, we compare the results from the db file
+        # the quota script the server uses compared to what
+        # smbcquota is telling us
+        return check_quota_limits(user_infos, int(db_user_info.softlimit), int(db_user_info.hardlimit))
+
+class settest(test_base):
+    def run(self, protocol):
+        init_quota_db(self.env.users, self.env.quota_db)
+        quotas = load_quotas(self.env.quota_db)
+        uid = get_uid(self.env.username, self.env.users)
+        old_db_user_info = get_quotas(uid, quotas)
+
+        #increase limits by 2 blocks
+        new_soft_limit = (int(old_db_user_info.softlimit) + 2) * BLOCK_SIZE
+        new_hard_limit = (int(old_db_user_info.hardlimit) + 2) * BLOCK_SIZE
+
+        new_limits = "UQLIM:%s:%d/%d"%(self.env.username, new_soft_limit, new_hard_limit)
+        logging.debug("setting new limits %s"%new_limits)
+
+        output = subprocess.Popen([self.env.smbcquotas, '-U' + self.env.username + "%" + self.env.password, '//' + self.env.server + '/quotadir', '-S', new_limits], stdout=subprocess.PIPE).communicate()[0].decode("utf-8").split('\n')
+        logging.debug("output from smbcquota is %s"%output)
+        user_infos = smbcquota_output_to_userinfo(output)
+        return check_quota_limits(user_infos, new_soft_limit / BLOCK_SIZE, new_hard_limit / BLOCK_SIZE)
+
+# map of tests
+subtest_descriptions = {
+        "list test" : listtest,
+        "get test" : gettest,
+        "set test" : settest
+}
+
+def main():
+    logging.basicConfig(format='%(asctime)s %(message)s', level=logging.DEBUG)
+
+    logging.debug("got args %s\n"%str(sys.argv))
+
+    if len(sys.argv) < 7:
+        logging.debug ("Usage: test_smbcquota.py server domain username password envdir smbcquotas\n")
+        sys.exit(1)
+    env = test_env()
+    env.server = sys.argv[1]
+    env.domain = sys.argv[2]
+    env.username = sys.argv[3]
+    env.password = sys.argv[4]
+    env.envdir = sys.argv[5]
+    env.smbcquotas = sys.argv[6]
+    quota_script = os.path.join(os.path.dirname(sys.argv[0]),
+                                "getset_quota.py")
+    #copy the quota script to the environment
+    shutil.copy2(quota_script, env.envdir)
+
+    env.quota_db = os.path.join(env.envdir, "quotas.db")
+    env.users = get_users()
+    for protocol in ['smb1', 'smb2']:
+        for key in subtest_descriptions.keys():
+            test = subtest_descriptions[key](env)
+            logging.debug("running subtest '%s' using protocol '%s'\n"%(key,protocol))
+            result = test.run(protocol)
+            if result == False:
+                logging.debug("subtest '%s' for '%s' failed\n"%(key,protocol))
+                sys.exit(1)
+            else:
+                logging.debug("subtest '%s' for '%s' passed\n"%(key,protocol))
+    sys.exit(0)
+
+if __name__ == '__main__':
+    main()