summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_pkgutil.py
blob: b3fc5d84e74365f191b11a398361491943815057 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
from test.test_support import run_unittest
import unittest
import sys
import imp
import pkgutil
import os
import os.path
import tempfile
import shutil
import zipfile



class PkgutilTests(unittest.TestCase):

    def setUp(self):
        self.dirname = tempfile.mkdtemp()
        sys.path.insert(0, self.dirname)

    def tearDown(self):
        del sys.path[0]
        shutil.rmtree(self.dirname)

    def test_getdata_filesys(self):
        pkg = 'test_getdata_filesys'

        # Include a LF and a CRLF, to test that binary data is read back
        RESOURCE_DATA = 'Hello, world!\nSecond line\r\nThird line'

        # Make a package with some resources
        package_dir = os.path.join(self.dirname, pkg)
        os.mkdir(package_dir)
        # Empty init.py
        f = open(os.path.join(package_dir, '__init__.py'), "wb")
        f.close()
        # Resource files, res.txt, sub/res.txt
        f = open(os.path.join(package_dir, 'res.txt'), "wb")
        f.write(RESOURCE_DATA)
        f.close()
        os.mkdir(os.path.join(package_dir, 'sub'))
        f = open(os.path.join(package_dir, 'sub', 'res.txt'), "wb")
        f.write(RESOURCE_DATA)
        f.close()

        # Check we can read the resources
        res1 = pkgutil.get_data(pkg, 'res.txt')
        self.assertEqual(res1, RESOURCE_DATA)
        res2 = pkgutil.get_data(pkg, 'sub/res.txt')
        self.assertEqual(res2, RESOURCE_DATA)

        del sys.modules[pkg]

    def test_getdata_zipfile(self):
        zip = 'test_getdata_zipfile.zip'
        pkg = 'test_getdata_zipfile'

        # Include a LF and a CRLF, to test that binary data is read back
        RESOURCE_DATA = 'Hello, world!\nSecond line\r\nThird line'

        # Make a package with some resources
        zip_file = os.path.join(self.dirname, zip)
        z = zipfile.ZipFile(zip_file, 'w')

        # Empty init.py
        z.writestr(pkg + '/__init__.py', "")
        # Resource files, res.txt, sub/res.txt
        z.writestr(pkg + '/res.txt', RESOURCE_DATA)
        z.writestr(pkg + '/sub/res.txt', RESOURCE_DATA)
        z.close()

        # Check we can read the resources
        sys.path.insert(0, zip_file)
        res1 = pkgutil.get_data(pkg, 'res.txt')
        self.assertEqual(res1, RESOURCE_DATA)
        res2 = pkgutil.get_data(pkg, 'sub/res.txt')
        self.assertEqual(res2, RESOURCE_DATA)
        del sys.path[0]

        del sys.modules[pkg]

class PkgutilPEP302Tests(unittest.TestCase):

    class MyTestLoader(object):
        def load_module(self, fullname):
            # Create an empty module
            mod = sys.modules.setdefault(fullname, imp.new_module(fullname))
            mod.__file__ = "<%s>" % self.__class__.__name__
            mod.__loader__ = self
            # Make it a package
            mod.__path__ = []
            # Count how many times the module is reloaded
            mod.__dict__['loads'] = mod.__dict__.get('loads',0) + 1
            return mod

        def get_data(self, path):
            return "Hello, world!"

    class MyTestImporter(object):
        def find_module(self, fullname, path=None):
            return PkgutilPEP302Tests.MyTestLoader()

    def setUp(self):
        sys.meta_path.insert(0, self.MyTestImporter())

    def tearDown(self):
        del sys.meta_path[0]

    def test_getdata_pep302(self):
        # Use a dummy importer/loader
        self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
        del sys.modules['foo']

    def test_alreadyloaded(self):
        # Ensure that get_data works without reloading - the "loads" module
        # variable in the example loader should count how many times a reload
        # occurs.
        import foo
        self.assertEqual(foo.loads, 1)
        self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
        self.assertEqual(foo.loads, 1)
        del sys.modules['foo']

def test_main():
    run_unittest(PkgutilTests, PkgutilPEP302Tests)
    # this is necessary if test is run repeated (like when finding leaks)
    import zipimport
    zipimport._zip_directory_cache.clear()

if __name__ == '__main__':
    test_main()
