codekingpro/portable-devtools

1"""
2Utility functions for
3
4- building and importing modules on test time, using a temporary location
5- detecting if compilers are present
6- determining paths to tests
7
8"""
9import atexit
10import concurrent.futures
11import contextlib
12import glob
13import os
14import shutil
15import subprocess
16import sys
17import tempfile
18from importlib import import_module
19from pathlib import Path
20
21import pytest
22
23import numpy
24from numpy._utils import asunicode
25from numpy.f2py._backends._meson import MesonBackend
26from numpy.testing import IS_WASM, temppath
27
28#
29# Check if compilers are available at all...
30#
31
32def check_language(lang, code_snippet=None):
33    if sys.platform == "win32":
34        pytest.skip("No Fortran tests on Windows (Issue #25134)", allow_module_level=True)
35    tmpdir = tempfile.mkdtemp()
36    try:
37        meson_file = os.path.join(tmpdir, "meson.build")
38        with open(meson_file, "w") as f:
39            f.write("project('check_compilers')\n")
40            f.write(f"add_languages('{lang}')\n")
41            if code_snippet:
42                f.write(f"{lang}_compiler = meson.get_compiler('{lang}')\n")
43                f.write(f"{lang}_code = '''{code_snippet}'''\n")
44                f.write(
45                    f"_have_{lang}_feature ="
46                    f"{lang}_compiler.compiles({lang}_code,"
47                    f" name: '{lang} feature check')\n"
48                )
49        try:
50            runmeson = subprocess.run(
51                ["meson", "setup", "btmp"],
52                check=False,
53                cwd=tmpdir,
54                capture_output=True,
55            )
56        except subprocess.CalledProcessError:
57            pytest.skip("meson not present, skipping compiler dependent test", allow_module_level=True)
58        return runmeson.returncode == 0
59    finally:
60        shutil.rmtree(tmpdir)
61
62
63fortran77_code = '''
64C Example Fortran 77 code
65      PROGRAM HELLO
66      PRINT *, 'Hello, Fortran 77!'
67      END
68'''
69
70fortran90_code = '''
71! Example Fortran 90 code
72program hello90
73  type :: greeting
74    character(len=20) :: text
75  end type greeting
76
77  type(greeting) :: greet
78  greet%text = 'hello, fortran 90!'
79  print *, greet%text
80end program hello90
81'''
82
83# Dummy class for caching relevant checks
84class CompilerChecker:
85    def __init__(self):
86        self.compilers_checked = False
87        self.has_c = False
88        self.has_f77 = False
89        self.has_f90 = False
90
91    def check_compilers(self):
92        if (not self.compilers_checked) and (not sys.platform == "cygwin"):
93            with concurrent.futures.ThreadPoolExecutor() as executor:
94                futures = [
95                    executor.submit(check_language, "c"),
96                    executor.submit(check_language, "fortran", fortran77_code),
97                    executor.submit(check_language, "fortran", fortran90_code)
98                ]
99
100                self.has_c = futures[0].result()
101                self.has_f77 = futures[1].result()
102                self.has_f90 = futures[2].result()
103
104            self.compilers_checked = True
105
106
107if not IS_WASM:
108    checker = CompilerChecker()
109    checker.check_compilers()
110
111def has_c_compiler():
112    return checker.has_c
113
114def has_f77_compiler():
115    return checker.has_f77
116
117def has_f90_compiler():
118    return checker.has_f90
119
120def has_fortran_compiler():
121    return (checker.has_f90 and checker.has_f77)
122
123
124#
125# Maintaining a temporary module directory
126#
127
128_module_dir = None
129_module_num = 5403
130
131if sys.platform == "cygwin":
132    NUMPY_INSTALL_ROOT = Path(__file__).parent.parent.parent
133    _module_list = list(NUMPY_INSTALL_ROOT.glob("**/*.dll"))
134
135
136def _cleanup():
137    global _module_dir
138    if _module_dir is not None:
139        try:
140            sys.path.remove(_module_dir)
141        except ValueError:
142            pass
143        try:
144            shutil.rmtree(_module_dir)
145        except OSError:
146            pass
147        _module_dir = None
148
149
150def get_module_dir():
151    global _module_dir
152    if _module_dir is None:
153        _module_dir = tempfile.mkdtemp()
154        atexit.register(_cleanup)
155        if _module_dir not in sys.path:
156            sys.path.insert(0, _module_dir)
157    return _module_dir
158
159
160def get_temp_module_name():
161    # Assume single-threaded, and the module dir usable only by this thread
162    global _module_num
163    get_module_dir()
164    name = "_test_ext_module_%d" % _module_num
165    _module_num += 1
166    if name in sys.modules:
167        # this should not be possible, but check anyway
168        raise RuntimeError("Temporary module name already in use.")
169    return name
170
171
172def _memoize(func):
173    memo = {}
174
175    def wrapper(*a, **kw):
176        key = repr((a, kw))
177        if key not in memo:
178            try:
179                memo[key] = func(*a, **kw)
180            except Exception as e:
181                memo[key] = e
182                raise
183        ret = memo[key]
184        if isinstance(ret, Exception):
185            raise ret
186        return ret
187
188    wrapper.__name__ = func.__name__
189    return wrapper
190
191
192#
193# Building modules
194#
195
196
197@_memoize
198def build_module(source_files, options=[], skip=[], only=[], module_name=None):
199    """
200    Compile and import a f2py module, built from the given files.
201
202    """
203
204    code = f"import sys; sys.path = {sys.path!r}; import numpy.f2py; numpy.f2py.main()"
205
206    d = get_module_dir()
207    # gh-27045 : Skip if no compilers are found
208    if not has_fortran_compiler():
209        pytest.skip("No Fortran compiler available")
210
211    # Copy files
212    dst_sources = []
213    f2py_sources = []
214    for fn in source_files:
215        if not os.path.isfile(fn):
216            raise RuntimeError(f"{fn} is not a file")
217        dst = os.path.join(d, os.path.basename(fn))
218        shutil.copyfile(fn, dst)
219        dst_sources.append(dst)
220
221        base, ext = os.path.splitext(dst)
222        if ext in (".f90", ".f95", ".f", ".c", ".pyf"):
223            f2py_sources.append(dst)
224
225    assert f2py_sources
226
227    # Prepare options
228    if module_name is None:
229        module_name = get_temp_module_name()
230    gil_options = []
231    if '--freethreading-compatible' not in options and '--no-freethreading-compatible' not in options:
232        # default to disabling the GIL if unset in options
233        gil_options = ['--freethreading-compatible']
234    f2py_opts = ["-c", "-m", module_name] + options + gil_options + f2py_sources
235    f2py_opts += ["--backend", "meson"]
236    if skip:
237        f2py_opts += ["skip:"] + skip
238    if only:
239        f2py_opts += ["only:"] + only
240
241    # Build
242    cwd = os.getcwd()
243    try:
244        os.chdir(d)
245        cmd = [sys.executable, "-c", code] + f2py_opts
246        p = subprocess.Popen(cmd,
247                             stdout=subprocess.PIPE,
248                             stderr=subprocess.STDOUT)
249        out, err = p.communicate()
250        if p.returncode != 0:
251            raise RuntimeError(f"Running f2py failed: {cmd[4:]}\n{asunicode(out)}")
252    finally:
253        os.chdir(cwd)
254
255        # Partial cleanup
256        for fn in dst_sources:
257            os.unlink(fn)
258
259    # Rebase (Cygwin-only)
260    if sys.platform == "cygwin":
261        # If someone starts deleting modules after import, this will
262        # need to change to record how big each module is, rather than
263        # relying on rebase being able to find that from the files.
264        _module_list.extend(
265            glob.glob(os.path.join(d, f"{module_name:s}*"))
266        )
267        subprocess.check_call(
268            ["/usr/bin/rebase", "--database", "--oblivious", "--verbose"]
269            + _module_list
270        )
271
272    # Import
273    return import_module(module_name)
274
275
276@_memoize
277def build_code(source_code,
278               options=[],
279               skip=[],
280               only=[],
281               suffix=None,
282               module_name=None):
283    """
284    Compile and import Fortran code using f2py.
285
286    """
287    if suffix is None:
288        suffix = ".f"
289    with temppath(suffix=suffix) as path:
290        with open(path, "w") as f:
291            f.write(source_code)
292        return build_module([path],
293                            options=options,
294                            skip=skip,
295                            only=only,
296                            module_name=module_name)
297
298
299#
300# Building with meson
301#
302
303
304class SimplifiedMesonBackend(MesonBackend):
305    def __init__(self, *args, **kwargs):
306        super().__init__(*args, **kwargs)
307
308    def compile(self):
309        self.write_meson_build(self.build_dir)
310        self.run_meson(self.build_dir)
311
312
313def build_meson(source_files, module_name=None, **kwargs):
314    """
315    Build a module via Meson and import it.
316    """
317
318    # gh-27045 : Skip if no compilers are found
319    if not has_fortran_compiler():
320        pytest.skip("No Fortran compiler available")
321
322    build_dir = get_module_dir()
323    if module_name is None:
324        module_name = get_temp_module_name()
325
326    # Initialize the MesonBackend
327    backend = SimplifiedMesonBackend(
328        modulename=module_name,
329        sources=source_files,
330        extra_objects=kwargs.get("extra_objects", []),
331        build_dir=build_dir,
332        include_dirs=kwargs.get("include_dirs", []),
333        library_dirs=kwargs.get("library_dirs", []),
334        libraries=kwargs.get("libraries", []),
335        define_macros=kwargs.get("define_macros", []),
336        undef_macros=kwargs.get("undef_macros", []),
337        f2py_flags=kwargs.get("f2py_flags", []),
338        sysinfo_flags=kwargs.get("sysinfo_flags", []),
339        fc_flags=kwargs.get("fc_flags", []),
340        flib_flags=kwargs.get("flib_flags", []),
341        setup_flags=kwargs.get("setup_flags", []),
342        remove_build_dir=kwargs.get("remove_build_dir", False),
343        extra_dat=kwargs.get("extra_dat", {}),
344    )
345
346    backend.compile()
347
348    # Import the compiled module
349    sys.path.insert(0, f"{build_dir}/{backend.meson_build_dir}")
350    return import_module(module_name)
351
352
353#
354# Unittest convenience
355#
356
357
358class F2PyTest:
359    code = None
360    sources = None
361    options = []
362    skip = []
363    only = []
364    suffix = ".f"
365    module = None
366    _has_c_compiler = None
367    _has_f77_compiler = None
368    _has_f90_compiler = None
369
370    @property
371    def module_name(self):
372        cls = type(self)
373        return f'_{cls.__module__.rsplit(".", 1)[-1]}_{cls.__name__}_ext_module'
374
375    @classmethod
376    def setup_class(cls):
377        if sys.platform == "win32":
378            pytest.skip("Fails with MinGW64 Gfortran (Issue #9673)")
379        F2PyTest._has_c_compiler = has_c_compiler()
380        F2PyTest._has_f77_compiler = has_f77_compiler()
381        F2PyTest._has_f90_compiler = has_f90_compiler()
382        F2PyTest._has_fortran_compiler = has_fortran_compiler()
383
384    def setup_method(self):
385        if self.module is not None:
386            return
387
388        codes = self.sources or []
389        if self.code:
390            codes.append(self.suffix)
391
392        needs_f77 = any(str(fn).endswith(".f") for fn in codes)
393        needs_f90 = any(str(fn).endswith(".f90") for fn in codes)
394        needs_pyf = any(str(fn).endswith(".pyf") for fn in codes)
395
396        if needs_f77 and not self._has_f77_compiler:
397            pytest.skip("No Fortran 77 compiler available")
398        if needs_f90 and not self._has_f90_compiler:
399            pytest.skip("No Fortran 90 compiler available")
400        if needs_pyf and not self._has_fortran_compiler:
401            pytest.skip("No Fortran compiler available")
402
403        # Build the module
404        if self.code is not None:
405            self.module = build_code(
406                self.code,
407                options=self.options,
408                skip=self.skip,
409                only=self.only,
410                suffix=self.suffix,
411                module_name=self.module_name,
412            )
413
414        if self.sources is not None:
415            self.module = build_module(
416                self.sources,
417                options=self.options,
418                skip=self.skip,
419                only=self.only,
420                module_name=self.module_name,
421            )
422
423
424#
425# Helper functions
426#
427
428
429def getpath(*a):
430    # Package root
431    d = Path(numpy.f2py.__file__).parent.resolve()
432    return d.joinpath(*a)
433
434
435@contextlib.contextmanager
436def switchdir(path):
437    curpath = Path.cwd()
438    os.chdir(path)
439    try:
440        yield
441    finally:
442        os.chdir(curpath)
443