Implement a test covering operation of the cat functionality file.

mig/shared/accountstate.py

@@ -31,6 +31,7 @@
 
 from __future__ import print_function
 from __future__ import absolute_import
+from past.builtins import basestring
 
 import os
 import time

mig/shared/functionality/cat.py

@@ -61,6 +61,13 @@ def main(client_id, user_arguments_dict, environ=None):
 
     (configuration, logger, output_objects, op_name) = \
         initialize_main_variables(client_id)
+
+    return _main(configuration, logger, op_name=op_name, output_objects=output_objects, client_id=client_id, user_arguments_dict=user_arguments_dict)
+
+def _main(configuration, logger, op_name='', output_objects=[], client_id=None, user_arguments_dict=None, environ=None):
+    if logger is None:
+        logger = configuration.logger
+
     client_dir = client_id_dir(client_id)
     defaults = signature()[1]
     status = returnvalues.OK
@@ -71,6 +78,7 @@ def main(client_id, user_arguments_dict, environ=None):
         client_id,
         configuration,
         allow_rejects=False,
+        environ=environ,
         # NOTE: path can use wildcards, dst cannot
         typecheck_overrides={'path': valid_path_pattern},
     )

tests/fixture/MiG-users.db--example.binary

@@ -0,0 +1,187 @@
+(dp0
+V/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=b'Test User'/emailAddress=dummy-user
+p1
+(dp2
+Vfull_name
+p3
+Vb'Test User'
+p4
+sVorganization
+p5
+VTest Org
+p6
+sVstate
+p7
+VNA
+p8
+sVcountry
+p9
+VDK
+p10
+sVemail
+p11
+Vdummy-user
+p12
+sVcomment
+p13
+VThis is the create comment
+p14
+sVpassword
+p15
+V
+p16
+sVpassword_hash
+p17
+VPBKDF2$sha256$10000$b't0JM/JjkQ347th0Q'$b'QupJt53hA5KhESEeqDhTQTCPOrCBvZ6H'
+p18
+sVdistinguished_name
+p19
+V/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=b'Test User'/emailAddress=dummy-user
+p20
+sVlocality
+p21
+g16
+sVorganizational_unit
+p22
+g16
+sVexpire
+p23
+I1757925298
+sVcreated
+p24
+F1726233828.2676349
+sVunique_id
+p25
+VktyCKIRg9HvsVzXMQ22EaKS67t9atchv9JKTiJqrtBiGN3qksKrbTTYIH8mitY2K
+p26
+sVopenid_names
+p27
+(lp28
+sVold_password_hash
+p29
+VPBKDF2$sha256$10000$b'GL7Qq92iLe/hZXBo'$b'ZwB/5IZqgU7onP+ZqZk9zcHVZOx7jmWz'
+p30
+sVrenewed
+p31
+F1726389298.7801197
+ssV/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=dummy-user
+p32
+(dp33
+Vfull_name
+p34
+VTest User
+p35
+sVorganization
+p36
+VTest Org
+p37
+sVstate
+p38
+VNA
+p39
+sVcountry
+p40
+VDK
+p41
+sVemail
+p42
+Vdummy-user
+p43
+sVcomment
+p44
+VThis is the create comment
+p45
+sVpassword
+p46
+g16
+sVpassword_hash
+p47
+VPBKDF2$sha256$10000$b'kZ8WgLNH+wg3X11d'$b't1d08MV4g215WYW7S7EbkjHqDF+MCjMa'
+p48
+sVdistinguished_name
+p49
+V/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=dummy-user
+p50
+sVlocality
+p51
+g16
+sVorganizational_unit
+p52
+g16
+sVexpire
+p53
+I1759243332
+sVcreated
+p54
+F1726602273.7987707
+sVunique_id
+p55
+VKdYHJ21t37jAoHUmBq6t8Xnsnih6JWR5i0QepHoVXfDpQxz9fQGnEmegoDNrPzbe
+p56
+sVopenid_names
+p57
+(lp58
+sVold_password_hash
+p59
+VPBKDF2$sha256$10000$b'yObizsUepZvvJ0/r'$b'uKIt7n6Lf/7WXD6pKDGyvT30L2uowBnV'
+p60
+sVrenewed
+p61
+F1727707333.0969944
+ssV/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]
+p62
+(dp63
+Vfull_name
+p64
+VTest User
+p65
+sVorganization
+p66
+VTest Org
+p67
+sVstate
+p68
+VNA
+p69
+sVcountry
+p70
+VDK
+p71
+sVemail
+p72
+[email protected]
+p73
+sVcomment
+p74
+VThis is the create comment
+p75
+sVpassword
+p76
+g16
+sVpassword_hash
+p77
+VPBKDF2$sha256$10000$b'/TkhLk4yMGf6XhaY'$b'7HUeQ9iwCkE4YMQAaCd+ZdrN+y8EzkJH'
+p78
+sVdistinguished_name
+p79
+g62
+sVlocality
+p80
+g16
+sVorganizational_unit
+p81
+g16
+sVexpire
+p82
+I1758970812
+sVcreated
+p83
+F1727434813.0792377
+sVunique_id
+p84
+VaTza92klrnN2wfylm6HnphCy9C3PReGpQ6jklJ7zF3xjeaUDw36tW95Avx43vtba
+p85
+sVopenid_names
+p86
+(lp87
+ss.

tests/support/__init__.py

@@ -276,7 +276,17 @@ def is_path_within(path, start=None, _msg=None):
     return not relative.startswith('..')
 
 
-def fixturefile(relative_path, fixture_format=None):
+def _ensuredirs(absolute_dir):
+    try:
+        os.makedirs(absolute_dir)
+    except OSError as oserr:
+        if oserr.errno != errno.EEXIST:
+            raise
+
+    return absolute_dir
+
+
+def fixturefile(relative_path, fixture_format=None, include_path=False):
     """Support function for loading fixtures from their serialised format.
 
     Doing so is a little more involved than it may seem because serialisation
@@ -296,12 +306,24 @@ def fixturefile(relative_path, fixture_format=None):
     #_, extension = os.path.splitext(os.path.basename(tmp_path))
     #assert fixture_format == extension, "fixture file does not match format"
 
-    if fixture_format == 'json':
-        return _fixturefile_json(tmp_path)
+    data = None
+
+    if fixture_format == 'binary':
+        with open(tmp_path, 'rb') as binfile:
+            data = binfile.read()
+    elif fixture_format == 'json':
+        data = _fixturefile_json(tmp_path)
     else:
         raise AssertionError(
             "unsupported fixture format: %s" % (fixture_format,))
 
+    return (data, tmp_path) if include_path else data
+
+
+def fixturefile_normname(relative_path, prefix=None):
+    normname, _ = relative_path.split('--')
+    return os.path.join(prefix, normname) if prefix else normname
+
 
 _FIXTUREFILE_HINTAPPLIERS = {
     'array_of_tuples': lambda value: [tuple(x) for x in value]
@@ -340,12 +362,17 @@ def temppath(relative_path, test_case, ensure_dir=False, skip_clean=False):
     """Get absolute temp path for relative_path"""
     assert isinstance(test_case, MigTestCase)
     tmp_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
+    return _temppath(tmp_path, test_case, ensure_dir=ensure_dir, skip_clean=skip_clean)
+
+
+def _temppath(tmp_path, test_case, ensure_dir=False, skip_clean=False):
     if ensure_dir:
         try:
             os.mkdir(tmp_path)
-        except FileExistsError:
-            raise AssertionError(
-                "ABORT: use of unclean output path: %s" % relative_path)
+        except OSError as oserr:
+            if oserr.errno == errno.EEXIST:
+                raise AssertionError(
+                    "ABORT: use of unclean output path: %s" % tmp_path)
     if not skip_clean:
         test_case._cleanup_paths.add(tmp_path)
     return tmp_path

tests/test_mig_shared_functionality_cat.py

@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# test_mig_shared_functionality_cat - cat functionality unit test
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG 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 2 of the License, or
+# (at your option) any later version.
+#
+# MiG 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, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+#
+# --- END_HEADER ---
+#
+
+"""Unit tests of the MiG functionality file implementing the cat resource."""
+
+from __future__ import print_function
+import importlib
+import os
+import shutil
+import sys
+
+from tests.support import MIG_BASE, MigTestCase, testmain, \
+    fixturefile, fixturefile_normname, \
+    _ensuredirs, _temppath
+
+from mig.shared.base import client_id_dir
+from mig.shared.functionality.cat import _main as main
+
+
+def create_http_environ(configuration, wsgi_variables={}):
+    """Small helper that can create a minimum viable environ dict suitable
+    for passing to http-facing code for the supplied configuration."""
+
+    environ = {}
+    environ['MIG_CONF'] = configuration.config_file
+    environ['HTTP_HOST'] = wsgi_variables.get('http_host', 'localhost')
+    environ['PATH_INFO'] = wsgi_variables.get('path_info', '/')
+    environ['REMOTE_ADDR'] = wsgi_variables.get('remote_addr', '127.0.0.1')
+    environ['SCRIPT_URI'] = ''.join(('http://', environ['HTTP_HOST'], environ['PATH_INFO']))
+    return environ
+
+
+class MigCgibinCat(MigTestCase):
+    TEST_CLIENT_ID = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]'
+
+    def _provide_configuration(self):
+        return 'testconfig'
+
+    def before_each(self):
+        # ensure a user home directory for our test user
+        conf_user_home = self.configuration.user_home[:-1]
+        test_client_dir = client_id_dir(self.TEST_CLIENT_ID)
+        test_user_dir = os.path.join(conf_user_home, test_client_dir)
+
+        # ensure a user db that includes our test user
+        conf_user_db_home = _ensuredirs(self.configuration.user_db_home)
+        _temppath(conf_user_db_home, self)
+        db_fixture, db_fixture_file = fixturefile('MiG-users.db--example', fixture_format='binary', include_path=True)
+        test_db_file = _temppath(fixturefile_normname('MiG-users.db--example', prefix=conf_user_db_home), self)
+        shutil.copyfile(db_fixture_file, test_db_file)
+
+        # create the test user home directory
+        self.test_user_dir = _ensuredirs(test_user_dir)
+        _temppath(self.test_user_dir, self)
+        self.test_environ = create_http_environ(self.configuration)
+
+    def test_returns_file_output_with_single_file_match(self):
+        with open(os.path.join(self.test_user_dir, 'foobar.txt'), 'w'):
+            pass
+        payload = {
+            'path': ['foobar.txt'],
+        }
+
+        (output_objects, status) = main(self.configuration, self.logger, client_id=self.TEST_CLIENT_ID, user_arguments_dict=payload, environ=self.test_environ)
+        self.assertEqual(len(output_objects), 1)
+        output_obj = output_objects[0]
+        self.assertEqual(output_obj['object_type'], 'file_output')
+
+
+if __name__ == '__main__':
+    testmain()