n class="hl num">0 < size); fd = open("/dev/urandom", O_RDONLY); if (fd < 0) Py_FatalError("Failed to open /dev/urandom"); while (0 < size) { do { n = read(fd, buffer, (size_t)size); } while (n < 0 && errno == EINTR); if (n <= 0) { /* stop on error or if read(size) returned 0 */ Py_FatalError("Failed to read bytes from /dev/urandom"); break; } buffer += n; size -= (Py_ssize_t)n; } close(fd); } /* Read size bytes from /dev/urandom into buffer. Return 0 on success, raise an exception and return -1 on error. */ static int dev_urandom_python(char *buffer, Py_ssize_t size) { int fd; Py_ssize_t n; if (size <= 0) return 0; Py_BEGIN_ALLOW_THREADS fd = open("/dev/urandom", O_RDONLY); Py_END_ALLOW_THREADS if (fd < 0) { PyErr_SetFromErrnoWithFilename(PyExc_OSError, "/dev/urandom"); return -1; } Py_BEGIN_ALLOW_THREADS do { do { n = read(fd, buffer, (size_t)size); } while (n < 0 && errno == EINTR); if (n <= 0) break; buffer += n; size -= (Py_ssize_t)n; } while (0 < size); Py_END_ALLOW_THREADS if (n <= 0) { /* stop on error or if read(size) returned 0 */ if (n < 0) PyErr_SetFromErrno(PyExc_OSError); else PyErr_Format(PyExc_RuntimeError, "Failed to read %zi bytes from /dev/urandom", size); close(fd); return -1; } close(fd); return 0; } #endif /* !defined(MS_WINDOWS) && !defined(__VMS) */ /* Fill buffer with pseudo-random bytes generated by a linear congruent generator (LCG): x(n+1) = (x(n) * 214013 + 2531011) % 2^32 Use bits 23..16 of x(n) to generate a byte. */ static void lcg_urandom(unsigned int x0, unsigned char *buffer, size_t size) { size_t index; unsigned int x; x = x0; for (index=0; index < size; index++) { x *= 214013; x += 2531011; /* modulo 2 ^ (8 * sizeof(int)) */ buffer[index] = (x >> 16) & 0xff; } } /* Fill buffer with size pseudo-random bytes, not suitable for cryptographic use, from the operating random number generator (RNG). Return 0 on success, raise an exception and return -1 on error. */ int _PyOS_URandom(void *buffer, Py_ssize_t size) { if (size < 0) { PyErr_Format(PyExc_ValueError, "negative argument not allowed"); return -1; } if (size == 0) return 0; #ifdef MS_WINDOWS return win32_urandom((unsigned char *)buffer, size, 1); #else # ifdef __VMS return vms_urandom((unsigned char *)buffer, size, 1); # else return dev_urandom_python((char*)buffer, size); # endif #endif } void _PyRandom_Init(void) { char *env; void *secret = &_Py_HashSecret; Py_ssize_t secret_size = sizeof(_Py_HashSecret_t); if (_Py_HashSecret_Initialized) return; _Py_HashSecret_Initialized = 1; /* By default, hash randomization is disabled, and only enabled if PYTHONHASHSEED is set to non-empty or if "-R" is provided at the command line: */ if (!Py_HashRandomizationFlag) { /* Disable the randomized hash: */ memset(secret, 0, secret_size); return; } /* Hash randomization is enabled. Generate a per-process secret, using PYTHONHASHSEED if provided. */ env = Py_GETENV("PYTHONHASHSEED"); if (env && *env != '\0' && strcmp(env, "random") != 0) { char *endptr = env; unsigned long seed; seed = strtoul(env, &endptr, 10); if (*endptr != '\0' || seed > 4294967295UL || (errno == ERANGE && seed == ULONG_MAX)) { Py_FatalError("PYTHONHASHSEED must be \"random\" or an integer " "in range [0; 4294967295]"); } if (seed == 0) { /* disable the randomized hash */ memset(secret, 0, secret_size); } else { lcg_urandom(seed, (unsigned char*)secret, secret_size); } } else { #ifdef MS_WINDOWS (void)win32_urandom((unsigned char *)secret, secret_size, 0); #else /* #ifdef MS_WINDOWS */ # ifdef __VMS vms_urandom((unsigned char *)secret, secret_size, 0); # else dev_urandom_noraise((char*)secret, secret_size); # endif #endif } }