14203589112

Committed 01 Apr 2025 06:45PM UTC coverage: 96.834% (-0.007%) from 96.841%

Build # 14203589112

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #84 from nbiotcloud/fix/check-strict

let 'ucdp check' fail with traceback on error

Run Details

7 of 7 new or added lines in 1 file covered. (100.0%)

3 existing lines in 2 files now uncovered.

4772 of 4928 relevant lines covered (96.83%)

7.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.46

/src/ucdp/_modloader.py

#
# MIT License
#
# Copyright (c) 2024-2025 nbiotcloud
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#

"""
Loading And Searching Facility.
"""

import re
import sys
from collections.abc import Iterable, Iterator
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
from importlib import import_module
from inspect import getfile, isclass
from pathlib import Path
from typing import TypeAlias

from .cache import CACHE
from .consts import PKG_PATHS
from .modbase import BaseMod, get_modbaseclss
from .modref import ModRef, get_modclsname
from .modtopref import TopModRef
from .object import Object
from .util import LOGGER, get_maxworkers, guess_path

Patterns: TypeAlias = Iterable[str]
Paths: TypeAlias = Iterable[Path]

_RE_IMPORT_UCDP = re.compile(r"^\s*class .*Mod\):")

_RE_TOPMODREFPAT = re.compile(
    # [tb]#
    r"((?P<tb>[a-zA-Z_0-9_\.\*]+)#)?"
    # top
    r"(?P<top>[a-zA-Z_0-9_\.\*]+)"
    # [-sub]
    r"(-(?P<sub>[a-zA-Z_0-9_\.\*]+))?"
)


class TopModRefPat(Object):
    """Top Module Reference Search Pattern Pattern."""

    top: str
    sub: str | None = None
    tb: str | None = None

    def __str__(self):
        result = self.top
        if self.sub:
            result = f"{result}-{self.sub}"
        if self.tb:
            result = f"{self.tb}#{result}"
        return result


@lru_cache
def build_top(modcls, **kwargs):
    """Build Top Module."""
    return modcls.build_top(**kwargs)


@lru_cache
def load_modcls(modref: ModRef) -> type[BaseMod]:
    """Load Module Class."""
    name = f"{modref.libname}.{modref.modname}"
    try:
        pymod = import_module(name)
    except ModuleNotFoundError as exc:
        if exc.name in (modref.libname, name):
            raise NameError(f"{name!r} not found.") from None
        raise exc
    modclsname = modref.get_modclsname()
    modcls = getattr(pymod, modclsname, None)
    if not modcls:
        raise NameError(f"{name!r} does not contain {modclsname}.") from None
    if not issubclass(modcls, BaseMod):
        raise ValueError(f"{modcls} is not a module aka child of <class ucdp.BaseMod>.")
    return modcls


def find_modrefs(local: bool | None = None) -> tuple[ModRef, ...]:
    # determine directories with python files
    dirpaths: set[Path] = set()
    modrefs: list[ModRef] = []
    for syspathstr in sys.path:
        syspath = Path(syspathstr).resolve()
        if local is not None and local is any(syspath.is_relative_to(pkg_path) for pkg_path in PKG_PATHS):
            continue
        for filepath in syspath.glob("*/*.py"):
            dirpath = filepath.parent
            if dirpath.name.startswith("_") or dirpath.name == "ucdp":
                continue
            dirpaths.add(dirpath)

    _find_modrefs_cached = CACHE.loader_cache.anycache(depfilefunc=_find_modrefs_files)(_find_modrefs)

    maxworkers = get_maxworkers()
    with ThreadPoolExecutor(max_workers=maxworkers) as exe:
        # start
        jobs = [
            exe.submit(_find_modrefs_cached, sys.path, tuple(sorted(dirpath.glob("*.py"))))
            for dirpath in sorted(dirpaths)
        ]
        # collect
        for job in jobs:
            modrefs.extend(job.result())
    return tuple(modrefs)


def _find_modrefs_files(modrefs: tuple[ModRef, ...], envpath: list[str], filepaths: tuple[Path, ...]) -> list[Path]:
    paths: set[Path] = set(filepaths)
    for modref in modrefs:
        modcls = load_modcls(modref)
        for basecls in get_modbaseclss(modcls):
            paths.add(Path(getfile(basecls)))
    return sorted(paths)


