• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Run the tests in Programs/_testembed.c (tests for the CPython embedding APIs)
2from test import support
3import unittest
4
5from collections import namedtuple
6import json
7import os
8import re
9import subprocess
10import sys
11import textwrap
12
13
14MS_WINDOWS = (os.name == 'nt')
15
16
17class EmbeddingTestsMixin:
18    def setUp(self):
19        here = os.path.abspath(__file__)
20        basepath = os.path.dirname(os.path.dirname(os.path.dirname(here)))
21        exename = "_testembed"
22        if MS_WINDOWS:
23            ext = ("_d" if "_d" in sys.executable else "") + ".exe"
24            exename += ext
25            exepath = os.path.dirname(sys.executable)
26        else:
27            exepath = os.path.join(basepath, "Programs")
28        self.test_exe = exe = os.path.join(exepath, exename)
29        if not os.path.exists(exe):
30            self.skipTest("%r doesn't exist" % exe)
31        # This is needed otherwise we get a fatal error:
32        # "Py_Initialize: Unable to get the locale encoding
33        # LookupError: no codec search functions registered: can't find encoding"
34        self.oldcwd = os.getcwd()
35        os.chdir(basepath)
36
37    def tearDown(self):
38        os.chdir(self.oldcwd)
39
40    def run_embedded_interpreter(self, *args, env=None):
41        """Runs a test in the embedded interpreter"""
42        cmd = [self.test_exe]
43        cmd.extend(args)
44        if env is not None and MS_WINDOWS:
45            # Windows requires at least the SYSTEMROOT environment variable to
46            # start Python.
47            env = env.copy()
48            env['SYSTEMROOT'] = os.environ['SYSTEMROOT']
49
50        p = subprocess.Popen(cmd,
51                             stdout=subprocess.PIPE,
52                             stderr=subprocess.PIPE,
53                             universal_newlines=True,
54                             env=env)
55        (out, err) = p.communicate()
56        if p.returncode != 0 and support.verbose:
57            print(f"--- {cmd} failed ---")
58            print(f"stdout:\n{out}")
59            print(f"stderr:\n{err}")
60            print(f"------")
61
62        self.assertEqual(p.returncode, 0,
63                         "bad returncode %d, stderr is %r" %
64                         (p.returncode, err))
65        return out, err
66
67    def run_repeated_init_and_subinterpreters(self):
68        out, err = self.run_embedded_interpreter("repeated_init_and_subinterpreters")
69        self.assertEqual(err, "")
70
71        # The output from _testembed looks like this:
72        # --- Pass 0 ---
73        # interp 0 <0x1cf9330>, thread state <0x1cf9700>: id(modules) = 139650431942728
74        # interp 1 <0x1d4f690>, thread state <0x1d35350>: id(modules) = 139650431165784
75        # interp 2 <0x1d5a690>, thread state <0x1d99ed0>: id(modules) = 139650413140368
76        # interp 3 <0x1d4f690>, thread state <0x1dc3340>: id(modules) = 139650412862200
77        # interp 0 <0x1cf9330>, thread state <0x1cf9700>: id(modules) = 139650431942728
78        # --- Pass 1 ---
79        # ...
80
81        interp_pat = (r"^interp (\d+) <(0x[\dA-F]+)>, "
82                      r"thread state <(0x[\dA-F]+)>: "
83                      r"id\(modules\) = ([\d]+)$")
84        Interp = namedtuple("Interp", "id interp tstate modules")
85
86        numloops = 0
87        current_run = []
88        for line in out.splitlines():
89            if line == "--- Pass {} ---".format(numloops):
90                self.assertEqual(len(current_run), 0)
91                if support.verbose > 1:
92                    print(line)
93                numloops += 1
94                continue
95
96            self.assertLess(len(current_run), 5)
97            match = re.match(interp_pat, line)
98            if match is None:
99                self.assertRegex(line, interp_pat)
100
101            # Parse the line from the loop.  The first line is the main
102            # interpreter and the 3 afterward are subinterpreters.
103            interp = Interp(*match.groups())
104            if support.verbose > 1:
105                print(interp)
106            self.assertTrue(interp.interp)
107            self.assertTrue(interp.tstate)
108            self.assertTrue(interp.modules)
109            current_run.append(interp)
110
111            # The last line in the loop should be the same as the first.
112            if len(current_run) == 5:
113                main = current_run[0]
114                self.assertEqual(interp, main)
115                yield current_run
116                current_run = []
117
118
119class EmbeddingTests(EmbeddingTestsMixin, unittest.TestCase):
120    def test_subinterps_main(self):
121        for run in self.run_repeated_init_and_subinterpreters():
122            main = run[0]
123
124            self.assertEqual(main.id, '0')
125
126    def test_subinterps_different_ids(self):
127        for run in self.run_repeated_init_and_subinterpreters():
128            main, *subs, _ = run
129
130            mainid = int(main.id)
131            for i, sub in enumerate(subs):
132                self.assertEqual(sub.id, str(mainid + i + 1))
133
134    def test_subinterps_distinct_state(self):
135        for run in self.run_repeated_init_and_subinterpreters():
136            main, *subs, _ = run
137
138            if '0x0' in main:
139                # XXX Fix on Windows (and other platforms): something
140                # is going on with the pointers in Programs/_testembed.c.
141                # interp.interp is 0x0 and interp.modules is the same
142                # between interpreters.
143                raise unittest.SkipTest('platform prints pointers as 0x0')
144
145            for sub in subs:
146                # A new subinterpreter may have the same
147                # PyInterpreterState pointer as a previous one if
148                # the earlier one has already been destroyed.  So
149                # we compare with the main interpreter.  The same
150                # applies to tstate.
151                self.assertNotEqual(sub.interp, main.interp)
152                self.assertNotEqual(sub.tstate, main.tstate)
153                self.assertNotEqual(sub.modules, main.modules)
154
155    def test_forced_io_encoding(self):
156        # Checks forced configuration of embedded interpreter IO streams
157        env = dict(os.environ, PYTHONIOENCODING="utf-8:surrogateescape")
158        out, err = self.run_embedded_interpreter("forced_io_encoding", env=env)
159        if support.verbose > 1:
160            print()
161            print(out)
162            print(err)
163        expected_stream_encoding = "utf-8"
164        expected_errors = "surrogateescape"
165        expected_output = '\n'.join([
166        "--- Use defaults ---",
167        "Expected encoding: default",
168        "Expected errors: default",
169        "stdin: {in_encoding}:{errors}",
170        "stdout: {out_encoding}:{errors}",
171        "stderr: {out_encoding}:backslashreplace",
172        "--- Set errors only ---",
173        "Expected encoding: default",
174        "Expected errors: ignore",
175        "stdin: {in_encoding}:ignore",
176        "stdout: {out_encoding}:ignore",
177        "stderr: {out_encoding}:backslashreplace",
178        "--- Set encoding only ---",
179        "Expected encoding: latin-1",
180        "Expected errors: default",
181        "stdin: latin-1:{errors}",
182        "stdout: latin-1:{errors}",
183        "stderr: latin-1:backslashreplace",
184        "--- Set encoding and errors ---",
185        "Expected encoding: latin-1",
186        "Expected errors: replace",
187        "stdin: latin-1:replace",
188        "stdout: latin-1:replace",
189        "stderr: latin-1:backslashreplace"])
190        expected_output = expected_output.format(
191                                in_encoding=expected_stream_encoding,
192                                out_encoding=expected_stream_encoding,
193                                errors=expected_errors)
194        # This is useful if we ever trip over odd platform behaviour
195        self.maxDiff = None
196        self.assertEqual(out.strip(), expected_output)
197
198    def test_pre_initialization_api(self):
199        """
200        Checks some key parts of the C-API that need to work before the runtine
201        is initialized (via Py_Initialize()).
202        """
203        env = dict(os.environ, PYTHONPATH=os.pathsep.join(sys.path))
204        out, err = self.run_embedded_interpreter("pre_initialization_api", env=env)
205        if MS_WINDOWS:
206            expected_path = self.test_exe
207        else:
208            expected_path = os.path.join(os.getcwd(), "spam")
209        expected_output = f"sys.executable: {expected_path}\n"
210        self.assertIn(expected_output, out)
211        self.assertEqual(err, '')
212
213    def test_pre_initialization_sys_options(self):
214        """
215        Checks that sys.warnoptions and sys._xoptions can be set before the
216        runtime is initialized (otherwise they won't be effective).
217        """
218        env = dict(os.environ, PYTHONPATH=os.pathsep.join(sys.path))
219        out, err = self.run_embedded_interpreter(
220                        "pre_initialization_sys_options", env=env)
221        expected_output = (
222            "sys.warnoptions: ['once', 'module', 'default']\n"
223            "sys._xoptions: {'not_an_option': '1', 'also_not_an_option': '2'}\n"
224            "warnings.filters[:3]: ['default', 'module', 'once']\n"
225        )
226        self.assertIn(expected_output, out)
227        self.assertEqual(err, '')
228
229    def test_bpo20891(self):
230        """
231        bpo-20891: Calling PyGILState_Ensure in a non-Python thread before
232        calling PyEval_InitThreads() must not crash. PyGILState_Ensure() must
233        call PyEval_InitThreads() for us in this case.
234        """
235        out, err = self.run_embedded_interpreter("bpo20891")
236        self.assertEqual(out, '')
237        self.assertEqual(err, '')
238
239    def test_initialize_twice(self):
240        """
241        bpo-33932: Calling Py_Initialize() twice should do nothing (and not
242        crash!).
243        """
244        out, err = self.run_embedded_interpreter("initialize_twice")
245        self.assertEqual(out, '')
246        self.assertEqual(err, '')
247
248    def test_initialize_pymain(self):
249        """
250        bpo-34008: Calling Py_Main() after Py_Initialize() must not fail.
251        """
252        out, err = self.run_embedded_interpreter("initialize_pymain")
253        self.assertEqual(out.rstrip(), "Py_Main() after Py_Initialize: sys.argv=['-c', 'arg2']")
254        self.assertEqual(err, '')
255
256
257class InitConfigTests(EmbeddingTestsMixin, unittest.TestCase):
258    maxDiff = 4096
259    UTF8_MODE_ERRORS = ('surrogatepass' if MS_WINDOWS else 'surrogateescape')
260
261    # core config
262    UNTESTED_CORE_CONFIG = (
263        # FIXME: untested core configuration variables
264        'dll_path',
265        'executable',
266        'module_search_paths',
267    )
268    # Mark config which should be get by get_default_config()
269    GET_DEFAULT_CONFIG = object()
270    DEFAULT_CORE_CONFIG = {
271        'install_signal_handlers': 1,
272        'ignore_environment': 0,
273        'use_hash_seed': 0,
274        'hash_seed': 0,
275        'allocator': None,
276        'dev_mode': 0,
277        'faulthandler': 0,
278        'tracemalloc': 0,
279        'import_time': 0,
280        'show_ref_count': 0,
281        'show_alloc_count': 0,
282        'dump_refs': 0,
283        'malloc_stats': 0,
284
285        'utf8_mode': 0,
286        'coerce_c_locale': 0,
287        'coerce_c_locale_warn': 0,
288
289        'program_name': './_testembed',
290        'argv': [],
291        'program': None,
292
293        'xoptions': [],
294        'warnoptions': [],
295
296        'module_search_path_env': None,
297        'home': None,
298
299        'prefix': GET_DEFAULT_CONFIG,
300        'base_prefix': GET_DEFAULT_CONFIG,
301        'exec_prefix': GET_DEFAULT_CONFIG,
302        'base_exec_prefix': GET_DEFAULT_CONFIG,
303
304        '_disable_importlib': 0,
305    }
306
307    # main config
308    UNTESTED_MAIN_CONFIG = (
309        # FIXME: untested main configuration variables
310        'module_search_path',
311    )
312    COPY_MAIN_CONFIG = (
313        # Copy core config to main config for expected values
314        'argv',
315        'base_exec_prefix',
316        'base_prefix',
317        'exec_prefix',
318        'executable',
319        'install_signal_handlers',
320        'prefix',
321        'warnoptions',
322        # xoptions is created from core_config in check_main_config()
323    )
324
325    # global config
326    UNTESTED_GLOBAL_CONFIG = (
327        # Py_HasFileSystemDefaultEncoding value depends on the LC_CTYPE locale
328        # and the platform. It is complex to test it, and it's value doesn't
329        # really matter.
330        'Py_HasFileSystemDefaultEncoding',
331    )
332    DEFAULT_GLOBAL_CONFIG = {
333        'Py_BytesWarningFlag': 0,
334        'Py_DebugFlag': 0,
335        'Py_DontWriteBytecodeFlag': 0,
336        'Py_FileSystemDefaultEncodeErrors': GET_DEFAULT_CONFIG,
337        'Py_FileSystemDefaultEncoding': GET_DEFAULT_CONFIG,
338        'Py_FrozenFlag': 0,
339        'Py_HashRandomizationFlag': 1,
340        'Py_InspectFlag': 0,
341        'Py_InteractiveFlag': 0,
342        'Py_IsolatedFlag': 0,
343        'Py_NoSiteFlag': 0,
344        'Py_NoUserSiteDirectory': 0,
345        'Py_OptimizeFlag': 0,
346        'Py_QuietFlag': 0,
347        'Py_UnbufferedStdioFlag': 0,
348        'Py_VerboseFlag': 0,
349    }
350    if MS_WINDOWS:
351        DEFAULT_GLOBAL_CONFIG.update({
352            'Py_LegacyWindowsFSEncodingFlag': 0,
353            'Py_LegacyWindowsStdioFlag': 0,
354        })
355    COPY_GLOBAL_CONFIG = [
356        # Copy core config to global config for expected values
357        # True means that the core config value is inverted (0 => 1 and 1 => 0)
358        ('Py_IgnoreEnvironmentFlag', 'ignore_environment'),
359        ('Py_UTF8Mode', 'utf8_mode'),
360    ]
361
362    def main_xoptions(self, xoptions_list):
363        xoptions = {}
364        for opt in xoptions_list:
365            if '=' in opt:
366                key, value = opt.split('=', 1)
367                xoptions[key] = value
368            else:
369                xoptions[opt] = True
370        return xoptions
371
372    def check_main_config(self, config):
373        core_config = config['core_config']
374        main_config = config['main_config']
375
376        # main config
377        for key in self.UNTESTED_MAIN_CONFIG:
378            del main_config[key]
379
380        expected = {}
381        for key in self.COPY_MAIN_CONFIG:
382            expected[key] = core_config[key]
383        expected['xoptions'] = self.main_xoptions(core_config['xoptions'])
384        self.assertEqual(main_config, expected)
385
386    def get_expected_config(self, expected_core, expected_global, env):
387        expected_core = dict(self.DEFAULT_CORE_CONFIG, **expected_core)
388        expected_global = dict(self.DEFAULT_GLOBAL_CONFIG, **expected_global)
389
390        code = textwrap.dedent('''
391            import json
392            import sys
393
394            data = {
395                'prefix': sys.prefix,
396                'base_prefix': sys.base_prefix,
397                'exec_prefix': sys.exec_prefix,
398                'base_exec_prefix': sys.base_exec_prefix,
399                'Py_FileSystemDefaultEncoding': sys.getfilesystemencoding(),
400                'Py_FileSystemDefaultEncodeErrors': sys.getfilesystemencodeerrors(),
401            }
402
403            data = json.dumps(data)
404            data = data.encode('utf-8')
405            sys.stdout.buffer.write(data)
406            sys.stdout.buffer.flush()
407        ''')
408
409        # Use -S to not import the site module: get the proper configuration
410        # when test_embed is run from a venv (bpo-35313)
411        args = (sys.executable, '-S', '-c', code)
412        env = dict(env)
413        if not expected_global['Py_IsolatedFlag']:
414            env['PYTHONCOERCECLOCALE'] = '0'
415            env['PYTHONUTF8'] = '0'
416        proc = subprocess.run(args, env=env,
417                              stdout=subprocess.PIPE,
418                              stderr=subprocess.STDOUT)
419        if proc.returncode:
420            raise Exception(f"failed to get the default config: "
421                            f"stdout={proc.stdout!r} stderr={proc.stderr!r}")
422        stdout = proc.stdout.decode('utf-8')
423        config = json.loads(stdout)
424
425        for key, value in expected_core.items():
426            if value is self.GET_DEFAULT_CONFIG:
427                expected_core[key] = config[key]
428        for key, value in expected_global.items():
429            if value is self.GET_DEFAULT_CONFIG:
430                expected_global[key] = config[key]
431        return (expected_core, expected_global)
432
433    def check_core_config(self, config, expected):
434        core_config = dict(config['core_config'])
435        for key in self.UNTESTED_CORE_CONFIG:
436            core_config.pop(key, None)
437        self.assertEqual(core_config, expected)
438
439    def check_global_config(self, config, expected, env):
440        core_config = config['core_config']
441
442        for item in self.COPY_GLOBAL_CONFIG:
443            if len(item) == 3:
444                global_key, core_key, opposite = item
445                expected[global_key] = 0 if core_config[core_key] else 1
446            else:
447                global_key, core_key = item
448                expected[global_key] = core_config[core_key]
449
450        global_config = dict(config['global_config'])
451        for key in self.UNTESTED_GLOBAL_CONFIG:
452            del global_config[key]
453        self.assertEqual(global_config, expected)
454
455    def check_config(self, testname, expected_core, expected_global):
456        env = dict(os.environ)
457        # Remove PYTHON* environment variables to get deterministic environment
458        for key in list(env):
459            if key.startswith('PYTHON'):
460                del env[key]
461        # Disable C locale coercion and UTF-8 mode to not depend
462        # on the current locale
463        env['PYTHONCOERCECLOCALE'] = '0'
464        env['PYTHONUTF8'] = '0'
465
466        out, err = self.run_embedded_interpreter(testname, env=env)
467        # Ignore err
468        config = json.loads(out)
469
470        expected_core, expected_global = self.get_expected_config(expected_core, expected_global, env)
471        self.check_core_config(config, expected_core)
472        self.check_main_config(config)
473        self.check_global_config(config, expected_global, env)
474
475    def test_init_default_config(self):
476        self.check_config("init_default_config", {}, {})
477
478    def test_init_global_config(self):
479        core_config = {
480            'program_name': './globalvar',
481            'utf8_mode': 1,
482        }
483        global_config = {
484            'Py_BytesWarningFlag': 1,
485            'Py_DontWriteBytecodeFlag': 1,
486            'Py_FileSystemDefaultEncodeErrors': self.UTF8_MODE_ERRORS,
487            'Py_FileSystemDefaultEncoding': 'utf-8',
488            'Py_InspectFlag': 1,
489            'Py_InteractiveFlag': 1,
490            'Py_NoSiteFlag': 1,
491            'Py_NoUserSiteDirectory': 1,
492            'Py_OptimizeFlag': 2,
493            'Py_QuietFlag': 1,
494            'Py_VerboseFlag': 1,
495            'Py_FrozenFlag': 1,
496            'Py_UnbufferedStdioFlag': 1,
497        }
498        self.check_config("init_global_config", core_config, global_config)
499
500    def test_init_from_config(self):
501        core_config = {
502            'install_signal_handlers': 0,
503            'use_hash_seed': 1,
504            'hash_seed': 123,
505            'allocator': 'malloc_debug',
506            'tracemalloc': 2,
507            'import_time': 1,
508            'show_ref_count': 1,
509            'show_alloc_count': 1,
510            'malloc_stats': 1,
511
512            'utf8_mode': 1,
513
514            'program_name': './conf_program_name',
515            'argv': ['-c', 'pass'],
516            'program': 'conf_program',
517            'xoptions': ['core_xoption1=3', 'core_xoption2=', 'core_xoption3'],
518            'warnoptions': ['default', 'error::ResourceWarning'],
519
520            'faulthandler': 1,
521        }
522        global_config = {
523            'Py_FileSystemDefaultEncodeErrors': self.UTF8_MODE_ERRORS,
524            'Py_FileSystemDefaultEncoding': 'utf-8',
525            'Py_NoUserSiteDirectory': 0,
526        }
527        self.check_config("init_from_config", core_config, global_config)
528
529    def test_init_env(self):
530        core_config = {
531            'use_hash_seed': 1,
532            'hash_seed': 42,
533            'allocator': 'malloc_debug',
534            'tracemalloc': 2,
535            'import_time': 1,
536            'malloc_stats': 1,
537            'utf8_mode': 1,
538            'faulthandler': 1,
539            'dev_mode': 1,
540        }
541        global_config = {
542            'Py_DontWriteBytecodeFlag': 1,
543            'Py_FileSystemDefaultEncodeErrors': self.UTF8_MODE_ERRORS,
544            'Py_FileSystemDefaultEncoding': 'utf-8',
545            'Py_InspectFlag': 1,
546            'Py_NoUserSiteDirectory': 1,
547            'Py_OptimizeFlag': 2,
548            'Py_UnbufferedStdioFlag': 1,
549            'Py_VerboseFlag': 1,
550        }
551        self.check_config("init_env", core_config, global_config)
552
553    def test_init_dev_mode(self):
554        core_config = {
555            'dev_mode': 1,
556            'faulthandler': 1,
557            'allocator': 'debug',
558        }
559        self.check_config("init_dev_mode", core_config, {})
560
561    def test_init_isolated(self):
562        core_config = {
563            'ignore_environment': 1,
564        }
565        global_config = {
566            'Py_IsolatedFlag': 1,
567            'Py_NoUserSiteDirectory': 1,
568        }
569        self.check_config("init_isolated", core_config, global_config)
570
571
572if __name__ == "__main__":
573    unittest.main()
574