def _find_modrefs(envpath: list[str], filepaths: tuple[Path, ...]) -> tuple[ModRef, ...]:  # noqa: C901
    modrefs = []
    for filepath in filepaths:
        pylibname = filepath.parent.name
        pymodname = filepath.stem
        if pymodname.startswith("_"):
            continue
        # skip non-ucdp files
        with filepath.open(encoding="utf-8") as file:
            try:
                for line in file:
                    if _RE_IMPORT_UCDP.match(line):
                        break
                else:
                    continue
            except Exception as exc:
                LOGGER.info(f"Skipping {str(filepath)!r} ({exc})")
                continue

        # import module
        try:
            pymod = import_module(f"{pylibname}.{pymodname}")
        except Exception as exc:
            LOGGER.warning(f"Skipping {str(filepath)!r} ({exc})")
            continue

        # Inspect Module
        for name in dir(pymod):
            # Load Class
            modcls = getattr(pymod, name)
            if not isclass(modcls) or not issubclass(modcls, BaseMod):
                continue

            # Ignore imported
            if filepath != Path(getfile(modcls)):
                continue

            # Create ModRefInfo
            modclsname = get_modclsname(pymodname)
            if modclsname == name:
                modref = ModRef(libname=pylibname, modname=pymodname)
            else:
                modref = ModRef(libname=pylibname, modname=pymodname, modclsname=name)
            modrefs.append(modref)

    return tuple(modrefs)


def get_topmodrefpats(patterns: Patterns | None) -> Iterator[TopModRefPat | TopModRef]:
    for pattern in patterns or []:
        path = guess_path(pattern)
        if path:
            yield TopModRef.cast(path)
        else:
            mat = _RE_TOPMODREFPAT.fullmatch(pattern)
            if mat:
                yield TopModRefPat(**mat.groupdict())
            else:
                yield TopModRefPat(top=".")  # never matching

1	#
2	# MIT License
3	#
4	# Copyright (c) 2024-2025 nbiotcloud
5	#
6	# Permission is hereby granted, free of charge, to any person obtaining a copy
7	# of this software and associated documentation files (the "Software"), to deal
8	# in the Software without restriction, including without limitation the rights
9	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10	# copies of the Software, and to permit persons to whom the Software is
11	# furnished to do so, subject to the following conditions:
12	#
13	# The above copyright notice and this permission notice shall be included in all
14	# copies or substantial portions of the Software.
15	#
16	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22	# SOFTWARE.
23	#
24
25	"""
26	Loading And Searching Facility.
27	"""
28
29	import re	8✔
30	import sys	8✔
31	from collections.abc import Iterable, Iterator	8✔
32	from concurrent.futures import ThreadPoolExecutor	8✔
33	from functools import lru_cache	8✔
34	from importlib import import_module	8✔
35	from inspect import getfile, isclass	8✔
36	from pathlib import Path	8✔
37	from typing import TypeAlias	8✔
38
39	from .cache import CACHE	8✔
40	from .consts import PKG_PATHS	8✔
41	from .modbase import BaseMod, get_modbaseclss	8✔
42	from .modref import ModRef, get_modclsname	8✔
43	from .modtopref import TopModRef	8✔
44	from .object import Object	8✔
45	from .util import LOGGER, get_maxworkers, guess_path	8✔
46
47	Patterns: TypeAlias = Iterable[str]	8✔
48	Paths: TypeAlias = Iterable[Path]	8✔
49
50	_RE_IMPORT_UCDP = re.compile(r"^\sclass .Mod\):")	8✔
51
52	_RE_TOPMODREFPAT = re.compile(	8✔
53	# [tb]#
54	r"((?P<tb>[a-zA-Z_0-9_\.\*]+)#)?"
55	# top
56	r"(?P<top>[a-zA-Z_0-9_\.\*]+)"
57	# [-sub]
58	r"(-(?P<sub>[a-zA-Z_0-9_\.\*]+))?"
59	)
60
61
62	class TopModRefPat(Object):	8✔
63	"""Top Module Reference Search Pattern Pattern."""
64
65	top: str	8✔
66	sub: str \| None = None	8✔
67	tb: str \| None = None	8✔
68
69	def __str__(self):	8✔
70	result = self.top	8✔
71	if self.sub:	8✔
72	result = f"{result}-{self.sub}"	×
73	if self.tb:	8✔
74	result = f"{self.tb}#{result}"	×
75	return result	8✔
76
77
78	@lru_cache	8✔
79	def build_top(modcls, **kwargs):	8✔
80	"""Build Top Module."""
81	return modcls.build_top(**kwargs)	8✔
82
83
84	@lru_cache	8✔
85	def load_modcls(modref: ModRef) -> type[BaseMod]:	8✔
86	"""Load Module Class."""
87	name = f"{modref.libname}.{modref.modname}"	8✔
88	try:	8✔
89	pymod = import_module(name)	8✔
90	except ModuleNotFoundError as exc:	8✔
91	if exc.name in (modref.libname, name):	8✔
92	raise NameError(f"{name!r} not found.") from None	8✔
93	raise exc	8✔
94	modclsname = modref.get_modclsname()	8✔
95	modcls = getattr(pymod, modclsname, None)	8✔
96	if not modcls:	8✔
97	raise NameError(f"{name!r} does not contain {modclsname}.") from None	8✔
98	if not issubclass(modcls, BaseMod):	8✔
99	raise ValueError(f"{modcls} is not a module aka child of <class ucdp.BaseMod>.")	8✔
100	return modcls	8✔
101
102
103	def find_modrefs(local: bool \| None = None) -> tuple[ModRef, ...]:	8✔
104	# determine directories with python files
105	dirpaths: set[Path] = set()	8✔
106	modrefs: list[ModRef] = []	8✔
107	for syspathstr in sys.path:	8✔
108	syspath = Path(syspathstr).resolve()	8✔
109	if local is not None and local is any(syspath.is_relative_to(pkg_path) for pkg_path in PKG_PATHS):	8✔
110	continue	8✔
111	for filepath in syspath.glob("/.py"):	8✔
112	dirpath = filepath.parent	8✔
113	if dirpath.name.startswith("_") or dirpath.name == "ucdp":	8✔
114	continue	8✔
115	dirpaths.add(dirpath)	8✔
116
117	_find_modrefs_cached = CACHE.loader_cache.anycache(depfilefunc=_find_modrefs_files)(_find_modrefs)	8✔
118
119	maxworkers = get_maxworkers()	8✔
120	with ThreadPoolExecutor(max_workers=maxworkers) as exe:	8✔
121	# start
122	jobs = [	8✔
123	exe.submit(_find_modrefs_cached, sys.path, tuple(sorted(dirpath.glob("*.py"))))
124	for dirpath in sorted(dirpaths)
125	]
126	# collect
127	for job in jobs:	8✔
128	modrefs.extend(job.result())	8✔
129	return tuple(modrefs)	8✔
130
131
132	def _find_modrefs_files(modrefs: tuple[ModRef, ...], envpath: list[str], filepaths: tuple[Path, ...]) -> list[Path]:	8✔
133	paths: set[Path] = set(filepaths)	8✔
134	for modref in modrefs:	8✔
135	modcls = load_modcls(modref)	8✔
136	for basecls in get_modbaseclss(modcls):	8✔
137	paths.add(Path(getfile(basecls)))	8✔
138	return sorted(paths)	8✔
139
140
141	def _find_modrefs(envpath: list[str], filepaths: tuple[Path, ...]) -> tuple[ModRef, ...]: # noqa: C901	8✔
142	modrefs = []	8✔
143	for filepath in filepaths:	8✔
144	pylibname = filepath.parent.name	8✔
145	pymodname = filepath.stem	8✔
146	if pymodname.startswith("_"):	8✔
147	continue	8✔
148	# skip non-ucdp files
149	with filepath.open(encoding="utf-8") as file:	8✔
150	try:	8✔
151	for line in file:	8✔
152	if _RE_IMPORT_UCDP.match(line):	8✔
153	break	8✔
154	else:
155	continue	8✔
156	except Exception as exc:	2✔
157	LOGGER.info(f"Skipping {str(filepath)!r} ({exc})")	2✔
158	continue	2✔
159
160	# import module
161	try:	8✔
162	pymod = import_module(f"{pylibname}.{pymodname}")	8✔
163	except Exception as exc:	8✔
164	LOGGER.warning(f"Skipping {str(filepath)!r} ({exc})")	8✔
165	continue	8✔
166
167	# Inspect Module
168	for name in dir(pymod):	8✔
169	# Load Class
170	modcls = getattr(pymod, name)	8✔
171	if not isclass(modcls) or not issubclass(modcls, BaseMod):	8✔
172	continue	8✔
173
174	# Ignore imported
175	if filepath != Path(getfile(modcls)):	8✔
176	continue	8✔
177
178	# Create ModRefInfo
179	modclsname = get_modclsname(pymodname)	8✔
180	if modclsname == name:	8✔
181	modref = ModRef(libname=pylibname, modname=pymodname)	8✔
182	else:
183	modref = ModRef(libname=pylibname, modname=pymodname, modclsname=name)	8✔
184	modrefs.append(modref)	8✔
185
186	return tuple(modrefs)	8✔
187
188
189	def get_topmodrefpats(patterns: Patterns \| None) -> Iterator[TopModRefPat \| TopModRef]:	8✔
190	for pattern in patterns or []:	8✔
191	path = guess_path(pattern)	8✔
192	if path:	8✔
UNCOV 193	yield TopModRef.cast(path)	×
194	else:
195	mat = _RE_TOPMODREFPAT.fullmatch(pattern)	8✔
196	if mat:	8✔
197	yield TopModRefPat(**mat.groupdict())	8✔
198	else:
199	yield TopModRefPat(top=".") # never matching	8✔

nbiotcloud / ucdp / 14203589112

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous